fix: do not recompute hash join exec properties if not required

Author: askalt

datafusion/physical-plan/src/execution_plan.rs

@@ -1508,7 +1508,7 @@ pub fn reset_plan_states(plan: Arc<dyn ExecutionPlan>) -> Result<Arc<dyn Executi
 /// replace is requested.
 /// The size of `children` must be equal to the size of `ExecutionPlan::children()`.
 pub fn has_same_children_properties(
-    plan: &Arc<impl ExecutionPlan>,
+    plan: &impl ExecutionPlan,
     children: &[Arc<dyn ExecutionPlan>],
 ) -> Result<bool> {
     let old_children = plan.children();
@@ -1531,7 +1531,10 @@ pub fn has_same_children_properties(
 #[macro_export]
 macro_rules! check_if_same_properties {
     ($plan: expr, $children: expr) => {
-        if $crate::execution_plan::has_same_children_properties(&$plan, &$children)? {
+        if $crate::execution_plan::has_same_children_properties(
+            $plan.as_ref(),
+            &$children,
+        )? {
             let plan = $plan.with_new_children_and_same_properties($children);
             return Ok(::std::sync::Arc::new(plan));
         }

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

@@ -23,7 +23,10 @@ use std::sync::{Arc, OnceLock};
 use std::{any::Any, vec};
 
 use crate::ExecutionPlanProperties;
-use crate::execution_plan::{EmissionType, boundedness_from_children, stub_properties};
+use crate::execution_plan::{
+    EmissionType, boundedness_from_children, has_same_children_properties,
+    stub_properties,
+};
 use crate::filter_pushdown::{
     ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase,
     FilterPushdownPropagation,
@@ -373,9 +376,9 @@ impl HashJoinExecBuilder {
             children.len() == 2,
             "wrong number of children passed into `HashJoinExecBuilder`"
         );
+        self.preserve_properties &= has_same_children_properties(&self.exec, &children)?;
         self.exec.right = children.swap_remove(1);
         self.exec.left = children.swap_remove(0);
-        self.preserve_properties = false;
         Ok(self)
     }
 

datafusion/physical-plan/src/sorts/sort.rs

@@ -1216,7 +1216,7 @@ impl ExecutionPlan for SortExec {
         assert_eq!(children.len(), 1, "SortExec should have exactly one child");
         new_sort.input = Arc::clone(&children[0]);
 
-        if !has_same_children_properties(&self, &children)? {
+        if !has_same_children_properties(self.as_ref(), &children)? {
             // Recompute the properties based on the new input since they may have changed
             let (cache, sort_prefix) = Self::compute_properties(
                 &new_sort.input,