Polars Kills Duplicate Subplans by Fingerprinting Them

TL;DR: The fingerprint of a plan node in Polars is the hash of the node itself plus the canonical ids of its children. The canonical id is obtained by interning that hash in a hashbrown::HashTable whose equality closure compares the children’s ids and the nodes “shallowly”. Two subplans are duplicates exactly when their roots get the same id, which makes the duplicate search a hash-table lookup instead of a subtree comparison. The rewrite replaces the duplicate with an IR::Cache node, and a later pass decides — with row estimates — whether sharing the subplan is actually cheaper than computing it twice.

The brief for this article mentioned a WithCluster rewrite and a Hash impl on IR nodes. Neither exists in that form at this commit. What exists is CanonicalIRMap in crates/polars-plan/src/plans/ir/canonical.rs and CanonicalExprMap in crates/polars-plan/src/plans/aexpr/canonical.rs, and the rewrite they enable is the insertion of IR::Cache nodes. The cluster_with_columns.rs, in the same optimizer directory, is a different optimisation entirely — it merges stacked with_columns stages by name — and it does not fingerprint anything. The title survives the pivot; the mechanism underneath it is better than the one I was promised.

The problem the pass solves

A lazy query like “read this Parquet file, join it to that one, aggregate” often mentions the same scan twice — once on each side of a join, or once per branch of a union. If the engine computes that subplan twice, it reads the file twice, decodes it twice and does the work twice. If it computes it once and shares the result, it holds the intermediate in memory.

Neither is always right. The decision depends on how expensive the subplan is and how expensive it is to materialise, which is why the pass is structured in three stages: detect, rewrite, then evaluate the trade-off.

Detection is gated on a cheap pre-check. Before the expensive pass runs, a member collector walks the plan and counts the scans:

rust
// Tracks canonical source identities to check if we have duplicate sources.
#[cfg(feature = "cse")]
#[derive(Default)]
struct UniqueScans {
    ids: PlIndexSet<CanonicalIRId>,
    count: usize,
}

#[cfg(feature = "cse")]
impl UniqueScans {
    fn insert(
        &mut self,
        node: Node,
        lp_arena: &Arena<IR>,
        expr_arena: &Arena<AExpr>,
        canonical_ir_map: &mut CanonicalIRMap,
    ) {
        self.ids
            .insert(canonical_ir_map.resolve(node, lp_arena, expr_arena));
        self.count += 1;
    }
}

crates/polars-plan/src/plans/optimizer/collect_members.rs:24

The collector puts every scan through the canonical map:

rust
                #[cfg(feature = "cse")]
                Scan { .. } => {
                    self.scans
                        .insert(_node, lp_arena, _expr_arena, &mut canonical_ir_map);
                },

crates/polars-plan/src/plans/optimizer/collect_members.rs:95

and the “are there duplicates?” question is answered by comparing the number of scans against the number of distinct scan ids:

rust
    #[cfg(feature = "cse")]
    pub(super) fn has_duplicate_scans(&self) -> bool {
        self.scans.count != self.scans.ids.len()
    }

crates/polars-plan/src/plans/optimizer/collect_members.rs:124

Note that this pre-check already uses the fingerprinting machinery. The collector interns every scan into a CanonicalIRMap and counts how many distinct ids there are. If there are as many ids as scans, nothing is duplicated and the whole common-subplan pass is skipped. This is the first place the “fingerprint” earns its keep: the duplicate check is a hash lookup, not a plan comparison.

The pass itself is then run conditionally, and only when there is a reason to believe it will find something:

rust
    if comm_subplan_elim {
        feature_gated!("cse", {
            let members = get_or_init_members!();
            if ((members.has_sink_multiple || members.has_joins || members.has_unions)
                && members.has_duplicate_scans())
                || members.has_cse_equivalent_resolvers()
            {
                if verbose {
                    eprintln!("found multiple sources; run comm_subplan_elim")
                }

                run_set_cache_states = cse::cspe::common_subplan_elimination(
                    root,
                    ir_arena,
                    expr_arena,
                    polars_config::config().allow_nested_cspe(),
                    !opt_flags.streaming(),
                );
            }
        });
    };

crates/polars-plan/src/plans/optimizer/mod.rs:165

The fingerprint

Here is the whole idea in one function. resolve_single is called for a node whose children already have canonical ids; it hashes the node and the child ids together, looks the hash up in a hash table, and returns the id of the equivalence class it belongs to:

rust
    fn resolve_single(
        &mut self,
        node: Node,
        child_ids: Vec<CanonicalIRId>,
        lp_arena: &Arena<IR>,
        expr_arena: &Arena<AExpr>,
    ) -> CanonicalIRId {
        let exprs_are_nondeterministic = lp_arena.get(node).exprs().any(|expr| {
            let expr_id = self.expr_map.resolve(expr.node(), expr_arena);
            self.expr_map.is_nondeterministic_excluding_udfs(expr_id)
        });

        let Self {
            deduplication_map,
            cache,
            eq_classes,
            expr_map,
        } = self;

        let expr_cmp = CanonicalExprMapWithArena::new(expr_map, expr_arena);
        let hash = combined_hash(node, &child_ids, lp_arena, &expr_cmp);
        let entry = deduplication_map.entry(
            hash,
            |&other| {
                let other = &eq_classes[other.index()];
                child_ids == other.child_ids
                    && lp_arena
                        .get(node)
                        .is_ir_equal_shallow(lp_arena.get(other.representative), &expr_cmp)
            },
            |&other| {
                let other = &eq_classes[other.index()];
                combined_hash(other.representative, &other.child_ids, lp_arena, &expr_cmp)
            },
        );

crates/polars-plan/src/plans/ir/canonical.rs:109

and the hash itself:

rust
/// Hashes the node itself, without descending into its inputs, combined with the
/// ids of those inputs.
fn combined_hash(
    node: Node,
    child_ids: &[CanonicalIRId],
    lp_arena: &Arena<IR>,
    expr_cmp: &CanonicalExprMapWithArena,
) -> u64 {
    let mut hasher = DefaultHasher::new();
    lp_arena
        .get(node)
        .hash_excluding_inputs(&mut hasher, expr_cmp);
    for child_id in child_ids {
        child_id.hash(&mut hasher);
    }
    hasher.finish()
}

crates/polars-plan/src/plans/ir/canonical.rs:214

This is a Merkle-style structural hash, and it is the reason the whole thing is cheap. The hash of a node depends only on (a) the node’s own contents and (b) the ids of its children — and those ids are already canonical, so two structurally identical subtrees produce the same child ids, hence the same parent hash, hence the same id, all the way up. The recursion is never done twice for the same subtree: the ids are memoised per node in a cache.

text

   subplan A                          subplan B
   ---------                          ---------
   Filter(a > 3)   hash -> h1         Filter(a > 3)   hash -> h1
        |                                     |
   child id 7                              child id 7
        |                                     |
   Scan(files)  hash -> h2               Scan(files)  hash -> h2
        |                                     |
   child id 11                             child id 11
        |                                     |
   Select(*)  hash -> h3                 Select(*)  hash -> h3
        |                                     |
   id = 42  <--------------------------->  id = 42     (same class)

Note the direction of the arrows: the id of a node is a function of its children’s ids, so the ids have to be computed bottom-up. That is exactly what the post-order traversal in resolve does.

Note the two-level structure: hash is a bucket selector, and the equality closure is the real test. Two nodes with the same hash are only the same if their child ids match and their shallow contents match. That is the standard hash-table contract, and it is why the equality closure has to be exactly as strong as the hash — if hash_excluding_inputs hashed something the equality check ignores, two different plans could land in the same class.

The Hash impl

The brief asked about a Hash impl for IR nodes. There is not one, and there cannot be a simple one: an IR node contains child plan nodes (Node indices into an arena) and child expression nodes. Hashing the arena indices would make the hash depend on the arena layout rather than on the plan, which is exactly what the canonical id is supposed to avoid.

So IR gets a hand-written method instead, and the method is explicit about what it does and does not cover:

rust
impl IR {
    /// Hash the contents of the enum, without descending into child IR nodes.
    /// The user can choose how to hash the referenced ExprIR nodes by providing `expr_hash`
    pub(crate) fn hash_excluding_inputs<H: Hasher>(
        &self,
        state: &mut H,
        expr_hash: &impl ExpressionHasher,
    ) {
        let hash_exprs = |exprs: &[ExprIR], state: &mut H| {
            for e in exprs {
                expr_hash.hash_expr(e, state);
            }
        };

        std::mem::discriminant(self).hash(state);

crates/polars-plan/src/plans/ir/hash.rs:21

and the arms themselves show the pattern — the node’s own payload is hashed, the child plan node is not:

rust
            IR::Slice {
                offset,
                len,
                input: _,
            } => {
                len.hash(state);
                offset.hash(state);
            },
            IR::Filter {
                input: _,
                predicate,
            } => {
                expr_hash.hash_expr(predicate, state);
            },

crates/polars-plan/src/plans/ir/hash.rs:81

The input: _ in every arm is the point: the children are not hashed here, they are contributed by the caller as canonical ids. The expressions that hang off the node are hashed, but through the pluggable ExpressionHasher, because an expression is hashed by its canonical id rather than by its contents:

rust
#[cfg(feature = "cse")]
impl<'a> ExpressionHasher for CanonicalExprMapWithArena<'a> {
    fn hash_expr<H: Hasher>(&self, expr: &ExprIR, state: &mut H) {
        self.map
            .borrow_mut()
            .resolve(expr.node(), self.arena)
            .hash(state);
        expr.output_name_inner().hash(state);
    }
}

crates/polars-plan/src/plans/aexpr/canonical.rs:231

The expression side has its own Hash impl, and it is a shallow one — the variant plus the attributes that matter for equality, never the children:

rust
impl Hash for AExpr {
    // This hashes the variant, not the whole expression
    // Make sure that all attributes that are important for equality are hashed. Nodes don't have
    // to be hashed.
    fn hash<H: Hasher>(&self, state: &mut H) {
        std::mem::discriminant(self).hash(state);

        match self {
            AExpr::Column(name) => name.hash(state),
            #[cfg(feature = "dtype-struct")]
            AExpr::StructField(name) => name.hash(state),
            AExpr::Literal(lv) => lv.hash(state),
            AExpr::Function {
                options,
                function,
                input: _,
            } => {
                options.hash(state);
                function.hash(state)
            },

crates/polars-plan/src/plans/aexpr/hash.rs:15

So there are two nested Merkle constructions: an expression’s id is the hash of its own contents plus its children’s ids, and a plan node’s id is the hash of its own contents plus its children’s ids. CanonicalExprMap and CanonicalIRMap are the same algorithm at two levels, and CanonicalIRMap literally embeds one inside the other:

rust
/// Assigns [`CanonicalIRId`]s to `IR` nodes.
pub struct CanonicalIRMap {
    deduplication_map: HashTable<CanonicalIRId>,
    cache: PlIndexMap<Node, CanonicalIRId>,
    /// Indexed by [`CanonicalIRId`], which are handed out densely.
    eq_classes: Vec<CanonicalIRClass>,
    expr_map: CanonicalExprMap,
}

crates/polars-plan/src/plans/ir/canonical.rs:33

Finding the duplicates and rewriting them

With ids assigned, the duplicate search is trivial: a hash map from canonical id to “how many times have I seen this?”. The first traversal assigns the ids and counts the hits:

rust
    fn post_visit(
        &mut self,
        key: Self::Key,
        storage: &mut Self::Storage,
        edges: &mut dyn NodeEdgesProvider<Self::Edge>,
    ) -> ControlFlow<Self::BreakValue> {
        let id = self.canonical_ir_map.resolve(key, storage, self.expr_arena);

        use indexmap::map::Entry;

        let entry_idx = match self.id_map.entry(id) {
            Entry::Occupied(mut e) => {
                e.get_mut().hits += 1;
                e.index()
            },
            Entry::Vacant(e) => {
                let idx = e.index();

                e.insert(IDState::new(
                    self.canonical_ir_map.is_nondeterministic_excluding_udfs(id),
                ));

                idx
            },
        };

crates/polars-plan/src/plans/optimizer/cse/cspe.rs:128

The second traversal decides which node to cache. The condition is subtle and worth reading closely:

rust
        // Cache the topmost deterministic node
        let should_cache = !curr_state.is_nondeterministic_excluding_udfs
            && if output_state.is_nondeterministic_excluding_udfs {
                curr_state.hits > 1
            } else {
                curr_state.hits > output_state.hits
            }
            && (self.cache_plain_scans || !is_plain_scan(storage.get(key)));

crates/polars-plan/src/plans/optimizer/cse/cspe.rs:213

Three conditions:

The node must be deterministic. Caching now() and reading it twice would give two different answers, so a non-deterministic subtree is never shared.
The node must be hit more often than its parent. If the parent is already shared, caching the child again buys nothing; the rule caches the topmost node that is duplicated, so one cache covers the whole duplicate subtree.
A plain scan is only cached if cache_plain_scans is set. Re-reading a file is often cheaper than materialising it — the file is on disk, the cache is in memory.

When the condition holds, the node is wrapped in an IR::Cache:

rust
        if should_cache {
            let replacement_ir = match storage.get(key) {
                ir @ IR::Cache { .. } => ir.clone(),
                _ => {
                    let ir = storage.take(key);
                    let new_key = storage.add(ir);

                    IR::Cache {
                        input: new_key,
                        id: UniqueId::new(),
                    }
                },
            };

            curr_state.replacement_ir = Some(replacement_ir);

crates/polars-plan/src/plans/optimizer/cse/cspe.rs:235

The id is a fresh UniqueId per cache node, and it is the thing that ties the duplicates together at execution time. Lowering the plan, the second occurrence of the cache id resolves to the same physical node:

rust
        IR::Cache { input, id } => {
            let id = *id;
            if let Some(cached) = cache_nodes.get(&id) {
                return Ok(*cached);
            }

            let phys_input = lower_ir!(*input)?;
            cache_nodes.insert(id, phys_input);
            return Ok(phys_input);
        },

crates/polars-stream/src/physical_plan/lower_ir.rs:1215

So the sharing is not a copy of the data and not a “the second branch reads from the first branch’s output” hack. It is a single physical node with two consumers.

The cost of the pass

The pass is not free, and Polars is explicit about the trade-off.

Planning cost. Two full traversals of the plan, each of which resolves canonical ids, which means hashing every node and every expression on every path. The resolution is memoised per node, so a subtree that is shared is hashed once for the id, but the first traversal still visits every node of the tree.

Memory cost. A cache node materialises the subtree’s output. That is the whole point — the subplan is computed once — but it means the shared data is held in memory for as long as any consumer needs it. For a subplan that produces a lot of rows, sharing it can cost more than recomputing it.

Correctness cost. Sharing a subplan is only sound if the subplan is deterministic and if the consumers want the same thing from it. If one branch filters and the other does not, the filter has to be pushed below the cache, or the cache has to be widened. Polars handles this in set_cache_states, which runs after predicate and projection pushdown and re-runs pushdown on the shared subplan.

And when the estimates are available, the pass makes the trade-off explicit and quantitative:

rust
                // Keeping the caches costs one evaluation of the subplan, plus a pass over the
                // materialized rows for every reference that reads and filters them.
                let keep_cost = if remove_caches && removal_cost.is_some() {
                    // Caches block pushdown, so their children may still contain cross joins.
                    let input = narrowed_shared_subplan(
                        &v.children,
                        &v.parents,
                        pushdown_maintain_errors,
                        &mut pred_pd,
                        lp_arena,
                        expr_arena,
                    )?;
                    shared_input = Some(input);
                    subplan_cost(input, lp_arena, expr_arena)
                        .map(|c| c.work + v.cache_nodes.len() as f64 * c.rows)
                } else {
                    None
                };

                if let Some((removal_cost, keep_cost)) = removal_cost.zip(keep_cost) {
                    remove_caches = removal_cost <= keep_cost;
                    if verbose {
                        eprintln!(
                            "cache removal estimated at {removal_cost:.3e} rows against \
                             {keep_cost:.3e} for sharing the subplan"
                        )
                    }
                }

crates/polars-plan/src/plans/optimizer/cse/cache_states.rs:417

removal_cost is the estimated work of running the subplan once per consumer; keep_cost is the estimated work of computing it once plus a pass over the materialised rows for every consumer. Whichever is smaller wins, and the caches are removed if sharing is not worth it. Without row estimates the pass falls back on the pushdown probe alone, which is conservative in the direction of keeping the cache.

The expression-level sibling

The same fingerprinting machinery is used one level down, for common subexpressions rather than common subplans, in crates/polars-plan/src/plans/optimizer/cse/csee.rs. The canonical map is the same CanonicalExprMap, and the counting pass is the same shape:

rust
        if subtree_is_valid
            && role == NodeRole::Candidate
            && (!self.is_group_by || self.has_frame_height(node.node(), arena))
        {
            let id = self.canonical_map.resolve(node.node(), arena);
            *self.se_count.entry(id).or_insert(0) += 1;
        }

crates/polars-plan/src/plans/optimizer/cse/csee.rs:248

and the rewrite replaces the repeated expression with a reference to a temporary column:

rust
    fn mutate(
        &mut self,
        mut node: Self::Node,
        arena: &mut Self::Arena,
    ) -> PolarsResult<Self::Node> {
        let (id, count) = self
            .candidate(node.node())
            .expect("mutated node was not a CSE candidate");
        debug_assert!(count > 1);

        node.assign(AExpr::col(cse_column_name(id)), arena);
        self.rewritten = true;

        Ok(node)
    }

crates/polars-plan/src/plans/optimizer/cse/csee.rs:363

The temporary column name is derived from the canonical id, so it is stable and unique across the plan:

rust
fn cse_column_name(id: CanonicalExprId) -> PlSmallStr {
    format_pl_smallstr!("{}{:#x}", CSE_REPLACED, id.as_u64())
}

crates/polars-plan/src/plans/optimizer/cse/csee.rs:65

The NodeRole classification is where the safety lives. A literal series that is too long to hash cheaply, a nested dtype, a rolling window, a nondeterministic expression — none of them are candidates, and a node with a non-candidate anywhere below it invalidates its whole subtree:

rust
/// Classifies how a visited node may participate in CSE.
fn classify(ae: &AExpr, is_group_by: bool) -> NodeRole {
    match ae {
        // Don't allow this for now, as we can get `null().cast()` in ternary expressions.
        // TODO! Add a typed null
        AExpr::Literal(LiteralValue::Scalar(sc)) if sc.is_null() => NodeRole::Refuse,
        AExpr::Literal(LiteralValue::Series(s)) => {
            let dtype = s.dtype();

            // Object and nested types are harder to hash and compare.
            let allow = !(dtype.is_nested() | dtype.is_object());

            if s.len() < CHEAP_SERIES_HASH_LIMIT && allow {
                NodeRole::Member
            } else {
                NodeRole::Refuse
            }
        },

crates/polars-plan/src/plans/optimizer/cse/csee.rs:162

The bottom line

What the fingerprint buys, concretely:

Duplicate detection is a hash lookup. Comparing two subplans structurally is O(size of the subtree) and is repeated for every pair. Interning them is O(size of the plan) once, and after that everything is integer comparison.
The detection and the rewrite share the same machinery. The same CanonicalIRMap first answers “are there duplicates?” (in the member collector, before the expensive pass is even scheduled) and then answers “which node do I cache?”.
The equality check is still there. The hash only selects a bucket; is_ir_equal_shallow plus the child-id comparison decide membership. A hash collision costs a comparison, never a wrong answer.
Determinism is a first-class property of the fingerprint. The canonical id carries is_nondeterministic_excluding_udfs, computed bottom-up, and a non-deterministic node is never shared.

The costs, stated plainly:

Two traversals and a hash of every node on every query where the cse feature is enabled and the plan has a shape that could contain duplicates. The pass is skipped entirely unless the plan has a sink with multiple inputs, a join or a union and the member collector found duplicate scans.
Materialising the shared subplan. Sharing trades compute for memory, and the memory is held until the last consumer is done with it. The pass models this with row estimates and removes the cache when the arithmetic says sharing is not worth it.
A second, similar mechanism to understand. There are two canonical maps, one for expressions and one for plan nodes, and the expression one is embedded in the plan one. They share the algorithm, not the code.

The practical takeaway: when you see IR::Cache in an optimised Polars plan, the reason it is there is that two parts of the plan fingerprinted identically, and the reason the plan does not have one is that either the fingerprints differed or the arithmetic said that recomputing was cheaper.