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]
