peterxcli opened a new issue, #5774:
URL: https://github.com/apache/datafusion-comet/issues/5774

   ### 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.
   


-- 
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]

Reply via email to