viirya commented on code in PR #25489:
URL: https://github.com/apache/datafusion/pull/25489#discussion_r4052406774
##########
datafusion/physical-plan/src/joins/sort_merge_join/filter.rs:
##########
@@ -153,27 +153,21 @@ pub fn get_filter_columns(
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
+ let Some(f) = join_filter else {
+ return vec![];
+ };
+
+ // The filter's intermediate schema lists its columns in `column_indices`
+ // order, which need not put every left column before every right one
+ // (a filter swapped along with the join's inputs does the opposite).
+ f.column_indices()
Review Comment:
Since you're rewriting this function anyway: could the doc comment say the
parameters are **join-side-relative, not streamed/buffered**?
The caller has to re-map them. `materializing_stream.rs:1551-1553` passes
them swapped for `JoinType::Right`, because there the probe side is the join's
right input (`exec.rs:238`) while the locals are named `left_columns` /
`right_columns` after streamed/buffered. The correctness of this fix depends on
that swap, but nothing here says so. A future caller inside the SMJ — where the
locals carry exactly those names — would pass them positionally and be silently
wrong for `Right` only, which is the same failure mode this PR is fixing.
Something like "callers pass arrays in join-side order, not
streamed/buffered order; see the `JoinType::Right` call site in
`materializing_stream.rs`" would earn its keep.
Separately, on the comment just above: it says what the order is, but not
why the old code was wrong. The load-bearing fact — that the result is zipped
against the filter's intermediate schema **positionally**, so a left-first
vector against a right-first schema silently puts the wrong array in each slot
— is in the PR description but not in the code. Having it here means the next
reader doesn't need to find this PR to reconstruct the invariant.
##########
datafusion/physical-plan/src/joins/sort_merge_join/filter.rs:
##########
@@ -153,27 +153,21 @@ pub fn get_filter_columns(
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
+ let Some(f) = join_filter else {
+ return vec![];
+ };
+
+ // The filter's intermediate schema lists its columns in `column_indices`
+ // order, which need not put every left column before every right one
+ // (a filter swapped along with the join's inputs does the opposite).
+ f.column_indices()
+ .iter()
+ .filter_map(|col_index| match col_index.side {
+ JoinSide::Left => Some(Arc::clone(&left_columns[col_index.index])),
+ JoinSide::Right =>
Some(Arc::clone(&right_columns[col_index.index])),
+ JoinSide::None => None,
Review Comment:
This is where the change diverges from `bitwise_stream.rs`, which the
description cites as the model: `bitwise_stream.rs:1245` and `:1253` return
`internal_err!("Unexpected JoinSide::None in filter")` rather than skipping.
I think erroring is the better behavior, because `None` in a *filter's*
`column_indices` looks like an invariant violation rather than a legitimate
case:
- `JoinSide::None` exists for the mark column in a join's **output**
`column_indices` (`utils.rs:316`, `:328` in `build_join_schema`), not a
filter's.
- Every construction site of a filter's `column_indices` treats it as
impossible: `projection_pushdown.rs:209`, `:217`, `:263`, `:642` are all
`unreachable!("Mark join not supported")`; `join_selection.rs:452` and
`sort_pushdown.rs:888` likewise.
- Decisively: the intermediate schema is built *from* `column_indices`, one
field per entry (`projection_pushdown.rs:211-219`). A `None` entry can't even
acquire a schema field.
So if one ever appeared, skipping yields a `filter_columns` shorter than
`f.schema()`, and the failure surfaces a few lines later at
`materializing_stream.rs:1566` as an opaque column-count mismatch instead of a
named internal error pointing at the real cause.
This is also the one line Codecov flags as uncovered, which is consistent
with it being unreachable by construction.
I realize matching `bitwise_stream.rs` means returning
`Result<Vec<ArrayRef>>`, which ripples to the two call sites, and there's a
fair argument that's beyond a minimal fix. If you'd rather keep the skip, could
you drop the "matches `bitwise_stream.rs`" / "as before" framing from the
description and leave a comment here on why skipping is safe? As written the
description reads as though the two implementations agree.
##########
datafusion/physical-plan/src/joins/sort_merge_join/tests.rs:
##########
@@ -943,6 +943,66 @@ 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?;
Review Comment:
Two suggestions on coverage.
**A `JoinType::Right` variant.** `Right` is the one join type in the
materializing stream whose probe side is the join's right input, so it's where
the join-side and streamed/buffered coordinate systems meet, and it takes the
swapped-argument path at `materializing_stream.rs:1551`. I traced it: it's
broken on main the same way and fixed by this change, but nothing pins it — so
a future refactor of that `if`/`else` could reintroduce the bug on `Right`
alone with the suite green. The existing filter tests
(`join_right_different_columns_count_with_filter`,
`join_left_different_columns_count_with_filter`, the mark/semi/anti ones) all
use canonical left-first layouts, so `Right` x right-before-left is the
highest-risk untested combination right now.
**A test through `swap_inputs()`.** This is the more valuable one, and
there's already a template for it: `swap_inputs_swaps_the_projection`
(`tests.rs:6735`) builds a join, calls `swap_inputs()`, executes both, and
asserts the results are equal. It passes `filter: None`, so the filter path is
untouched. Passing a real `JoinFilter` through that same shape would exercise
the documented trigger end-to-end, and the differential assertion (swap must
not change the result) is stronger than a snapshot because it states the
contract rather than one expected table. It would also guard the rest of that
path, which nothing does today.
--
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]
