apache / apache/datafusion-comet

Support dynamic filter pushdown from ungrouped MIN/MAX into native Parquet scans

Open
#5,774 4 comments 0 reactions 1 assignee Assigned to @pingzh View on GitHub
requires-triage
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.