Repository navigation
Enable TopK dynamic filter pushdown into native Parquet scans #5775
Description
What is the problem the feature request solves?
DataFusion's TopK dynamic filter can exclude input that cannot beat the current top K rows. For queries such as SELECT x FROM t ORDER BY x LIMIT 16, this can avoid reading later Parquet row groups after a useful threshold is established.
Comet already constructs DataFusion SortExec with a fetch for TopK, but its current Spark execution path prevents reader pushdown: CometTakeOrderedAndProjectExec runs its child separately and passes Arrow batches into a new native TopKInput plan. The Parquet reader is outside that plan. Connecting a filter only below the existing TopK cannot recover scan I/O already performed.
Describe the potential solution
First make the per-partition TopK and an eligible native scan part of the same native execution, while preserving Spark's local TopK, shuffle, final TopK, projection, and offset behavior. Then connect DataFusion's live TopK predicate to that reader using the attachment introduced in #5699. Start with one direct signed-integer sort key and native Parquet input.
- Preserve the local candidate count and final global result across one and multiple Spark partitions, including LIMIT/OFFSET and already ordered inputs. Keep threshold state local to the native execution; do not push it across Spark exchanges or JVM/Arrow boundaries.
- Let DataFusion update the threshold as the heap fills and improves. Use fresh producer/reader state for each execution and preserve Spark metric ownership.
- Check ascending/descending order, ties, null placement, empty input, fewer than K rows, offsets, and repeated execution. Keep unsupported expressions/types conservative until their Spark semantics are validated.
- Demonstrate the actual Spark native plan contains both TopK and its Parquet reader, and that the reader receives updates. Compare filtering off/on for favorable, random, and unfavorable layouts and different K values, measuring row groups/rows pruned, bytes scanned, and elapsed query time.
Additional context
Source verification against Comet 4c14448c2799e519d2648357db4442132beb4651 and DataFusion 55.0.0:
- CometTakeOrderedAndProjectExec calls
child.executeColumnar()and creates separate native executions around the shuffle. getTopKNativePlan createsScan(source = TopKInput)plusSort(fetch = limit). - The native planner constructs
SortExec.with_fetchbut does not attach its filter. A normal Comet sort does not set fetch, so it is not an alternative reader-connected TopK path today. - DataFusion creates the dynamic filter in with_fetch and offers it for pushdown. Comet needs to connect producer and consumer explicitly.
Follow-up to #5699. The scan benefit requires the execution-boundary work above; enabling a DataFusion configuration flag alone is insufficient.
A scratch native test on this Comet revision manually built SortExec(fetch=16) over Comet's Parquet reader and attached its dynamic predicate through the existing reader helper. With DataFusion 55.0.0, an ascending INT32 key, 40 row groups of 1,024 rows, row filtering enabled, and page-index pruning disabled, ascending TopK returned identical results with filtering off/on. Reported bytes_scanned fell from 164,640 to 4,116 and scan output rows from 40,960 to 1,024 (97.5% reductions). This demonstrates the reader benefit when producer and reader share a native plan; the current Spark TopKInput boundary was deliberately absent in the probe. It is not evidence that current Spark TopK queries already benefit, nor an elapsed-time benchmark. One scratch test covering MIN, MAX, and TopK passed.
Activity
peterxcli commentedon Sep 8, 2026
take
peterxcli commentedon Sep 8, 2026
cc @pingzh feel free to take. plan to work on this in few days.
pingzh commentedon Sep 8, 2026
take
peterxcli commentedon Sep 11, 2026
untake
pingzh commentedon Sep 15, 2026
Implementation plan: four smaller PRs
Following the implementation and review in #5785, split the work into four focused PRs. Each PR will include its own tests and documentation, with benchmarks alongside the behavior they measure.
dynamic_filter_join.rs, extract the shared Parquet attachment helper, and standardizedynamic_filter_join_*metrics. Preserve existing filtering behavior.Dependencies
This separates review of the refactoring, correctness fixes, fusion, and reader filtering. The correctness safeguards and lifecycle coverage must be in place before enabling TopK reader filtering.
pingzh commentedon Sep 30, 2026
Completion and benchmark results
TL;DR: On a fresh release build of merged
main, ascending layouts reduced scan data bytes by 46–82%. For one partition, 16 BIGINT payload columns, and K=16, pruning reduced mean query time from 214.5 ms to 76 ms versus fused TopK (64.6% lower), and from 230.5 ms to 76 ms versus unfused TopK (67.0% lower). Descending layouts pruned nothing; the wide K=16 cases remained about 31% slower than unfused execution because the fusion cost remains. Both options stay disabled by default.The updated four-PR plan is complete: #5937, #6067, #6181, and #6263 are merged. The former optional PR5 is no longer part of this issue's scope; #6263's older description still refers to that earlier plan.
Setup
9c7fcc5aa4a343f53b4bfc8707eeb8dfd17a2c48, freshly fetched Apachemainon September 30, 2026; includes the merged reader-pruning implementation.target-cpu=native, 8 GiB JVM heap.SELECT * FROM topk_input ORDER BY k ASC LIMIT K. These layout names do not describe different SQL sort directions.local[1], AQE off, one Parquet file per scan partition, 5–6 verified row groups per file, Snappy, dictionaries disabled, batch size 4096. Four partitions exercise separate task thresholds, not parallel scaling.unfused,fused,pruning, thenpruning,fused,unfused. Each case has at least 500 ms warmup, at least 500 ms measurement, and at least five measured iterations: 24 scenarios × 3 modes × 2 runs = 144 mode cases.unfuseddisables both options;fusedenables onlyspark.comet.exec.topK.fusion.enabled;pruningenables fusion plusspark.comet.exec.topK.dynamicFilter.enabled.Times below average the two runs' reported mean milliseconds and include query planning and collection. Fixture generation, result comparisons, and plan/metric checks are outside timings. These are warmed local microbenchmarks; small timing differences should not be treated as established regressions or speedups.
Representative results
One partition, 16 BIGINT payload columns, K=16:
Across the full matrix:
All 24 scenarios: mean query times and data-byte savings
P = scan partitions; C = BIGINT payload columns. Negative time change means faster. Both time comparisons are shown: versus fused isolates reader pruning; versus unfused includes fusion's cost.
All 24 scenarios: exact reader counters
Both JVM runs produced identical reader counters. Both baselines emitted all 1,048,576 rows, read the same data bytes, and skipped zero row groups. The table shows pruning mode's emitted rows and skipped groups.
Counters come from separate untimed validation executions.
bytes_scanned(requested ranges) equaledscan_io_data_bytes(returned data-page bytes) in every case; neither measures physical disk/network traffic, and both exclude footer/page-index reads in these fixtures.scan_io_metadata_byteswas 524,288 bytes per file in every mode (524,288 or 2,097,152 bytes per query). There are no Bloom-filter reads in these fixtures.row_groups_pruned_statistics,page_index_rows_pruned, andpushdown_rows_prunedwere zero throughout. This fixture has one file per task and demonstrates within-file pruning throughrow_groups_pruned_dynamic_filter; in other workloads, a threshold available when opening a later file can instead incrementrow_groups_pruned_statistics.Validation and completed scope
BUILD SUCCESS. Maven'sexec:javarunner emitted a Hadoop shutdown-hook classloader warning after completion; there were no benchmark execution failures.This completes the reader-pruning scope of the revised four-PR plan. Broader eligibility, fusion performance improvements, and additional TopK instrumentation can be separate follow-ups.
Reproduce
Use a full JDK 21 with its
bindirectory onPATH, Rust, and the pinned revision above:These runs used the equivalent Maven benchmark invocation from the Makefile after the full release reactor build, with an isolated Maven cache and the freshly built JNI library selected explicitly.
pingzh commentedon Sep 30, 2026
@sunchao @andygrove i think this issue can be closed 🙏
sunchao commentedon Oct 1, 2026
Closed, thanks @pingzh !