WIP: Refactor NLJ with JoinAccelerator

benchmarks/.gitignore

@@ -1,5 +1,5 @@
 data
 data_csv
-./results/
+results
 venv
 !sql_benchmarks/**/results/

datafusion/core/src/physical_planner.rs

@@ -42,7 +42,8 @@ use crate::physical_plan::explain::ExplainExec;
 use crate::physical_plan::filter::FilterExecBuilder;
 use crate::physical_plan::joins::utils as join_utils;
 use crate::physical_plan::joins::{
-    CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec,
+    CrossJoinExec, HashJoinExec, NestedLoopJoinExecBuilder, PartitionMode,
+    SortMergeJoinExec,
 };
 use crate::physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
 use crate::physical_plan::projection::{ProjectionExec, ProjectionExpr};
@@ -1668,13 +1669,12 @@ impl DefaultPhysicalPlanner {
                         let left_side = side_of(lhs_logical)?;
                         let right_side = side_of(rhs_logical)?;
                         if left_side == Side::Both || right_side == Side::Both {
-                            return Ok(Arc::new(NestedLoopJoinExec::try_new(
+                            return build_nested_loop_join(
                                 physical_left,
                                 physical_right,
                                 join_filter,
-                                join_type,
-                                None,
-                            )?));
+                                *join_type,
+                            );
                         }
 
                         if left_side == Side::Right && right_side == Side::Left {
@@ -1709,13 +1709,12 @@ impl DefaultPhysicalPlanner {
                         )?)
                     } else {
                         // there is no equal join condition, use the nested loop join
-                        Arc::new(NestedLoopJoinExec::try_new(
+                        build_nested_loop_join(
                             physical_left,
                             physical_right,
                             join_filter,
-                            join_type,
-                            None,
-                        )?)
+                            *join_type,
+                        )?
                     }
                 } else if session_state.config().target_partitions() > 1
                     && session_state.config().repartition_joins()
@@ -2371,6 +2370,19 @@ fn extract_update_assignments(input: &Arc<LogicalPlan>) -> Result<Vec<(String, E
     Ok(assignments)
 }
 
+fn build_nested_loop_join(
+    physical_left: Arc<dyn ExecutionPlan>,
+    physical_right: Arc<dyn ExecutionPlan>,
+    join_filter: Option<join_utils::JoinFilter>,
+    join_type: JoinType,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    Ok(Arc::new(
+        NestedLoopJoinExecBuilder::new(physical_left, physical_right, join_type)
+            .with_filter(join_filter)
+            .build()?,
+    ))
+}
+
 /// Check if an assignment is an identity assignment (column = column)
 /// These are columns that are not being modified in the UPDATE
 fn is_identity_assignment(expr: &Expr, column_name: &str) -> bool {