Skip to content

Commit 2836480

Browse files
authored
fix: ProjectionPushdown internal error on NestedLoopJoin mark joins (#22902)
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - Closes #22901. ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> A query whose subquery becomes a mark join (`LeftMark`/`RightMark`) can panic during the `ProjectionPushdown` physical optimization with an internal assertion error. The pushdown helper `try_pushdown_through_join` assumes the join output schema is the plain concatenation of its two children (`left ++ right`) and uses `join_table_borders` to split the projected columns into a left group and a right group by column index. Mark joins break this assumption: they append an extra `mark` boolean column that does not originate from either child, so the column-index split misroutes columns to the wrong side and the subsequent child-projection rewrite fails its name-match assertion. ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> In `HashJoinExec::try_swapping_with_projection` and `NestedLoopJoinExec::try_swapping_with_projection`, skip the `try_pushdown_through_join` path for mark joins (`LeftMark`/`RightMark`) and fall through to embedding the projection into the join instead. ## Are these changes tested? Yes. <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> ## Are there any user-facing changes? Yes, only bug fixes. <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> <!-- If there are any breaking changes to public APIs, please add the `api change` label. -->
1 parent ddc157d commit 2836480

3 files changed

Lines changed: 85 additions & 26 deletions

File tree

datafusion/physical-plan/src/joins/hash_join/exec.rs

Lines changed: 16 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1520,20 +1520,23 @@ impl ExecutionPlan for HashJoinExec {
15201520
return Ok(None);
15211521
}
15221522

1523+
// TODO: split by `col`/`JoinSide` instead so mark joins can also push down to children.
15231524
let schema = self.schema();
1524-
if let Some(JoinData {
1525-
projected_left_child,
1526-
projected_right_child,
1527-
join_filter,
1528-
join_on,
1529-
}) = try_pushdown_through_join(
1530-
projection,
1531-
self.left(),
1532-
self.right(),
1533-
self.on(),
1534-
&schema,
1535-
self.filter(),
1536-
)? {
1525+
if !matches!(self.join_type(), JoinType::LeftMark | JoinType::RightMark)
1526+
&& let Some(JoinData {
1527+
projected_left_child,
1528+
projected_right_child,
1529+
join_filter,
1530+
join_on,
1531+
}) = try_pushdown_through_join(
1532+
projection,
1533+
self.left(),
1534+
self.right(),
1535+
self.on(),
1536+
&schema,
1537+
self.filter(),
1538+
)?
1539+
{
15371540
self.builder()
15381541
.with_new_children(vec![
15391542
Arc::new(projected_left_child),

datafusion/physical-plan/src/joins/nested_loop_join.rs

Lines changed: 16 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -734,20 +734,23 @@ impl ExecutionPlan for NestedLoopJoinExec {
734734
return Ok(None);
735735
}
736736

737+
// TODO: split by `col`/`JoinSide` instead so mark joins can also push down to children.
737738
let schema = self.schema();
738-
if let Some(JoinData {
739-
projected_left_child,
740-
projected_right_child,
741-
join_filter,
742-
..
743-
}) = try_pushdown_through_join(
744-
projection,
745-
self.left(),
746-
self.right(),
747-
&[],
748-
&schema,
749-
self.filter(),
750-
)? {
739+
if !matches!(self.join_type(), JoinType::LeftMark | JoinType::RightMark)
740+
&& let Some(JoinData {
741+
projected_left_child,
742+
projected_right_child,
743+
join_filter,
744+
..
745+
}) = try_pushdown_through_join(
746+
projection,
747+
self.left(),
748+
self.right(),
749+
&[],
750+
&schema,
751+
self.filter(),
752+
)?
753+
{
751754
Ok(Some(Arc::new(NestedLoopJoinExec::try_new(
752755
Arc::new(projected_left_child),
753756
Arc::new(projected_right_child),

datafusion/sqllogictest/test_files/subquery.slt

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1317,6 +1317,59 @@ physical_plan
13171317
04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
13181318
05)------DataSourceExec: partitions=1, partition_sizes=[2]
13191319

1320+
query TT
1321+
explain select t1_id from t1
1322+
where t1_id > 40 or exists (select 1 from t2 where t2.t2_int = t1.t1_int)
1323+
----
1324+
logical_plan
1325+
01)Projection: t1.t1_id
1326+
02)--Filter: t1.t1_id > Int32(40) OR __correlated_sq_1.mark
1327+
03)----Projection: t1.t1_id, __correlated_sq_1.mark
1328+
04)------LeftMark Join: t1.t1_int = __correlated_sq_1.t2_int
1329+
05)--------TableScan: t1 projection=[t1_id, t1_int]
1330+
06)--------SubqueryAlias: __correlated_sq_1
1331+
07)----------TableScan: t2 projection=[t2_int]
1332+
physical_plan
1333+
01)FilterExec: t1_id@0 > 40 OR mark@1, projection=[t1_id@0]
1334+
02)--HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(t2_int@0, t1_int@1)], projection=[t1_id@0, mark@2]
1335+
03)----DataSourceExec: partitions=1, partition_sizes=[2]
1336+
04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
1337+
05)------DataSourceExec: partitions=1, partition_sizes=[2]
1338+
1339+
query I rowsort
1340+
select t1_id from t1
1341+
where t1_id > 40 or exists (select 1 from t2 where t2.t2_int = t1.t1_int)
1342+
----
1343+
11
1344+
33
1345+
44
1346+
1347+
query TT
1348+
explain select t1_id from t1
1349+
where t1_id > 40 or not exists (select 1 from t2 where t2.t2_int > t1.t1_int)
1350+
----
1351+
logical_plan
1352+
01)Projection: t1.t1_id
1353+
02)--Filter: t1.t1_id > Int32(40) OR NOT __correlated_sq_1.mark
1354+
03)----Projection: t1.t1_id, __correlated_sq_1.mark
1355+
04)------LeftMark Join: Filter: __correlated_sq_1.t2_int > t1.t1_int
1356+
05)--------TableScan: t1 projection=[t1_id, t1_int]
1357+
06)--------SubqueryAlias: __correlated_sq_1
1358+
07)----------TableScan: t2 projection=[t2_int]
1359+
physical_plan
1360+
01)FilterExec: t1_id@0 > 40 OR NOT mark@1, projection=[t1_id@0]
1361+
02)--NestedLoopJoinExec: join_type=RightMark, filter=t2_int@1 > t1_int@0, projection=[t1_id@0, mark@2]
1362+
03)----DataSourceExec: partitions=1, partition_sizes=[2]
1363+
04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
1364+
05)------DataSourceExec: partitions=1, partition_sizes=[2]
1365+
1366+
query I rowsort
1367+
select t1_id from t1
1368+
where t1_id > 40 or not exists (select 1 from t2 where t2.t2_int > t1.t1_int)
1369+
----
1370+
33
1371+
44
1372+
13201373
statement ok
13211374
set datafusion.explain.logical_plan_only = true;
13221375

0 commit comments

Comments
 (0)