Skip to content

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

Open

Description

@peterxcli

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. Comet uses DataFusion's min/max UDAFs.
  • DataFusion initializes dynamic filters for eligible ungrouped Partial MIN/MAX aggregates and updates the bound as batches are consumed: eligibility, bound updates.
  • The merged Parquet reader attachment 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.

Activity

  1. peterxcli commented on Sep 8, 2026

    @peterxcli
    MemberAuthor

    take

  2. peterxcli commented on Sep 8, 2026

    @peterxcli
    MemberAuthor

    cc @pingzh feel free to take. plan to work on this in few days.

  3. pingzh commented on Sep 8, 2026

    @pingzh
    Contributor

    take

  4. peterxcli commented on Sep 11, 2026

    @peterxcli
    MemberAuthor

    untake

  5. added
    enhancementNew feature or request
    area:scanParquet scan / data reading
    area:aggregationHash aggregates, aggregate expressions
    and removed on Sep 14, 2026
  6. pingzh commented on Nov 10, 2026

    @pingzh
    Contributor

    I plan to implement this in three small, dependent PRs:

    1. Native MIN/MAX reader filtering. Add an internal MinMaxReaderFilterExec that creates a fresh aggregate and live predicate for each execution and attaches that predicate to the existing native Parquet reader. Include attachment metrics and core native correctness, pruning, and execution-lifecycle tests. Spark planner wiring is deferred to PR 2.
    2. Spark integration, disabled by default. Add spark.comet.exec.aggregate.dynamicFilter.enabled, capture it in the serialized aggregate plan, wire the native wrapper into eligible aggregate planning, and expose attachment metrics in Spark. Check filtering off/on against Spark, including multiple Spark partitions and unsupported-plan fallback.
    3. Benchmark and documentation. Compare favorable and unfavorable layouts for MIN and MAX, recording scan rows, row groups pruned, bytes scanned, and elapsed query time. Document configuration and supported plans; keep the default disabled.

    The first version will support one unfiltered MIN or MAX over a direct signed-integer column, with no grouping, in raw-input Partial mode, reading an eligible native Parquet scan. Bounds remain local to each execution. Mixed aggregates, Final/PartialMerge stages, intervening expressions/filters/limits, and JVM/exchange boundaries remain ineligible. Reader-internal column mapping can use the existing attachment logic.

    The native lifecycle will follow the existing TopK reader-filter wrapper while retaining DataFusion's aggregate bound updates. Each execution must reconstruct the aggregate as well as its predicate, because cloning or replacing only the predicate can retain shared accumulator-bound state. Aggregate metrics remain with the aggregate and pruning/byte metrics remain with the scan.

    Core checks will cover empty/all-null input, duplicates and negative values, missing null-count statistics, repeated execution, and actual reader pruning. Later scope expansions can be separate PRs after this initial path is validated.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

area:aggregationHash aggregates, aggregate expressionsarea:scanParquet scan / data readingenhancementNew feature or request

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions