You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/16 10:12:59 UTC
(datafusion) branch gh-readonly-queue/main/pr-23591-80bc64f3d225fb528417008587507277c37e7455 created (now a287c0a959)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-23591-80bc64f3d225fb528417008587507277c37e7455
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at a287c0a959 fix: handle interleaved HashJoin projections in sort pushdown (#23591)
This branch includes the following new commits:
new a287c0a959 fix: handle interleaved HashJoin projections in sort pushdown (#23591)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: handle interleaved HashJoin projections in sort pushdown (#23591)
Posted by gi...@apache.org.
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-23591-80bc64f3d225fb528417008587507277c37e7455
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit a287c0a959310a6a52cafaf9492adc65d0755037
Author: xudong.w <wx...@gmail.com>
AuthorDate: Thu Jul 16 18:12:38 2026 +0800
fix: handle interleaved HashJoin projections in sort pushdown (#23591)
## Which issue does this PR close?
- Closes #23590.
## Rationale for this change
Sort pushdown through a `HashJoinExec` with an embedded projection
assumed that all projected left-side columns formed a contiguous prefix.
That assumption is not valid when the projection interleaves columns
from the two join inputs.
For a `CollectLeft` right join projected as `[right.col_a,
left.nullable_col]`, a required sort on output column 1 was therefore
mistaken for a sort from the right child. The optimizer pushed a sort on
`right.col_a` below the join, and `SanityCheckPlan` rejected the
resulting plan because it did not satisfy the parent sort on
`left.nullable_col`.
## What changes are included in this PR?
- Determine whether required sort columns come from the right child by
inspecting each projected output column's `JoinSide`, instead of
comparing output positions with the number of projected left columns.
- Add a regression test with an interleaved hash join projection. The
test runs `SanityCheckPlan` and snapshots the valid optimized plan.
## Are these changes tested?
Yes.
- On unmodified `main`, the new regression test fails in
`SanityCheckPlan`: the right input is sorted by `col_a`, while the
parent requires `nullable_col`.
- With this change, the focused regression test passes.
- `cargo fmt --all`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `RUST_BACKTRACE=1 cargo test --profile ci --exclude
datafusion-examples --exclude datafusion-benchmarks --exclude
datafusion-cli --workspace --lib --tests --bins --features
avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption`
## Are there any user-facing changes?
There are no API changes. Queries whose hash join projection interleaves
columns from both inputs no longer receive an invalid sort pushdown that
can fail physical-plan sanity checking.
---
.../tests/physical_optimizer/enforce_sorting.rs | 45 +++++++++++++++++++++-
.../enforce_sorting/sort_pushdown.rs | 11 +++---
2 files changed, 49 insertions(+), 7 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
index 8e8d222bb0..e9ad978b2e 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
@@ -33,7 +33,7 @@ use arrow::compute::{SortOptions};
use arrow::datatypes::{DataType, SchemaRef};
use datafusion_common::config::{ConfigOptions, CsvOptions};
use datafusion_common::tree_node::{TreeNode, TransformedResult};
-use datafusion_common::{create_array, Result, TableReference};
+use datafusion_common::{create_array, NullEquality, Result, TableReference};
use datafusion_datasource::file_scan_config::FileScanConfigBuilder;
use datafusion_datasource::source::DataSourceExec;
use datafusion_expr_common::operator::Operator;
@@ -44,6 +44,7 @@ use datafusion_physical_expr_common::sort_expr::{
};
use datafusion_physical_expr::{Distribution, Partitioning, PhysicalExpr};
use datafusion_physical_expr::expressions::{col, BinaryExpr, Column, NotExpr};
+use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode};
use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
use datafusion_physical_plan::repartition::RepartitionExec;
use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
@@ -237,6 +238,48 @@ async fn test_remove_unnecessary_sort5() -> Result<()> {
Ok(())
}
+#[tokio::test]
+async fn test_hash_join_interleaved_projection_preserves_parent_sort() -> Result<()> {
+ let left_schema = create_test_schema()?;
+ let right_schema = create_test_schema2()?;
+ let left = parquet_exec(left_schema.clone());
+ let right = parquet_exec(right_schema.clone());
+ let on = vec![(
+ Arc::new(Column::new_with_schema("nullable_col", &left_schema)?) as _,
+ Arc::new(Column::new_with_schema("col_a", &right_schema)?) as _,
+ )];
+ let join = Arc::new(HashJoinExec::try_new(
+ left,
+ right,
+ on,
+ None,
+ &JoinType::Right,
+ // Interleave a right-side column before a left-side column.
+ Some(vec![2, 0]),
+ PartitionMode::CollectLeft,
+ NullEquality::NullEqualsNothing,
+ false,
+ )?);
+ let ordering = [sort_expr("nullable_col", &join.schema())].into();
+ let physical_plan = sort_exec(ordering, join);
+
+ let mut config = ConfigOptions::new();
+ config.execution.target_partitions = 10;
+ let optimized_plan =
+ EnsureRequirements::new().optimize(Arc::clone(&physical_plan), &config)?;
+ let optimized_plan = SanityCheckPlan::new().optimize(optimized_plan, &config)?;
+
+ assert_snapshot!(displayable(optimized_plan.as_ref()).indent(true), @r"
+ SortPreservingMergeExec: [nullable_col@1 ASC]
+ SortExec: expr=[nullable_col@1 ASC], preserve_partitioning=[true]
+ HashJoinExec: mode=CollectLeft, join_type=Right, on=[(nullable_col@0, col_a@0)], projection=[col_a@2, nullable_col@0]
+ DataSourceExec: file_groups={1 group: [[x]]}, projection=[nullable_col, non_nullable_col], file_type=parquet
+ RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
+ DataSourceExec: file_groups={1 group: [[x]]}, projection=[col_a, col_b], file_type=parquet
+ ");
+ Ok(())
+}
+
#[tokio::test]
async fn test_do_not_remove_sort_with_limit() -> Result<()> {
let schema = create_test_schema()?;
diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs
index c1e42a7c9a..5c17ffbd1e 100644
--- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs
@@ -983,12 +983,11 @@ fn handle_hash_join(
} else {
column_indices.iter().collect()
};
- let len_of_left_fields = projected_indices
- .iter()
- .filter(|ci| ci.side == JoinSide::Left)
- .count();
-
- let all_from_right_child = all_indices.iter().all(|i| *i >= len_of_left_fields);
+ let all_from_right_child = all_indices.iter().all(|i| {
+ projected_indices
+ .get(*i)
+ .is_some_and(|ci| ci.side == JoinSide::Right)
+ });
let plan_children = plan.children();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org