andygrove opened a new pull request, #6835: URL: https://github.com/apache/datafusion-comet/pull/6835
## Which issue does this PR close? Closes #6787. First item of the in-memory cache audit, #6834. ## Rationale for this change Spark's hash joins report the streamed side's order as their own, so Spark drops a sort above the join that this order satisfies. DataFusion's hash join keeps the unmatched probe rows of a right outer join in place only when the probe input declares an order, and otherwise appends them after the matched rows of each batch (`append_right_indices` in `datafusion-physical-plan`). Comet builds every hash join on the build side, so Spark's `LeftOuter` with `BuildRight` and `RightOuter` with `BuildLeft` both run as that right outer join. A streamed side that enters the native plan from the JVM arrives through a `ScanExec`, which declares no order, so the join returned those rows out of order with no sort left in the plan to fix them. #6787 reproduced this through `spark.comet.convert.inMemoryCache.enabled`, which is opt-in. Since #5634 made the in-memory cache default-on, `CometInMemoryTableScan` reaches the same path with default settings: a relation cached after `sortWithinPartitions` reports its order, and a left outer `CometHashJoin` or `CometBroadcastHashJoin` over it returns unsorted partitions where Spark's own read of the same cache is sorted. A Comet-side `outputOrdering` override on the join cannot fix this: `RemoveRedundantSorts` runs before Comet converts the join, so the sort is already gone by then. The native join has to honor Spark's claim instead. ## What changes are included in this PR? - `HashJoin` gets a `streamed_sort_orders` field. `CometHashJoin.doConvert` fills it with the streamed side's `outputOrdering` for the two outer join shapes above, and sends the join back to Spark with a fallback reason if that order cannot be expressed natively. Every other join type keeps the probe order as it is, so nothing is sent for them. - The planner declares that order on the probe input through a new pass-through `SortedInputExec` when the input declares none. The declaration is placed on that one input rather than on every `ScanExec` because an order visible to the whole native plan would also change how DataFusion's aggregates group. The dynamic join filter's consumer forwards its input's properties, so the declaration survives it. ## How are these changes tested? - `CometInMemoryCacheSuite`: "hash joins over a sorted cache keep the streamed side's order", for both `CometHashJoin` and `CometBroadcastHashJoin` over the native cache scan. The left outer join must keep the partitions sorted with no sort in the plan, and the inner join must still report the cache's order so that no sort is added. - `CometJoinSuite`: "outer hash joins keep a streamed side sorted outside the native plan in order", the #6787 shape through `CometSparkToColumnar`. - Both new tests fail on `main`. The full `CometJoinSuite` and `CometInMemoryCacheSuite` pass locally on Spark 4.1 (151 tests), as do the planner unit tests and clippy. -- 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]
