(datafusion) branch main updated: Fix panic on RightMark hash joins when propagating ordering (#24718)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git


The following commit(s) were added to refs/heads/main by this push:
     new 1038d35634 Fix panic on RightMark hash joins when propagating ordering (#24718)
1038d35634 is described below

commit 1038d356347bcb67bcef85d91fe74c14273a6508
Author: Thomas Tanon <th...@pellissier-tanon.fr>
AuthorDate: Thu Aug 27 12:10:26 2026 +0000

    Fix panic on RightMark hash joins when propagating ordering (#24718)
    
    ## Which issue does this PR close?
    
    - Closes #24717 (affecting DF 55)
    
    ## Rationale for this change
    
    Fixes a panic in `enforce_sorting` when given a `RightMark` hash join:
    `HashjoinExec::maintains_input_order` has been changed to also return
    `true` on `RightMark` but `enforce_sorting` has not been updated
    
    ## What changes are included in this PR?
    
    Adds support of `RightMark` joins to `enforce_sorting` they keep the
    input sorting.
---
 .../tests/physical_optimizer/enforce_sorting.rs    | 34 ++++++++++++++++++++++
 .../enforce_sorting/sort_pushdown.rs               |  4 +--
 2 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
index a8162f137e..76af9b0c29 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_sorting.rs
@@ -243,6 +243,40 @@ async fn test_remove_unnecessary_sort5() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_propagate_in_right_mark_join() -> Result<()> {
+    let left_schema = create_test_schema2()?;
+    let right_schema = create_test_schema3()?;
+    let left_input = memory_exec(&left_schema);
+    let parquet_ordering = [sort_expr("a", &right_schema)].into();
+    let right_input =
+        parquet_exec_with_sort(right_schema.clone(), vec![parquet_ordering]);
+    let on = vec![(
+        Arc::new(Column::new_with_schema("col_a", &left_schema)?) as _,
+        Arc::new(Column::new_with_schema("c", &right_schema)?) as _,
+    )];
+    let join = hash_join_exec(left_input, right_input, on, None, &JoinType::RightMark)?;
+    let physical_plan = sort_exec([sort_expr("a", &join.schema())].into(), join);
+
+    let test = EnforceSortingTest::new(physical_plan).with_repartition_sorts(true);
+    assert_snapshot!(test.run(), @r"
+    Input Plan:
+    SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
+      HashJoinExec: mode=Partitioned, join_type=RightMark, on=[(col_a@0, c@2)]
+        DataSourceExec: partitions=1, partition_sizes=[0]
+        DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
+
+    Optimized Plan:
+    SortPreservingMergeExec: [a@0 ASC]
+      HashJoinExec: mode=Partitioned, join_type=RightMark, on=[(col_a@0, c@2)]
+        RepartitionExec: partitioning=Hash([col_a@0], 10), input_partitions=1
+          DataSourceExec: partitions=1, partition_sizes=[0]
+        RepartitionExec: partitioning=Hash([c@2], 10), input_partitions=1, maintains_sort_order=true
+          DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
+    ");
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_hash_join_interleaved_projection_preserves_parent_sort() -> Result<()> {
     let left_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 5c17ffbd1e..03a28e5f64 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
@@ -954,7 +954,7 @@ fn handle_custom_pushdown(
 }
 
 // For hash join we only maintain the input order for the right child
-// for join type: Inner, Right, RightSemi, RightAnti
+// for join type: Inner, Right, RightSemi, RightAnti, RightMark
 fn handle_hash_join(
     plan: &HashJoinExec,
     parent_required: OrderingRequirements,
@@ -1045,7 +1045,7 @@ fn build_join_column_index(plan: &HashJoinExec) -> Vec<ColumnIndex> {
                 .chain(map_fields(plan.right().schema(), JoinSide::Right))
                 .collect::<Vec<_>>()
         }
-        JoinType::RightSemi | JoinType::RightAnti => {
+        JoinType::RightSemi | JoinType::RightAnti | JoinType::RightMark => {
             map_fields(plan.right().schema(), JoinSide::Right)
         }
         _ => unreachable!("unexpected join type: {}", plan.join_type()),


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org