kosiew commented on code in PR #24573:
URL: https://github.com/apache/datafusion/pull/24573#discussion_r3835972373
##########
datafusion/physical-plan/src/joins/sort_merge_join/tests.rs:
##########
@@ -5985,3 +5985,97 @@ async fn bitwise_spill_pending_stream() -> Result<()> {
Ok(())
}
+
+/// Regression test: deferred-filtered outer joins must not reorder their
+/// output.
+///
+/// `LEFT JOIN` advertises `maintains_input_order = [true, false]`, so the
+/// output must stay ordered on the streamed side. The final flush used to
+/// emit its batch directly instead of through the `output` coalescer, so any
+/// rows still buffered there from an earlier flush were emitted *after* it.
+///
+/// The shape below reproduces that: the first five keys each match a large
+/// buffered group, so the deferred-filter gate fires once per key and pushes
+/// a single-row batch into `output` (too small to complete a batch), while
+/// the last two keys match a single row each and so never trip the gate —
+/// leaving their rows for the final flush.
+#[tokio::test]
+async fn left_join_with_filter_preserves_streamed_order() -> Result<()> {
Review Comment:
Could we also add the symmetric RIGHT JOIN regression? RIGHT JOIN streams
the opposite child and has a different output-column and nulling layout, while
advertising `maintains_input_order = [false, true]`. A test asserting that the
right-side keys remain ordered would help protect the ordering contract on both
paths.
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -1665,20 +1667,19 @@ impl MaterializingSortMergeJoinStream {
// Multiple source batches: map each buffered_batch_idx to a
// contiguous source index, reserving source 0 for a null sentinel.
- let mut batch_idx_to_source: HashMap<usize, usize> = HashMap::new();
+ // A group spans only a handful of buffered batches, so a linear
+ // scan beats hashing here.
let mut source_batches: Vec<usize> = Vec::new();
- for (batch_idx, _, _) in matched_chunks {
- batch_idx_to_source.entry(*batch_idx).or_insert_with(|| {
- let idx = source_batches.len() + 1;
- source_batches.push(*batch_idx);
- idx
- });
- }
-
let mut interleave_indices: Vec<(usize, usize)> =
Vec::with_capacity(total_matched_rows);
for (batch_idx, _, right) in matched_chunks {
- let source = batch_idx_to_source[batch_idx];
+ let source = match source_batches.iter().position(|b| b ==
batch_idx) {
Review Comment:
Could we keep the previous `HashMap` approach here, or use another O(chunks)
index map? This changes the batch-to-source lookup from O(chunks) construction
to repeated linear searches, which can become O(chunks²) in this hot path. A
same-key buffered group can span many input batches, and `append_output_pair`
creates one chunk per buffered batch, so I don't think we can rely on the group
containing only a handful of chunks. With a large, batch-fragmented
duplicate-key group, this could result in a significant number of comparisons
during a freeze.
--
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]
