Improve hash join exec builder

Prior the patch HashJoinExecBuilder constructed from an existing node
reseted some fields of the node, e.g. dynamic filters, metrics. It
significantly reduces usage scope of the builder.

This patch improves the implementation. Now builder created from the
existing node preserves all fields in case they have not been explicitly
updated. Also builder now tracks flag if it must recompute plan properties.

Closes apache#20270

Author: askalt

datafusion/physical-optimizer/src/enforce_distribution.rs

@@ -49,7 +49,7 @@ use datafusion_physical_plan::aggregates::{
 use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec;
 use datafusion_physical_plan::execution_plan::EmissionType;
 use datafusion_physical_plan::joins::{
-    CrossJoinExec, HashJoinExec, HashJoinExecBuilder, PartitionMode, SortMergeJoinExec,
+    CrossJoinExec, HashJoinExec, PartitionMode, SortMergeJoinExec,
 };
 use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
 use datafusion_physical_plan::repartition::RepartitionExec;
@@ -286,39 +286,26 @@ pub fn adjust_input_keys_ordering(
 ) -> Result<Transformed<PlanWithKeyRequirements>> {
     let plan = Arc::clone(&requirements.plan);
 
-    if let Some(HashJoinExec {
-        left,
-        right,
-        on,
-        filter,
-        join_type,
-        projection,
-        mode,
-        null_equality,
-        null_aware,
-        ..
-    }) = plan.as_any().downcast_ref::<HashJoinExec>()
+    if let Some(
+        exec @ HashJoinExec {
+            left,
+            on,
+            join_type,
+            mode,
+            ..
+        },
+    ) = plan.as_any().downcast_ref::<HashJoinExec>()
     {
         match mode {
             PartitionMode::Partitioned => {
                 let join_constructor = |new_conditions: (
                     Vec<(PhysicalExprRef, PhysicalExprRef)>,
                     Vec<SortOptions>,
                 )| {
-                    HashJoinExecBuilder::new(
-                        Arc::clone(left),
-                        Arc::clone(right),
-                        new_conditions.0,
-                        *join_type,
-                    )
-                    .with_filter(filter.clone())
-                    // TODO: although projection is not used in the join here, because projection pushdown is after enforce_distribution. Maybe we need to handle it later. Same as filter.
-                    .with_projection_ref(projection.clone())
-                    .with_partition_mode(PartitionMode::Partitioned)
-                    .with_null_equality(*null_equality)
-                    .with_null_aware(*null_aware)
-                    .build()
-                    .map(|e| Arc::new(e) as _)
+                    exec.builder()
+                        .with_partition_mode(PartitionMode::Partitioned)
+                        .with_on(new_conditions.0)
+                        .build_exec()
                 };
                 return reorder_partitioned_join_keys(
                     requirements,
@@ -612,18 +599,15 @@ pub fn reorder_join_keys_to_inputs(
     plan: Arc<dyn ExecutionPlan>,
 ) -> Result<Arc<dyn ExecutionPlan>> {
     let plan_any = plan.as_any();
-    if let Some(HashJoinExec {
-        left,
-        right,
-        on,
-        filter,
-        join_type,
-        projection,
-        mode,
-        null_equality,
-        null_aware,
-        ..
-    }) = plan_any.downcast_ref::<HashJoinExec>()
+    if let Some(
+        exec @ HashJoinExec {
+            left,
+            right,
+            on,
+            mode,
+            ..
+        },
+    ) = plan_any.downcast_ref::<HashJoinExec>()
     {
         if matches!(mode, PartitionMode::Partitioned) {
             let (join_keys, positions) = reorder_current_join_keys(
@@ -639,20 +623,11 @@ pub fn reorder_join_keys_to_inputs(
                     right_keys,
                 } = join_keys;
                 let new_join_on = new_join_conditions(&left_keys, &right_keys);
-                return Ok(Arc::new(
-                    HashJoinExecBuilder::new(
-                        Arc::clone(left),
-                        Arc::clone(right),
-                        new_join_on,
-                        *join_type,
-                    )
-                    .with_filter(filter.clone())
-                    .with_projection_ref(projection.clone())
+                return exec
+                    .builder()
                     .with_partition_mode(PartitionMode::Partitioned)
-                    .with_null_equality(*null_equality)
-                    .with_null_aware(*null_aware)
-                    .build()?,
-                ));
+                    .with_on(new_join_on)
+                    .build_exec();
             }
         }
     } else if let Some(SortMergeJoinExec {

datafusion/physical-optimizer/src/join_selection.rs

@@ -34,7 +34,7 @@ use datafusion_physical_expr::expressions::Column;
 use datafusion_physical_plan::execution_plan::EmissionType;
 use datafusion_physical_plan::joins::utils::ColumnIndex;
 use datafusion_physical_plan::joins::{
-    CrossJoinExec, HashJoinExec, HashJoinExecBuilder, NestedLoopJoinExec, PartitionMode,
+    CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode,
     StreamJoinPartitionMode, SymmetricHashJoinExec,
 };
 use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
@@ -192,14 +192,16 @@ pub(crate) fn try_collect_left(
                 Ok(Some(hash_join.swap_inputs(PartitionMode::CollectLeft)?))
             } else {
                 Ok(Some(Arc::new(
-                    HashJoinExecBuilder::from(hash_join)
+                    hash_join
+                        .builder()
                         .with_partition_mode(PartitionMode::CollectLeft)
                         .build()?,
                 )))
             }
         }
         (true, false) => Ok(Some(Arc::new(
-            HashJoinExecBuilder::from(hash_join)
+            hash_join
+                .builder()
                 .with_partition_mode(PartitionMode::CollectLeft)
                 .build()?,
         ))),
@@ -243,7 +245,8 @@ pub(crate) fn partitioned_hash_join(
         };
 
         Ok(Arc::new(
-            HashJoinExecBuilder::from(hash_join)
+            hash_join
+                .builder()
                 .with_partition_mode(partition_mode)
                 .build()?,
         ))

datafusion/physical-plan/src/execution_plan.rs

@@ -25,6 +25,7 @@ pub use crate::ordering::InputOrderMode;
 use crate::sort_pushdown::SortOrderPushdownResult;
 pub use crate::stream::EmptyRecordBatchStream;
 
+use arrow_schema::Schema;
 pub use datafusion_common::hash_utils;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 pub use datafusion_common::utils::project_schema;
@@ -38,7 +39,7 @@ pub use datafusion_physical_expr::{
 
 use std::any::Any;
 use std::fmt::Debug;
-use std::sync::Arc;
+use std::sync::{Arc, LazyLock};
 
 use crate::coalesce_partitions::CoalescePartitionsExec;
 use crate::display::DisplayableExecutionPlan;
@@ -1433,6 +1434,16 @@ pub enum CardinalityEffect {
     GreaterEqual,
 }
 
+/// Can be used in contexts where properties have not yet been initialized properly.
+pub(crate) static STUB_PROPERTIES: LazyLock<PlanProperties> = LazyLock::new(|| {
+    PlanProperties::new(
+        EquivalenceProperties::new(Arc::new(Schema::empty())),
+        Partitioning::UnknownPartitioning(1),
+        EmissionType::Final,
+        Boundedness::Bounded,
+    )
+});
+
 #[cfg(test)]
 mod tests {
     use std::any::Any;