schenksj opened a new pull request, #4810:
URL: https://github.com/apache/datafusion-comet/pull/4810
## Which issue does this PR close?
Closes #4807.
Follow-up issues for IO-level pushdown into kernel scans: #4808 (Iceberg),
#4809 (Delta).
## Rationale for this change
DataFusion 54 ships runtime dynamic filtering for hash joins (`HashJoinExec`
+ `DynamicFilterPhysicalExpr` + `SharedBuildAccumulator`), but the
consumer-side wiring lives in DataFusion's physical optimizer, which Comet does
not run — so Comet never benefits. Build-side runtime filtering is one of the
largest TPC-DS levers in comparable engines (Velox
`HashProbe::pushdownDynamicFilters`; Databricks DFP). This PR wires it up on
Comet's side, scan-agnostically, so all probe-side scans (Parquet
`DataSourceExec`, JVM-fed `ScanExec`, Iceberg, and future Delta kernel scans)
inherit row-level filtering.
## What changes are included in this PR?
- New `CometDynamicFilterExec` operator
(`native/core/src/execution/operators/dynamic_filter.rs`): evaluates the join's
shared `DynamicFilterPhysicalExpr` against probe-side batches before the hash
probe, with
- a pass-through fast path while the filter is the constant-`true`
placeholder (correctness never depends on population),
- a selectivity guard that permanently disables evaluation on a partition
stream when it keeps > 95% of rows after 64K rows observed,
- metrics: `dynamic_filter_rows_pruned` + baseline
elapsed-compute/output-rows.
- `attach_join_dynamic_filter` helper: rewires an eligible `HashJoinExec`
(including the `ProjectionExec`-over-join shape produced by `swap_inputs`) so
the join and the new probe-side wrapper share one filter, via the public
`HashJoinExecBuilder` + `with_dynamic_filter_expr` APIs.
- Eligibility gate mirrors DataFusion's own: probe side must be preserved
under the ON clause (`JoinType::on_lr_is_preserved().1` — Inner, LeftOuter,
LeftSemi/RightSemi, LeftAnti, LeftMark); null-aware anti joins are excluded.
- Planner integration in both HashJoin branches (build-left and swap), proto
field `HashJoin.dynamic_filter_enabled`, and a new config
`spark.comet.exec.join.dynamicFilter.enabled` (default `false`, experimental).
Timing note: each Comet native execution corresponds to a single Spark
partition, so the build accumulator completes after that task's build and the
filter is populated before probe batches flow. Broadcast joins get a
broadcast-wide filter; shuffled hash joins get a per-partition filter. If
DataFusion declines to populate (session flag off, etc.), the placeholder stays
`true` and the wrapper is a no-op.
## How are these changes tested?
- New Rust unit tests
(`native/core/src/execution/operators/dynamic_filter.rs`): pass-through
pre-population, filtering after `update()`, constant-false (empty build)
pruning, metric counting; end-to-end `HashJoinExec` execution asserting
identical results with and without the filter attached and
`dynamic_filter_rows_pruned > 0` (proving the build phase populates the
manually-attached filter); attach-helper gating for ineligible join types
(Right/Full/RightAnti). Full native suite passes (128 tests), clippy + rustfmt
clean.
- New Scala test (`CometJoinSuite`: "HashJoin with dynamic filter enabled"):
result parity vs. Spark with the config enabled across broadcast + shuffled
hash joins, inner/left-outer/semi/anti, a selective build side, an empty build
side, and NULL join keys. Verified under the spark-3.5 profile.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_019dAmX1gT713ChZjYtKKb7p
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
