Module semi_join_reduction 

Semi-join reduction through an aggregate.

When a grouped aggregate is inner-joined on one of its own grouping keys, only the groups whose key survives the join can appear in the result. Every other group is computed and then discarded. Filtering the aggregate’s input down to the surviving keys first produces exactly the same groups, because an aggregate value depends only on the rows sharing its key.

The query that motivated this

TPC-H q17 decorrelates to this shape:

Inner Join: part.p_partkey = __scalar_sq_1.l_partkey
  ├── Inner Join: lineitem.l_partkey = part.p_partkey
  │     └── Filter: p_brand = 'Brand#23' AND p_container = 'MED BOX'
  └── __scalar_sq_1:
        Aggregate: groupBy=[l_partkey], aggr=[avg(l_quantity)]
          TableScan: lineitem

At SF100 that aggregate groups all 600M lineitem rows into ~20M groups, and the join then keeps the ~2000 partkeys matching the brand and container — four orders of magnitude of thrown-away work. Measured with explain --analyze, it was 221.03 s of a 252 s query, 88% of all compute, with spill_count=0: not a memory problem, just work that need not happen.

DataFusion’s dynamic filter does not help here, and it is worth recording why, because the plan looks like it should:

DynamicFilter [ ... l_partkey >= 7682 AND l_partkey <= 19999654 AND hash_lookup ... ]

The min/max bounds span essentially the whole key domain, since the 2000 surviving partkeys are scattered uniformly across it, so row-group pruning removes nothing. And the filter belongs to the join, which sits downstream of the aggregate — no amount of selectivity there can reduce what the aggregate already had to read.

What this rule does

It rewrites the aggregate’s input to a LeftSemi join against the smallest subtree of the other side that still produces the join key and contains a filter:

Aggregate: groupBy=[l_partkey], aggr=[avg(l_quantity)]
  LeftSemi Join: lineitem.l_partkey = part.p_partkey
    TableScan: lineitem
    Projection: part.p_partkey
      Filter: p_brand = 'Brand#23' AND p_container = 'MED BOX'

Why it is safe

• Inner joins only. Under a left/right/full join the unmatched rows are preserved, so dropping groups would change the result. Anti/semi joins are also excluded — they have their own null semantics.
• The key must be a grouping column, matched by schema position rather than by name, so requalification through SubqueryAlias and projections cannot silently pair the wrong columns.
• Aggregate values are unchanged. Removing rows whose key is not in the probe side removes whole groups; it never removes part of a surviving group, so no aggregate is computed over a different row set.
• Nulls agree. A null key never satisfies an equi-join, so a null group would be dropped by the original join anyway; LeftSemi drops it too.
• No duplication. LeftSemi emits each left row at most once regardless of how many probe rows match, so counts and sums cannot inflate.

Why it is guarded

The probe subtree is evaluated a second time, so the rule only fires when that subtree contains a Filter — evidence there is real selectivity to exploit. Against an unfiltered scan the semi-join would remove nothing and we would have paid for the extra pass. Descent also stops at the filter rather than continuing to the scan beneath it, which is what keeps the probe small (~2000 rows in q17 rather than the whole part table).

Set KRISHIV_SEMI_JOIN_REDUCTION=off to disable.

Structs

SemiJoinPushdownThroughInnerJoin

Push an existing semi-join down through an inner join, so the selective side filters one join input instead of the join’s output.

SemiJoinReductionFromSelectiveDimension

Reduce a fact stream by a selective dimension it is inner-joined to, before the join that needs it.

SemiJoinReductionThroughAggregate

Push a semi-join built from an inner join’s other side into the input of a grouped aggregate, when the join key is one of the grouping columns.

Constants

SEMI_JOIN_DIMENSION_ENV

Environment switch for reduction from a selective dimension (the q7 rule).

SEMI_JOIN_PUSHDOWN_ENV

Environment switch for pushdown through an inner join (the q18 rule).

SEMI_JOIN_REDUCTION_ENV

Environment switch for reduction through an aggregate (the q17 rule).

Functions

semi_join_dimension_reduction_enabled

Whether reduction from a selective dimension is enabled (default: no).

semi_join_pushdown_enabled

Whether semi-join pushdown through an inner join is enabled (default: yes).

semi_join_reduction_enabled

Whether semi-join reduction through aggregates is enabled (default: yes).