update

Author: xudong963

datafusion/core/tests/physical_optimizer/partition_statistics.rs

@@ -1452,7 +1452,7 @@ mod test {
                 },
             ],
         };
-        assert_eq!(statistics[0], expected_p0_statistics);
+        assert_eq!(*statistics[0], expected_p0_statistics);
 
         // Test Partitioned mode
         let partitioned_join = Arc::new(HashJoinExec::try_new(
@@ -1526,7 +1526,7 @@ mod test {
                 },
             ],
         };
-        assert_eq!(statistics[0], expected_p0_statistics);
+        assert_eq!(*statistics[0], expected_p0_statistics);
 
         // Test Auto mode - should fall back to getting all partition statistics
         let auto_join = Arc::new(HashJoinExec::try_new(
@@ -1600,7 +1600,7 @@ mod test {
                 },
             ],
         };
-        assert_eq!(statistics[0], expected_p0_statistics);
+        assert_eq!(*statistics[0], expected_p0_statistics);
         Ok(())
     }
 }

datafusion/physical-plan/src/execution_plan.rs

@@ -1781,7 +1781,10 @@ mod tests {
             unimplemented!()
         }
 
-        fn partition_statistics(&self, _partition: Option<usize>) -> Result<Statistics> {
+        fn partition_statistics(
+            &self,
+            _partition: Option<usize>,
+        ) -> Result<Arc<Statistics>> {
             unimplemented!()
         }
     }

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

@@ -1441,7 +1441,7 @@ impl ExecutionPlan for HashJoinExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(&self, partition: Option<usize>) -> Result<Statistics> {
+    fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {
         let stats = match (partition, self.mode) {
             // For CollectLeft mode, the left side is collected into a single partition,
             // so all left partitions are available to each output partition.
@@ -1451,8 +1451,8 @@ impl ExecutionPlan for HashJoinExec {
                 let right_stats = self.right.partition_statistics(Some(partition))?;
 
                 estimate_join_statistics(
-                    left_stats,
-                    right_stats,
+                    (*left_stats).clone(),
+                    (*right_stats).clone(),
                     &self.on,
                     &self.join_type,
                     &self.join_schema,
@@ -1466,8 +1466,8 @@ impl ExecutionPlan for HashJoinExec {
                 let right_stats = self.right.partition_statistics(Some(partition))?;
 
                 estimate_join_statistics(
-                    left_stats,
-                    right_stats,
+                    (*left_stats).clone(),
+                    (*right_stats).clone(),
                     &self.on,
                     &self.join_type,
                     &self.join_schema,
@@ -1480,9 +1480,11 @@ impl ExecutionPlan for HashJoinExec {
                 // TODO stats: it is not possible in general to know the output size of joins
                 // There are some special cases though, for example:
                 // - `A LEFT JOIN B ON A.col=B.col` with `COUNT_DISTINCT(B.col)=COUNT(B.col)`
+                let left_stats = self.left.partition_statistics(None)?;
+                let right_stats = self.right.partition_statistics(None)?;
                 estimate_join_statistics(
-                    self.left.partition_statistics(None)?,
-                    self.right.partition_statistics(None)?,
+                    (*left_stats).clone(),
+                    (*right_stats).clone(),
                     &self.on,
                     &self.join_type,
                     &self.join_schema,

docs/source/library-user-guide/upgrading/54.0.0.md

@@ -77,7 +77,7 @@ fn apply_expressions(
 
 **Node whose only expressions are in `output_ordering()` (e.g. a synthetic test node with no owned expression fields):**
 
-````rust,ignore
+```rust,ignore
 fn apply_expressions(
     &self,
     f: &mut dyn FnMut(&dyn PhysicalExpr) -> Result<TreeNodeRecursion>,
@@ -90,6 +90,7 @@ fn apply_expressions(
     }
     Ok(tnr)
 }
+
 ### `ExecutionPlan::partition_statistics` now returns `Arc<Statistics>`
 
 `ExecutionPlan::partition_statistics` now returns `Result<Arc<Statistics>>` instead of `Result<Statistics>`. This avoids cloning `Statistics` when it is shared across multiple consumers.
@@ -100,7 +101,7 @@ fn apply_expressions(
 fn partition_statistics(&self, partition: Option<usize>) -> Result<Statistics> {
     Ok(Statistics::new_unknown(&self.schema()))
 }
-````
+```
 
 **After:**