Repository navigation
fix: sort a Spark window's doubted input by the whole order it reports - #7555
Conversation
sortDoubtedWindowInputs sorted the input of a window that stays in Spark by the window's own keys only. Spark had already planned the operators above around the longer order the input reported, so a second window ordered by more keys, or a sortWithinPartitions it had removed, saw rows out of order. The sort now uses the whole reported order when it covers the window's keys. Closes apache#7553.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
The PR backlog is huge and the maintainers don't have the bandwidth to review them all, so here is an automated review. Feel free to ignore it.
I found no blocking concerns. Sorting by the whole order the window's input reports is the right shape for this fix. Spark's WindowExec passes child.outputOrdering on as its own, so the operators Spark planned above it can rely on more than requiredChildOrdering, and a sort by the window's own keys loses that. I checked that the new sortDoubtedWindowInputs branch behaves sensibly in the cases I could construct. When the child reports an order that satisfies the window's keys, the sort keeps the longer order. When Spark had to plan its own sort, mayReportFalseInputOrder stops at the SortExec and the rule does not fire. Because the rule is transformUp, a second Spark window above the first sees a SortExec below it and is not sorted again.
Against branch-1.2 the rule does not exist at all, so this only refines unreleased code from #7256 and changes nothing a user on the release sees. The contributor guide paragraph in adding_a_new_operator.md matches the new behavior.
The tests exercise both ways the order can be lost, and assertOnlySortUnderLowerWindow pins the plan shape the test depends on, which keeps the tests from passing for the wrong reason.
It seems to be stuck sitting in queuing. Not sure why. |
|
@andygrove CI on the earlier merge commit failed before starting any jobs, so |
…anner-strategy A native nested loop join over a sorted cache now stays in Spark, so the two window tests from apache#7555 move to a broadcast outer hash join over a native filter, which a Spark window above still doubts.
Which issue does this PR close?
Closes #7553.
#6785 should merge after this one: it restores the whole reported order the same way for windows that run natively, and its Spark fallback relies on this rule.
Rationale for this change
When a window stays in Spark over an input whose order Comet may report falsely,
sortDoubtedWindowInputssorts that input again, but only by the window's own keys. Spark had already planned the operators above the window around the longer order the input reported, so a second window ordered by more keys, or asortWithinPartitionsSpark removed as redundant, sees rows out of order. That happens when the input really is out of order past the window's keys (a native nested loop join with more than one build row), and also when it is in order but the inserted sort spills and the merge reorders ties.What changes are included in this PR?
How are these changes tested?
Two tests in
CometInMemoryCacheSuiterun a Spark window ordered byvunder one ordered by(v, w)over a sorted cache joined through a native nested loop join, and check the rows against Spark, the order of asortWithinPartitions("v", "w")above the lower window, and that the only sort in the plan is the one this rule adds, by(v, w). One uses two build rows, the other one build row with a forced spill. Both fail without the change.