feat: Add HashJoin support for BuildRight (#437)

* feat: Add HashJoin support for BuildRight * Enable test * Update plan stability * More * Update plan stability * Refine * Fix * Update diffs to fix Spark tests * Update diff * Update Spark 3.4.3 diff * Use BuildSide enum * Update diffs * Update plan stability for Spark 4.0 * Update q5a plan
apache · Jun 8, 2024 · 32c61f5 · 32c61f5
1 parent 311e13e
commit 32c61f5
Show file tree

Hide file tree

Showing 479 changed files with 56,791 additions and 58,646 deletions.
diff --git a/core/Cargo.lock b/core/Cargo.lock
diff --git a/core/src/execution/datafusion/planner.rs b/core/src/execution/datafusion/planner.rs
@@ -35,6 +35,7 @@ use datafusion::{
         },
         AggregateExpr, PhysicalExpr, PhysicalSortExpr, ScalarFunctionExpr,
     },
+    physical_optimizer::join_selection::swap_hash_join,
     physical_plan::{
         aggregates::{AggregateMode as DFAggregateMode, PhysicalGroupBy},
         filter::FilterExec,
@@ -91,7 +92,7 @@ use crate::{
             agg_expr::ExprStruct as AggExprStruct, expr::ExprStruct, literal::Value, AggExpr, Expr,
             ScalarFunc,
         },
-        spark_operator::{operator::OpStruct, JoinType, Operator},
+        spark_operator::{operator::OpStruct, BuildSide, JoinType, Operator},
         spark_partitioning::{partitioning::PartitioningStruct, Partitioning as SparkPartitioning},
     },
 };
@@ -965,7 +966,7 @@ impl PhysicalPlanner {
                     join.join_type,
                     &join.condition,
                 )?;
-                let join = Arc::new(HashJoinExec::try_new(
+                let hash_join = Arc::new(HashJoinExec::try_new(
                     join_params.left,
                     join_params.right,
                     join_params.join_on,
@@ -977,7 +978,15 @@ impl PhysicalPlanner {
                     // `EqualNullSafe`, Spark will rewrite it during planning.
                     false,
                 )?);
-                Ok((scans, join))
+
+                // If the hash join is build right, we need to swap the left and right
+                let hash_join = if join.build_side == BuildSide::BuildLeft as i32 {
+                    hash_join
+                } else {
+                    swap_hash_join(hash_join.as_ref(), PartitionMode::Partitioned)?
+                };
+
+                Ok((scans, hash_join))
             }
         }
     }

diff --git a/core/src/execution/proto/operator.proto b/core/src/execution/proto/operator.proto
@@ -95,6 +95,7 @@ message HashJoin {
   repeated spark.spark_expression.Expr right_join_keys = 2;
   JoinType join_type = 3;
   optional spark.spark_expression.Expr condition = 4;
+  BuildSide build_side = 5;
 }
 
 message SortMergeJoin {
@@ -114,3 +115,8 @@ enum JoinType {
   LeftAnti = 6;
   RightAnti = 7;
 }
+
+enum BuildSide {
+  BuildLeft = 0;
+  BuildRight = 1;
+}