sunchao opened a new pull request, #7190: URL: https://github.com/apache/datafusion-comet/pull/7190
## Which issue does this PR close? Closes #7189. ## Rationale for this change A left-semi join needs one matching right row for each left row. If a right-side integer key occurs thousands of times, sorting every copy adds work and retains data that cannot change the result. Removing repeated tuples before the sort can reduce both costs. ## What changes are included in this PR? - Add `spark.comet.exec.sortMergeJoin.rightKeyReduction.enabled`, off by default. The planner can insert the reducer below an explicit, matching right-side sort for an unfiltered batch left-semi join. It checks the complete integer input and the sort expressions, including Spark's encoding for null-safe equality. - Group equal rows within a batch using a DataFusion partial aggregate. After a short prefix, the reducer tries eligible batches and switches the partition to pass-through if a batch keeps at least 80% of its rows or the memory pool refuses a reservation. The sort still receives every right-side membership key, and the join preserves repeated left rows. - Expose input, output, bypassed rows, and reduction time in Spark SQL metrics. Add duplicate-heavy and all-unique cases to `CometSortMergeJoinBenchmark`. ## How are these changes tested? Five native operator tests passed with DataFusion 55.2.0 and Arrow 59.3.1. They cover nulls and composite signed keys, high-cardinality bypass, batch-size limits, memory refusal, error propagation, and sliced Arrow buffers. Two additional native controls exercised the real planner helper, including a filtered right input and null-safe keys. Spark 3.5.9 `test-compile`, Spotless, and `cargo fmt --check` passed. An optimized, single-partition native `SortExec` + `SortMergeJoinExec` control produced identical left rows with the feature off and on. Each arm used the same 524,288 prepared `Int32` rows in 128 batches of 4,096. Times cover execution of the in-memory sort and join, and exclude Spark, shuffle and I/O. The table gives medians of ten runs per arm after two warm-ups, with the execution order alternating. | Right-side input | Rows sent to sort, off → on | Native time, off → on | | --- | ---: | ---: | | 32 repeated keys plus nulls | 524,288 → 12,350 | 19.47 ms → 5.23 ms | | 524,288 unique keys | 524,288 → 524,288 | 32.43 ms → 32.95 ms | The in-tree Spark benchmark can be run with `make benchmark-org.apache.spark.sql.benchmark.CometSortMergeJoinBenchmark`. Its new cases verify the join result and reduction counters before timing the query. The local registry lacks the workspace's pinned `aws-config 1.13.0`. Upstream CI will validate the committed lockfile. -- 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]
