This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25489-4fab63d0dfc48b8dd641f534f0da367144a184f2 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 0b95fc29ff143397c4dccb462ce4cf69dcb33ad0 Author: Jay Zhan <[email protected]> AuthorDate: Mon Sep 21 13:31:55 2026 +0000 fix: sort-merge join filter columns follow the filter's own column order (#25489) ## Which issue does this PR close? - No separate issue. Split out of #25217 so it can be reviewed on its own. ## Rationale for this change A `SortMergeJoinExec` with a join filter returns wrong rows when the filter's `column_indices` do not list every left column before every right column. `JoinFilter::swap` produces exactly that layout, so any plan that goes through the public `SortMergeJoinExec::swap_inputs()` is affected, as is any `JoinFilter` built by hand or by a custom optimizer rule. For example, a Left join on `t2.b1 = t1.b1 AND t2.a2 > t1.a1` whose filter indices are `[Right(a1), Left(a2)]` is evaluated as `a1 > a2`: expected actual | 10 | 4 | 1 | 4 | 7 | | 10 | 4 | | | | | 20 | 5 | | | | | 20 | 5 | 21 | 5 | 8 | When the misplaced columns share a type there is no error, only wrong results. Queries planned from SQL are not affected today: the physical planner always builds the filter with left columns first, and `JoinSelection` does not swap sort-merge joins. ## What changes are included in this PR? `get_filter_columns` (used by the materializing SMJ stream) collected all left columns, then all right columns, and the result was zipped against the filter's intermediate schema, which is in `column_indices` order. It now walks `column_indices` once and takes each column from the side it names. This matches what the semi/anti/mark stream already does in `bitwise_stream.rs`. Columns with `JoinSide::None` are skipped, as before. ## What is the testing strategy for this PR? New unit test `join_left_with_filter_columns_right_before_left` in `sort_merge_join/tests.rs`: a Left join whose filter lists a right column before a left one. It fails on `main` with the wrong rows shown above and passes with the fix. The existing `sort_merge_join` tests pass unchanged. ## Are there any user-facing changes? No API changes. Sort-merge joins with a filter in right-before-left column order now return correct results. --- .../src/joins/sort_merge_join/filter.rs | 50 ++-- .../joins/sort_merge_join/materializing_stream.rs | 7 +- .../src/joins/sort_merge_join/tests.rs | 270 +++++++++++++++++++++ 3 files changed, 302 insertions(+), 25 deletions(-) diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/filter.rs b/datafusion/physical-plan/src/joins/sort_merge_join/filter.rs index 306a154666..890272b3cf 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/filter.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/filter.rs @@ -32,7 +32,7 @@ use arrow::array::{ use arrow::compute::kernels::zip::zip; use arrow::compute::{self, filter_record_batch}; use arrow::datatypes::SchemaRef; -use datafusion_common::{JoinSide, JoinType, Result}; +use datafusion_common::{JoinSide, JoinType, Result, internal_err}; use crate::joins::utils::JoinFilter; @@ -148,32 +148,36 @@ pub fn needs_deferred_filtering( /// Gets the arrays which join filters are applied on /// /// Extracts the columns needed for filter evaluation from left and right batch columns +/// +/// `left_columns` and `right_columns` are relative to the join's inputs (the +/// [`JoinSide`] recorded in the filter's `column_indices`), not to the +/// streamed/buffered sides. The two differ for `JoinType::Right`, where the +/// streamed side is the join's right input, so callers holding +/// streamed/buffered arrays must swap them; see the `JoinType::Right` call +/// site in `materializing_stream.rs`. pub fn get_filter_columns( join_filter: Option<&JoinFilter>, left_columns: &[ArrayRef], right_columns: &[ArrayRef], -) -> Vec<ArrayRef> { - let mut filter_columns = vec![]; - - if let Some(f) = join_filter { - let left_columns: Vec<ArrayRef> = f - .column_indices() - .iter() - .filter(|col_index| col_index.side == JoinSide::Left) - .map(|i| Arc::clone(&left_columns[i.index])) - .collect(); - let right_columns: Vec<ArrayRef> = f - .column_indices() - .iter() - .filter(|col_index| col_index.side == JoinSide::Right) - .map(|i| Arc::clone(&right_columns[i.index])) - .collect(); - - filter_columns.extend(left_columns); - filter_columns.extend(right_columns); - } - - filter_columns +) -> Result<Vec<ArrayRef>> { + let Some(f) = join_filter else { + return Ok(vec![]); + }; + + // The returned arrays are zipped positionally against the filter's + // intermediate schema, which lists its columns in `column_indices` order. + // That order need not put every left column before every right one (a + // filter swapped along with the join's inputs does the opposite), so + // emitting all left arrays and then all right arrays would silently put + // the wrong array in each slot. + f.column_indices() + .iter() + .map(|col_index| match col_index.side { + JoinSide::Left => Ok(Arc::clone(&left_columns[col_index.index])), + JoinSide::Right => Ok(Arc::clone(&right_columns[col_index.index])), + JoinSide::None => internal_err!("Unexpected JoinSide::None in filter"), + }) + .collect() } /// Determines if current index is the last occurrence of a row diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs index 96b903f63b..1f9875aca7 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs @@ -1547,10 +1547,13 @@ impl MaterializingSortMergeJoinStream { let right_columns = self.materialize_right_columns(matched_chunks, total_matched_rows)?; + // `get_filter_columns` takes arrays in join-side order. `left_columns` + // / `right_columns` here are streamed / buffered, and for a Right join + // the streamed side is the join's right input, hence the swap. let filter_columns = if self.join_type == JoinType::Right { - get_filter_columns(self.filter.as_ref(), &right_columns, &left_columns) + get_filter_columns(self.filter.as_ref(), &right_columns, &left_columns)? } else { - get_filter_columns(self.filter.as_ref(), &left_columns, &right_columns) + get_filter_columns(self.filter.as_ref(), &left_columns, &right_columns)? }; let columns = if self.join_type != JoinType::Right { diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs index ea9299abdd..2d97df9964 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs @@ -943,6 +943,200 @@ async fn join_left_different_columns_count_with_filter() -> Result<()> { Ok(()) } +/// A filter whose intermediate schema lists a right column before a left one +/// (the layout `JoinFilter::swap` produces when a join's inputs are swapped) +#[tokio::test] +async fn join_left_with_filter_columns_right_before_left() -> Result<()> { + // select * + // from t2 + // left join t1 on t2.b1 = t1.b1 and t2.a2 > t1.a1 + + let left = build_table_two_cols( + ("a2", &vec![10, 20, 30]), + ("b1", &vec![4, 5, 6]), // 6 does not exist on the right + ); + + let right = build_table( + ("a1", &vec![1, 21, 3]), // 20(t2.a2) > 1(t1.a1) + ("b1", &vec![4, 5, 7]), + ("c1", &vec![7, 8, 9]), + ); + + let on = vec![( + Arc::new(Column::new_with_schema("b1", &left.schema())?) as _, + Arc::new(Column::new_with_schema("b1", &right.schema())?) as _, + )]; + + let filter = JoinFilter::new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("a2", 1)), + Operator::Gt, + Arc::new(Column::new("a1", 0)), + )), + vec![ + ColumnIndex { + index: 0, + side: JoinSide::Right, + }, + ColumnIndex { + index: 0, + side: JoinSide::Left, + }, + ], + Arc::new(Schema::new(vec![ + Field::new("a1", DataType::Int32, true), + Field::new("a2", DataType::Int32, true), + ])), + ); + + let (_, batches) = join_collect_with_filter(left, right, on, filter, Left).await?; + + assert_snapshot!(batches_to_string(&batches), @r" + +----+----+----+----+----+ + | a2 | b1 | a1 | b1 | c1 | + +----+----+----+----+----+ + | 10 | 4 | 1 | 4 | 7 | + | 20 | 5 | | | | + | 30 | 6 | | | | + +----+----+----+----+----+ + "); + Ok(()) +} + +/// Same filter layout for a right join, whose streamed side is the join's +/// right input, so the streamed/buffered arrays reach `get_filter_columns` +/// swapped back into join-side order +#[tokio::test] +async fn join_right_with_filter_columns_right_before_left() -> Result<()> { + // select * + // from t2 + // right join t1 on t2.b1 = t1.b1 and t2.a2 > t1.a1 + + let left = build_table_two_cols(("a2", &vec![10, 20, 30]), ("b1", &vec![4, 5, 6])); + + let right = build_table( + ("a1", &vec![1, 21, 3]), // 20(t2.a2) > 21(t1.a1) is false + ("b1", &vec![4, 5, 7]), // 7 does not exist on the left + ("c1", &vec![7, 8, 9]), + ); + + let on = vec![( + Arc::new(Column::new_with_schema("b1", &left.schema())?) as _, + Arc::new(Column::new_with_schema("b1", &right.schema())?) as _, + )]; + + let filter = JoinFilter::new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("a2", 1)), + Operator::Gt, + Arc::new(Column::new("a1", 0)), + )), + vec![ + ColumnIndex { + index: 0, + side: JoinSide::Right, + }, + ColumnIndex { + index: 0, + side: JoinSide::Left, + }, + ], + Arc::new(Schema::new(vec![ + Field::new("a1", DataType::Int32, true), + Field::new("a2", DataType::Int32, true), + ])), + ); + + let (_, batches) = join_collect_with_filter(left, right, on, filter, Right).await?; + + assert_snapshot!(batches_to_sort_string(&batches), @r" + +----+----+----+----+----+ + | a2 | b1 | a1 | b1 | c1 | + +----+----+----+----+----+ + | | | 21 | 5 | 8 | + | | | 3 | 7 | 9 | + | 10 | 4 | 1 | 4 | 7 | + +----+----+----+----+----+ + "); + Ok(()) +} + +/// A filter whose intermediate schema interleaves the two sides (Left, Right, +/// Left), which neither "left columns first" nor "right columns first" produces +#[tokio::test] +async fn join_left_with_filter_columns_interleaved() -> Result<()> { + // select * + // from t1 + // left join t2 on t1.b1 = t2.b1 and t1.a1 < t2.a2 and t2.a2 < t1.c1 + + let left = build_table( + ("a1", &vec![1, 2, 3]), + ("b1", &vec![4, 5, 6]), // 6 does not exist on the right + ("c1", &vec![10, 3, 30]), + ); + + // Grouping the filter columns by side would evaluate `a1 < c1 and c1 < a2`, + // which is false for b1 = 4 and true for b1 = 5: the opposite of the filter. + let right = build_table( + ("a2", &vec![5, 20, 7]), // 1 < 5 < 10, but 2 < 20 < 3 is false + ("b1", &vec![4, 5, 7]), + ("c2", &vec![70, 80, 90]), + ); + + let on = vec![( + Arc::new(Column::new_with_schema("b1", &left.schema())?) as _, + Arc::new(Column::new_with_schema("b1", &right.schema())?) as _, + )]; + + let filter = JoinFilter::new( + Arc::new(BinaryExpr::new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("a1", 0)), + Operator::Lt, + Arc::new(Column::new("a2", 1)), + )), + Operator::And, + Arc::new(BinaryExpr::new( + Arc::new(Column::new("a2", 1)), + Operator::Lt, + Arc::new(Column::new("c1", 2)), + )), + )), + vec![ + ColumnIndex { + index: 0, + side: JoinSide::Left, + }, + ColumnIndex { + index: 0, + side: JoinSide::Right, + }, + ColumnIndex { + index: 2, + side: JoinSide::Left, + }, + ], + Arc::new(Schema::new(vec![ + Field::new("a1", DataType::Int32, true), + Field::new("a2", DataType::Int32, true), + Field::new("c1", DataType::Int32, true), + ])), + ); + + let (_, batches) = join_collect_with_filter(left, right, on, filter, Left).await?; + + assert_snapshot!(batches_to_string(&batches), @r" + +----+----+----+----+----+----+ + | a1 | b1 | c1 | a2 | b1 | c2 | + +----+----+----+----+----+----+ + | 1 | 4 | 10 | 5 | 4 | 70 | + | 2 | 5 | 3 | | | | + | 3 | 6 | 30 | | | | + +----+----+----+----+----+----+ + "); + Ok(()) +} + #[tokio::test] async fn join_left_mark_different_columns_count_with_filter() -> Result<()> { // select * @@ -6774,6 +6968,82 @@ async fn swap_inputs_swaps_the_projection() -> Result<()> { Ok(()) } +/// Swapping the inputs swaps the filter too, which lists its right columns before +/// its left ones afterwards. The swapped join must still evaluate the same predicate. +#[tokio::test] +async fn swap_inputs_swaps_the_filter() -> Result<()> { + let left = build_table( + ("a1", &vec![1, 2, 3, 4]), // 4 does not exist on the right + ("b1", &vec![10, 20, 30, 40]), + ("c1", &vec![100, 200, 300, 400]), + ); + let right = build_table( + ("a2", &vec![1, 2, 3, 5]), // 5 does not exist on the left + ("b2", &vec![11, 15, 33, 55]), + ("c2", &vec![111, 222, 333, 555]), + ); + let on: JoinOn = vec![( + Arc::new(Column::new("a1", 0)) as _, + Arc::new(Column::new("a2", 0)) as _, + )]; + // b1 > b2 holds for key 2 alone, and b2 > b1 for keys 1 and 3, so reading the + // two columns from the wrong sides changes the result. + let filter = JoinFilter::new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("b1", 0)), + Operator::Gt, + Arc::new(Column::new("b2", 1)), + )), + vec![ + ColumnIndex { + index: 1, + side: JoinSide::Left, + }, + ColumnIndex { + index: 1, + side: JoinSide::Right, + }, + ], + Arc::new(Schema::new(vec![ + Field::new("b1", DataType::Int32, true), + Field::new("b2", DataType::Int32, true), + ])), + ); + + for join_type in [ + Inner, Left, Right, Full, LeftSemi, LeftAnti, RightSemi, RightAnti, + ] { + let join = SortMergeJoinExec::try_new( + Arc::clone(&left), + Arc::clone(&right), + on.clone(), + Some(filter.clone()), + join_type, + vec![SortOptions::default()], + NullEquality::NullEqualsNothing, + )?; + + let swapped = join.swap_inputs()?; + assert_eq!( + swapped.schema().fields(), + join.schema().fields(), + "swapping must not change what the {join_type:?} join emits" + ); + + let task_ctx = Arc::new(TaskContext::default()); + let expected = common::collect(join.execute(0, Arc::clone(&task_ctx))?).await?; + let actual = common::collect(swapped.execute(0, task_ctx)?).await?; + // Swapping changes which side is streamed, and with it the row order. + assert_eq!( + batches_to_sort_string(&expected), + batches_to_sort_string(&actual), + "swapping must not change the result of the {join_type:?} join" + ); + } + + Ok(()) +} + /// An empty projection still changes the output schema, and the row count has to /// survive it: `SELECT count(1)` over a join needs the rows but none of the columns. #[tokio::test] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
