pingzh opened a new pull request, #6181: URL: https://github.com/apache/datafusion-comet/pull/6181
## Which issue does this PR close? Part 3 of the [five-PR plan for #5775](https://github.com/apache/datafusion-comet/issues/5775#issuecomment-5672549320), following #5937 and #6067. #5775 remains open for TopK reader pushdown and the subsequent metrics work. ## Rationale for this change The existing local TopK consumes batches from a separately executed native scan through an Arrow input. Putting the scan and local selection in one native block removes that boundary and provides the plan structure needed for the following reader-pushdown PR. Fusion changes the execution pipeline and is workload-dependent. This PR makes it explicitly opt-in, with the default remaining disabled. ## What changes are included in this PR? - Add `spark.comet.exec.topK.fusion.enabled=false` and `CometLocalTopKExec` for a direct native Parquet scan with one signed integer sort key. Both directions and null orderings are supported. - Keep per-partition candidates for the physical limit, which already includes the offset. Apply the offset and output projection once in the final stage. - For one input partition, use a final native limit and projection without another TopK heap. For multiple partitions, shuffle local candidates and retain the final TopK. - Remove the inserted local node when reverting to Spark, preventing duplicated offsets/projections. Unsupported inputs retain their existing path. - Add focused tests, a configurable release benchmark, and user/contributor documentation. Reader thresholds, reader pruning, and residual-filter metrics are subsequent PRs. Three reviewable commits: implementation and initial coverage; boundary/reversion coverage; benchmark and documentation. Each commit received an independent subagent review, with findings addressed before publication. ## How are these changes tested? - Based on freshly fetched and pulled Apache `main` at `ccc780a64`, including the merged #6067 fix. - Spark 4.1.3 / JDK 21: all 23 targeted tests passed: 16 fusion cases, two transition-reversion cases, and five existing TakeOrderedAndProject cases. - Coverage includes AQE on/off, one/multiple partitions, all four signed integer key types, ASC/DESC and null orderings, ties, offsets, zero/oversized limits, empty input, repeated execution, projection of payload without the sort key, and unsupported-plan fallback. - Mutation verification: removing the local-node reversion safeguard makes both AQE cases fail. `LIMIT 5 OFFSET 2` returns `[4,5,6]` instead of `[2,3,4,5,6]`. Restoring it makes both tests pass again. - The Maven install also passed all 18 Parquet encryption integration tests. Confirmed that the packaged JNI library matches the current release build. - Scala compilation, Spotless, Scalastyle, Prettier, suite-registration checks, and `git diff --check` passed. The new suite is registered in both Linux and macOS CI. - Release benchmark verifies full results against Spark, actual scan partition counts, fused serialized plans, single-partition final selection, and identical decoded rows and bytes for both modes. ## Results **TL;DR:** Fusion reduced time by 14–22% for the narrow ascending/random cases at K=16, but increased time by 29–38% for wide descending data at K=16. Rows and bytes read were identical in every comparison. Fusion remains **disabled by default**. ### Setup - Same PR checkout and native release binary for both modes. Rust 1.97.1, the repository release profile (optimization level 3, thin LTO, one codegen unit), Spark 4.1.3, JDK 21.0.12.1. Linux x86-64 on AMD EPYC 9V74, 32 available CPUs and about 126 GiB RAM. - Spark runs with **`local[1]`**: the four-partition cases process four scan tasks sequentially, not on four concurrent Spark workers. AQE is disabled. Scan batch size is 4,096. - Each fixture contains 1,048,576 unique signed INT keys and either one or sixteen BIGINT payload columns. Keys are written ascending, descending, or as a deterministic permutation. One Parquet file per scan partition, Snappy compression, dictionaries disabled. Actual scan partition counts are asserted. - Queries use `SELECT * ... ORDER BY k ASC LIMIT K`, with K=16 or 100,000. File generation and Spark/native result/plan checks are outside the measurements. Query planning, scanning, selection, any shuffle, and collection are timed; local filesystem caches are warm. - Two separate JVM processes run each configuration in opposite mode orders: OFF→ON, then ON→OFF. Each case/mode/process has at least seven measured iterations, with 500 ms warmup and a 500 ms minimum measurement interval. The table gives pooled iteration-weighted means over at least 14 measurements per mode. Small changes should be treated as indicative. - All 96 native result/plan validations across the two passes succeeded. Both modes decoded all 1,048,576 rows and requested equal Parquet data bytes for every configuration. This PR does not add reader pruning. ### Timings Positive change means fusion was slower. | Layout | Partitions | Payload columns | K | Fusion off ms | Fusion on ms | Change | | --- | ---: | ---: | ---: | ---: | ---: | ---: | | ascending | 1 | 1 | 16 | 73.5 | 57.1 | -22.3% | | ascending | 1 | 1 | 100,000 | 123.7 | 108.8 | -12.1% | | descending | 1 | 1 | 16 | 182.9 | 182.5 | -0.2% | | descending | 1 | 1 | 100,000 | 457.1 | 473.8 | +3.6% | | random | 1 | 1 | 16 | 57.2 | 49.0 | -14.2% | | random | 1 | 1 | 100,000 | 598.9 | 583.9 | -2.5% | | ascending | 1 | 16 | 16 | 223.9 | 210.9 | -5.8% | | ascending | 1 | 16 | 100,000 | 374.8 | 365.7 | -2.4% | | descending | 1 | 16 | 16 | 295.8 | 407.1 | +37.6% | | descending | 1 | 16 | 100,000 | 682.8 | 812.5 | +19.0% | | random | 1 | 16 | 16 | 220.3 | 209.4 | -5.0% | | random | 1 | 16 | 100,000 | 952.8 | 1002.3 | +5.2% | | ascending | 4 | 1 | 16 | 81.6 | 64.4 | -21.0% | | ascending | 4 | 1 | 100,000 | 360.9 | 347.9 | -3.6% | | descending | 4 | 1 | 16 | 195.1 | 192.6 | -1.2% | | descending | 4 | 1 | 100,000 | 553.9 | 554.2 | +0.1% | | random | 4 | 1 | 16 | 70.9 | 57.1 | -19.5% | | random | 4 | 1 | 100,000 | 818.4 | 814.9 | -0.4% | | ascending | 4 | 16 | 16 | 248.8 | 234.4 | -5.8% | | ascending | 4 | 16 | 100,000 | 808.6 | 812.0 | +0.4% | | descending | 4 | 16 | 16 | 338.8 | 438.0 | +29.3% | | descending | 4 | 16 | 100,000 | 1073.3 | 1138.6 | +6.1% | | random | 4 | 16 | 16 | 251.9 | 232.1 | -7.9% | | random | 4 | 16 | 100,000 | 1376.2 | 1515.3 | +10.1% | The wide descending K=16 slowdown reproduced in both mode orders: the single-partition case rose from approximately 297→409 ms and 295→405 ms; the four-partition case rose from 337→439 ms and 341→437 ms. Fusion can reduce overlap between scan decoding and heap processing. These results measure its total effect; they do not isolate the overlap cost. Large-K wide random cases also regressed by 5–10%. The benchmark command is documented in `CometTopKBenchmark.scala`; the measured arguments were `1048576 1,4 1,16 16,100000 ascending,descending,random 7 off,on`, followed by the same arguments ending in `on,off`. Both benchmark processes completed all checks and measurements and reported `BUILD SUCCESS` with exit code 0. The Maven `exec:java` launcher subsequently printed a Hadoop shutdown-hook classloader error, after the timed work had completed. -- 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]
