pingzh opened a new pull request, #5785: URL: https://github.com/apache/datafusion-comet/pull/5785
## Which issue does this PR close? Closes #5775. ## Rationale for this change The existing `TopKInput` boundary reads Parquet before executing native TopK. Sharing the scan and per-partition TopK in one native execution lets DataFusion's improving heap threshold prune later row groups. This builds on the Parquet reader attachment introduced in #5699. ## What changes are included in this PR? Three reviewable commits, each independently reviewed: 1. Fuse eligible local TopK with its native Parquet scan, preserving local candidate counts, shuffle, final projection, and offsets. 2. Attach a fresh live TopK predicate to the reader on each execution. Reuse the shared reader attachment helper, preserve metric ownership, and add native correctness/lifecycle tests. 3. Add Spark pruning/reexecution coverage, an off/on benchmark across three layouts and three K values, and configuration/metrics documentation. `spark.comet.exec.topK.dynamicFilter.enabled` defaults to `false`. Initial support is one direct signed integer sort key over native Parquet. Reader attachment skips projected top-level timestamps to preserve conversion overflow errors. Thresholds stay inside each task's native execution. ## How are these changes tested? - 28 native dynamic-filter tests passed, including existing join tests, 80 Parquet ordering/null/input-shape comparisons, changing thresholds, remapping, reset/reexecution, and stream cleanup. - 17 Spark TopK tests passed on Spark 4.0.4/JDK 17, including one/multiple partitions, AQE, offsets, ordered/empty inputs, four page-index/row-filter combinations, and timestamp overflow. The four reader/reexecution tests passed again after the review correction. - Cargo formatting, Scalafix, Spotless, Scalastyle, Clippy (`--all-targets --workspace -- -D warnings`), native builds, JVM compile/package/install, and jar-content checks passed. - All nine optimized benchmark preflights matched Spark and verified serialized native TopK → Parquet scan plans in both arms. Favorable layouts scanned about 99.5% fewer bytes and improved mean query time by 2.1–2.3×. Random K=256 regressed from 61 ms to 121 ms, and unfavorable layouts were 22–54% slower without I/O savings. The feature remains opt-in. See the [benchmark report](https://github.com/pingzh/datafusion-comet/blob/pingzh-topk-reader-filters/benchmarks/results/topk-reader-filters.md) for every result, the optimized CI profile used, configuration, and reproduction commands. -- 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]
