apache / apache/datafusion-comet
Support dynamic filter pushdown from ungrouped MIN/MAX into native Parquet scans
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
### What is the problem the feature request solves?
After #5699, Comet connects native hash-join dynamic filters to eligible Parquet readers, but does not yet connect DataFusion's aggregate dynamic filters.
An ungrouped `SELECT MIN(x) FROM t` or `SELECT MAX(x) FROM t` can benefit when an early batch establishes a bound that excludes later Parquet row groups. Each Spark task can prune its own remaining scan while preserving the existing partial/final aggregate. The benefit depends on data layout and when a useful bound becomes available.
### Describe the potential solution
Connect the live predicate produced by an eligible DataFusion `AggregateExec` to the native Parquet scan in the same Comet execution. Reuse the reader attachment from #5699 and DataFusion's bound updates. A conservative first scope is a single, unfiltered `MIN` or `MAX` over a direct signed-integer column, with no grouping, in raw-input Partial mode.
- Preserve Spark's partial/final aggregate buffers and task partitioning. Do not apply raw-input pruning to Final or PartialMerge stages, or share bounds across Spark exchanges.
- Stop at unsupported projections, expressions, filters, limits, and JVM/Arrow boundaries. Mixed aggregates such as `MIN(x), SUM(y)` must remain ineligible. Validate multiple MIN/MAX expressions, filtered aggregates, and additional types separately before enabling them.
- Create fresh predicate and accumulator state per execution, preserve metric ownership, and expose attachment/pruning metrics plus an enable/disable control.
- Add Spark/native result checks for empty/all-null input, duplicates, negative values, multiple Spark partitions, and repeated execution. Compare filtering off/on for favorable and unfavorable Parquet layouts, recording rows/row groups pruned, bytes scanned, and elapsed query time.
### Additional context
Source verification against Comet `4c14448c2799e519d2648357db4442132beb4651` and DataFusion 55.0.0:
- Comet maps Spark Partial to DataFusion Partial and directly constructs `AggregateExec`, with no aggregate reader-filter attachment: [planner](https://github.com/apache/datafusion-comet/blob/4c14448c2799e519d2648357db4442132beb4651/native/core/src/execution/planner.rs#L1343-L1465). Comet uses DataFusion's [min/max UDAFs](https://github.com/apache/datafusion-comet/blob/4c14448c2799e519d2648357db4442132beb4651/native/core/src/execution/planner.rs#L2833-L2859).
- DataFusion initializes dynamic filters for eligible ungrouped Partial MIN/MAX aggregates and updates the bound as batches are consumed: [eligibility](https://github.com/apache/datafusion/blob/55.0.0/datafusion/physical-plan/src/aggregates/mod.rs#L1734-L1788), [bound updates](https://github.com/apache/datafusion/blob/55.0.0/datafusion/physical-plan/src/aggregates/aggregate_stream.rs#L95-L205).
- The merged [Parquet reader attachment](https://github.com/apache/datafusion-comet/blob/4c14448c2799e519d2648357db4442132beb4651/native/core/src/execution/operators/dynamic_filter.rs#L246-L319) provides a starting point. The hash join's completed-build lifecycle should not be copied onto an aggregate that tightens its filter while consuming input.
A scratch native test on this Comet revision manually connected the existing aggregate producer to `try_attach_parquet_reader_filter`. With DataFusion 55.0.0, a projected INT32 key, 40 Parquet row groups of 1,024 rows, row filtering enabled, and page-index pruning disabled, separate MIN and MAX runs both returned identical partial aggregate results with filtering off/on. On favorable layouts (ascending input for MIN, descending for MAX), reported `bytes_scanned` fell from 164,640 to 4,116 and scan output rows from 40,960 to 1,024 (97.5% reductions). This verifies potential native reader savings with explicit wiring; it is not a Spark end-to-end test or an elapsed-time benchmark. One scratch test covering MIN, MAX, and TopK passed.
Contributor guide
Assessment
This issue has not been assessed yet.