feat: support co-partitioned range right-side equi hash joins

datafusion/core/tests/physical_optimizer/enforce_distribution.rs

@@ -870,6 +870,87 @@ fn range_inner_hash_join_rehashes_incompatible_range_partitioning() -> Result<()
     Ok(())
 }
 
+// Kept as a unit test: `RightMark` join plans cannot be produced from SQL (they
+// only arise when a statistical swap flips a `LeftMark` join around), so this
+// reuse plan shape has no `range_partitioning.slt` equivalent. The end-to-end
+// mark-join coverage lives in that file's "Mark Join Marker Semantics" case.
+#[test]
+fn range_right_mark_hash_join_reuses_range_partitioning() -> Result<()> {
+    let left = parquet_exec_with_output_partitioning(range_partitioning(
+        "a",
+        [10, 20, 30],
+        SortOptions::default(),
+    )?);
+    let right = parquet_exec_with_output_partitioning(range_partitioning(
+        "a",
+        [10, 20, 30],
+        SortOptions::default(),
+    )?);
+    let join_on = vec![(
+        Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
+        Arc::new(Column::new_with_schema("a", &right.schema())?) as _,
+    )];
+    let join = hash_join_exec(left, right, &join_on, &JoinType::RightMark);
+
+    let plan = TestConfig::default()
+        .with_query_execution_partitions(4)
+        .to_plan(join, &DISTRIB_DISTRIB_SORT);
+
+    assert_plan!(
+        plan,
+        @r"
+    HashJoinExec: mode=Partitioned, join_type=RightMark, on=[(a@0, a@0)]
+      DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), file_type=parquet
+      DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), file_type=parquet
+    "
+    );
+
+    Ok(())
+}
+
+// Kept as a unit test: a descending Range input cannot be registered through the
+// sqllogictest fixtures (they only declare ascending Range layouts), so the
+// sort-direction axis of the co-partition check has no `range_partitioning.slt`
+// equivalent. Verifies that opposite sort options force Hash repartitioning.
+#[test]
+fn range_right_semi_hash_join_rehashes_incompatible_sort_options() -> Result<()> {
+    let left = parquet_exec_with_output_partitioning(range_partitioning(
+        "a",
+        [10, 20, 30],
+        SortOptions::default(),
+    )?);
+    let right = parquet_exec_with_output_partitioning(range_partitioning(
+        "a",
+        [30, 20, 10],
+        SortOptions {
+            descending: true,
+            nulls_first: true,
+        },
+    )?);
+    let join_on = vec![(
+        Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
+        Arc::new(Column::new_with_schema("a", &right.schema())?) as _,
+    )];
+    let join = hash_join_exec(left, right, &join_on, &JoinType::RightSemi);
+
+    let plan = TestConfig::default()
+        .with_query_execution_partitions(4)
+        .to_plan(join, &DISTRIB_DISTRIB_SORT);
+
+    assert_plan!(
+        plan,
+        @r"
+    HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(a@0, a@0)]
+      RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4
+        DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 ASC], [(10), (20), (30)], 4), file_type=parquet
+      RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4
+        DataSourceExec: file_groups={4 groups: [[p0], [p1], [p2], [p3]]}, projection=[a, b, c, d, e], output_partitioning=Range([a@0 DESC], [(30), (20), (10)], 4), file_type=parquet
+    "
+    );
+
+    Ok(())
+}
+
 #[test]
 fn multi_hash_joins() -> Result<()> {
     let left = parquet_exec();

datafusion/core/tests/physical_optimizer/sanity_checker.rs

@@ -443,6 +443,32 @@ fn test_partitioned_hash_join_requires_co_partitioned_children() -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn test_partitioned_right_hash_join_requires_co_partitioned_children() -> Result<()> {
+    let schema = create_test_schema2();
+    let join_on = vec![(col("a", &schema)?, col("a", &schema)?)];
+
+    let compatible_join = hash_join_exec(
+        range_partitioned_exec(&schema, "a", [10])?,
+        range_partitioned_exec(&schema, "a", [10])?,
+        join_on.clone(),
+        None,
+        &JoinType::Right,
+    )?;
+    assert_sanity_check(&compatible_join, true);
+
+    let incompatible_join = hash_join_exec(
+        range_partitioned_exec(&schema, "a", [10])?,
+        range_partitioned_exec(&schema, "a", [20])?,
+        join_on,
+        None,
+        &JoinType::Right,
+    )?;
+    assert_sanity_check(&incompatible_join, false);
+
+    Ok(())
+}
+
 #[tokio::test]
 /// Tests that plan is valid when the sort requirements are satisfied.
 async fn test_bounded_window_agg_sort_requirement() -> Result<()> {

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

@@ -1293,7 +1293,16 @@ impl ExecutionPlan for HashJoinExec {
             ]),
         };
 
-        if self.mode == PartitionMode::Partitioned && self.join_type == JoinType::Inner {
+        if self.mode == PartitionMode::Partitioned
+            && matches!(
+                self.join_type,
+                JoinType::Inner
+                    | JoinType::Right
+                    | JoinType::RightSemi
+                    | JoinType::RightAnti
+                    | JoinType::RightMark
+            )
+        {
             requirements.allow_range_satisfaction_for_key_partitioning()
         } else {
             requirements

datafusion/sqllogictest/src/test_context/range_partitioning.rs

@@ -84,7 +84,7 @@ pub(super) fn register_range_partitioned_table(ctx: &SessionContext) {
         "range_partitioned_shifted",
         Path::new(env!("CARGO_MANIFEST_DIR"))
             .join("test_files/scratch_range_partitioning/range_partitioned_shifted"),
-        schema,
+        Arc::clone(&schema),
         [
             "1,1,10\n5,2,50\n10,1,100\n",
             "15,2,150\n",
@@ -93,6 +93,34 @@ pub(super) fn register_range_partitioned_table(ctx: &SessionContext) {
         ],
         Some(shifted_output_partitioning),
     );
+
+    // Same rows as `range_partitioned` but split into only three range
+    // partitions on `range_key`. Used to exercise the co-partition check when
+    // two Range inputs disagree on partition count.
+    let narrow_output_partitioning = Partitioning::Range(
+        RangePartitioning::try_new(
+            vec![col("range_key").sort(true, true)],
+            vec![
+                SplitPoint::new(vec![ScalarValue::Int32(Some(10))]),
+                SplitPoint::new(vec![ScalarValue::Int32(Some(20))]),
+            ],
+        )
+        .expect("range partitioning should be valid"),
+    );
+
+    register_csv_listing_table(
+        ctx,
+        "range_partitioned_narrow",
+        Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("test_files/scratch_range_partitioning/range_partitioned_narrow"),
+        schema,
+        [
+            "1,1,10\n5,2,50\n",
+            "10,1,100\n15,2,150\n",
+            "20,1,200\n25,2,250\n30,1,300\n35,2,350\n",
+        ],
+        Some(narrow_output_partitioning),
+    );
 }
 
 fn register_csv_listing_table(