Deprecate partition_statistics, add partition_statistics_with_context

asolimando · asolimando · commit b3808936f1ce · 2026-05-03T19:31:59.000+02:00
Non-breaking change per API health policy: existing impls continue
to work via default delegation. Fixes missed ScalarSubqueryExec.
diff --git a/datafusion-examples/examples/relation_planner/table_sample.rs b/datafusion-examples/examples/relation_planner/table_sample.rs
@@ -722,7 +722,7 @@ impl ExecutionPlan for SampleExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/core/tests/custom_sources_cases/mod.rs b/datafusion/core/tests/custom_sources_cases/mod.rs
@@ -180,7 +180,7 @@ impl ExecutionPlan for CustomExecutionPlan {
         Ok(Box::pin(TestCustomRecordBatchStream { nb_batch: 1 }))
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/core/tests/custom_sources_cases/statistics.rs b/datafusion/core/tests/custom_sources_cases/statistics.rs
@@ -176,7 +176,7 @@ impl ExecutionPlan for StatisticsValidation {
         unimplemented!("This plan only serves for testing statistics")
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs
@@ -1140,7 +1140,7 @@ impl ExecutionPlan for StatisticsExec {
         unimplemented!("This plan only serves for testing statistics")
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/core/tests/physical_optimizer/test_utils.rs b/datafusion/core/tests/physical_optimizer/test_utils.rs
@@ -983,7 +983,7 @@ impl ExecutionPlan for TestScan {
         internal_err!("TestScan is for testing optimizer only, not for execution")
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/datasource/src/source.rs b/datafusion/datasource/src/source.rs
@@ -452,7 +452,7 @@ impl ExecutionPlan for DataSourceExec {
         Some(metrics)
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/physical-optimizer/src/output_requirements.rs b/datafusion/physical-optimizer/src/output_requirements.rs
@@ -243,7 +243,7 @@ impl ExecutionPlan for OutputRequirementExec {
         unreachable!();
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -1556,7 +1556,7 @@ impl ExecutionPlan for AggregateExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
@@ -2709,7 +2709,7 @@ mod tests {
             Ok(Box::pin(stream))
         }
 
-        fn partition_statistics(
+        fn partition_statistics_with_context(
             &self,
             partition: Option<usize>,
             _ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/buffer.rs b/datafusion/physical-plan/src/buffer.rs
@@ -245,7 +245,7 @@ impl ExecutionPlan for BufferExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/coalesce_batches.rs b/datafusion/physical-plan/src/coalesce_batches.rs
@@ -223,7 +223,7 @@ impl ExecutionPlan for CoalesceBatchesExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/coalesce_partitions.rs b/datafusion/physical-plan/src/coalesce_partitions.rs
@@ -240,7 +240,7 @@ impl ExecutionPlan for CoalescePartitionsExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/coop.rs b/datafusion/physical-plan/src/coop.rs
@@ -307,7 +307,7 @@ impl ExecutionPlan for CooperativeExec {
         Ok(make_cooperative(child_stream))
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/display.rs b/datafusion/physical-plan/src/display.rs
@@ -1232,7 +1232,7 @@ mod tests {
             todo!()
         }
 
-        fn partition_statistics(
+        fn partition_statistics_with_context(
             &self,
             partition: Option<usize>,
             _ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/empty.rs b/datafusion/physical-plan/src/empty.rs
@@ -160,7 +160,7 @@ impl ExecutionPlan for EmptyExec {
         )?))
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
@@ -546,6 +546,29 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync {
         None
     }
 
+    /// Returns statistics for a specific partition of this `ExecutionPlan` node.
+    ///
+    /// Deprecated: use [`Self::partition_statistics_with_context`] instead,
+    /// which accepts a [`StatisticsContext`] carrying pre-computed child
+    /// statistics.
+    #[deprecated(
+        since = "54.0.0",
+        note = "Use partition_statistics_with_context instead"
+    )]
+    fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {
+        if let Some(idx) = partition {
+            // Validate partition index
+            let partition_count = self.properties().partitioning.partition_count();
+            assert_or_internal_err!(
+                idx < partition_count,
+                "Invalid partition index: {}, the partition count is {}",
+                idx,
+                partition_count
+            );
+        }
+        Ok(Arc::new(Statistics::new_unknown(&self.schema())))
+    }
+
     /// Returns statistics for a specific partition of this `ExecutionPlan` node.
     /// If statistics are not available, should return [`Statistics::new_unknown`]
     /// (the default), not an error.
@@ -558,22 +581,13 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync {
     ///
     /// [`StatisticsContext`]: crate::statistics_context::StatisticsContext
     /// [`compute_statistics`]: crate::statistics_context::compute_statistics
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         _ctx: &StatisticsContext,
     ) -> Result<Arc<Statistics>> {
-        if let Some(idx) = partition {
-            // Validate partition index
-            let partition_count = self.properties().partitioning.partition_count();
-            assert_or_internal_err!(
-                idx < partition_count,
-                "Invalid partition index: {}, the partition count is {}",
-                idx,
-                partition_count
-            );
-        }
-        Ok(Arc::new(Statistics::new_unknown(&self.schema())))
+        #[expect(deprecated)]
+        self.partition_statistics(partition)
     }
 
     /// Returns `true` if a limit can be safely pushed down through this
@@ -1668,7 +1682,7 @@ mod tests {
             unimplemented!()
         }
 
-        fn partition_statistics(
+        fn partition_statistics_with_context(
             &self,
             _partition: Option<usize>,
             _ctx: &StatisticsContext,
@@ -1738,7 +1752,7 @@ mod tests {
             unimplemented!()
         }
 
-        fn partition_statistics(
+        fn partition_statistics_with_context(
             &self,
             _partition: Option<usize>,
             _ctx: &StatisticsContext,
@@ -1803,7 +1817,7 @@ mod tests {
             unimplemented!()
         }
 
-        fn partition_statistics(
+        fn partition_statistics_with_context(
             &self,
             _partition: Option<usize>,
             _ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs
@@ -577,7 +577,7 @@ impl ExecutionPlan for FilterExec {
 
     /// The output statistics of a filtering operation can be estimated if the
     /// predicate's selectivity value can be determined for the incoming data.
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs
@@ -381,7 +381,7 @@ impl ExecutionPlan for CrossJoinExec {
         }
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/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(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs
@@ -688,7 +688,7 @@ impl ExecutionPlan for NestedLoopJoinExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
@@ -582,7 +582,7 @@ impl ExecutionPlan for SortMergeJoinExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/limit.rs b/datafusion/physical-plan/src/limit.rs
@@ -235,7 +235,7 @@ impl ExecutionPlan for GlobalLimitExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
@@ -416,7 +416,7 @@ impl ExecutionPlan for LocalLimitExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs
@@ -1139,7 +1139,7 @@ mod tests {
             unimplemented!()
         }
 
-        fn partition_statistics(
+        fn partition_statistics_with_context(
             &self,
             _partition: Option<usize>,
             _ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/placeholder_row.rs b/datafusion/physical-plan/src/placeholder_row.rs
@@ -174,7 +174,7 @@ impl ExecutionPlan for PlaceholderRowExec {
         Ok(Box::pin(cooperative(ms)))
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
@@ -360,7 +360,7 @@ impl ExecutionPlan for ProjectionExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs
@@ -1185,7 +1185,7 @@ impl ExecutionPlan for RepartitionExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/scalar_subquery.rs b/datafusion/physical-plan/src/scalar_subquery.rs
@@ -35,6 +35,7 @@ use datafusion_physical_expr::PhysicalExpr;
 
 use crate::execution_plan::{CardinalityEffect, ExecutionPlan, PlanProperties};
 use crate::joins::utils::{OnceAsync, OnceFut};
+use crate::statistics_context::StatisticsContext;
 use crate::stream::RecordBatchStreamAdapter;
 use crate::{DisplayAs, DisplayFormatType, SendableRecordBatchStream};
 
@@ -244,8 +245,13 @@ impl ExecutionPlan for ScalarSubqueryExec {
         vec![false; self.subqueries.len() + 1]
     }
 
-    fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {
-        self.input.partition_statistics(partition)
+    fn partition_statistics_with_context(
+        &self,
+        _partition: Option<usize>,
+        ctx: &StatisticsContext,
+    ) -> Result<Arc<Statistics>> {
+        // Passthrough: forward child stats
+        Ok(Arc::clone(&ctx.child_stats()[0]))
     }
 
     fn cardinality_effect(&self) -> CardinalityEffect {
diff --git a/datafusion/physical-plan/src/sorts/partial_sort.rs b/datafusion/physical-plan/src/sorts/partial_sort.rs
@@ -346,7 +346,7 @@ impl ExecutionPlan for PartialSortExec {
         Some(self.metrics_set.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs
@@ -1277,7 +1277,7 @@ impl ExecutionPlan for SortExec {
         Some(self.metrics_set.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs
@@ -394,7 +394,7 @@ impl ExecutionPlan for SortPreservingMergeExec {
         Some(self.metrics.clone_inner())
     }
 
-    fn partition_statistics(
+    fn partition_statistics_with_context(
         &self,
         _partition: Option<usize>,
         _ctx: &StatisticsContext,
diff --git a/datafusion/physical-plan/src/statistics_context.rs b/datafusion/physical-plan/src/statistics_context.rs
diff --git a/datafusion/physical-plan/src/test.rs b/datafusion/physical-plan/src/test.rs
diff --git a/datafusion/physical-plan/src/test/exec.rs b/datafusion/physical-plan/src/test/exec.rs
diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs
diff --git a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs
diff --git a/datafusion/physical-plan/src/windows/window_agg_exec.rs b/datafusion/physical-plan/src/windows/window_agg_exec.rs
diff --git a/datafusion/physical-plan/src/work_table.rs b/datafusion/physical-plan/src/work_table.rs
diff --git a/docs/source/library-user-guide/upgrading/54.0.0.md b/docs/source/library-user-guide/upgrading/54.0.0.md

Original file line number	Diff line number	Diff line change
`@@ -722,7 +722,7 @@ impl ExecutionPlan for SampleExec {`
`722`	`722`	`Some(self.metrics.clone_inner())`
`723`	`723`	`}`
`724`	`724`
`725`		`- fn partition_statistics(`
	`725`	`+ fn partition_statistics_with_context(`
`726`	`726`	`&self,`
`727`	`727`	`_partition: Option<usize>,`
`728`	`728`	`ctx: &StatisticsContext,`
Original file line number	Diff line number	Diff line change
`@@ -180,7 +180,7 @@ impl ExecutionPlan for CustomExecutionPlan {`
`180`	`180`	`Ok(Box::pin(TestCustomRecordBatchStream { nb_batch: 1 }))`
`181`	`181`	`}`
`182`	`182`
`183`		`- fn partition_statistics(`
	`183`	`+ fn partition_statistics_with_context(`
`184`	`184`	`&self,`
`185`	`185`	`partition: Option<usize>,`
`186`	`186`	`_ctx: &StatisticsContext,`
Original file line number	Diff line number	Diff line change
`@@ -176,7 +176,7 @@ impl ExecutionPlan for StatisticsValidation {`
`176`	`176`	`unimplemented!("This plan only serves for testing statistics")`
`177`	`177`	`}`
`178`	`178`
`179`		`- fn partition_statistics(`
	`179`	`+ fn partition_statistics_with_context(`
`180`	`180`	`&self,`
`181`	`181`	`partition: Option<usize>,`
`182`	`182`	`_ctx: &StatisticsContext,`
Original file line number	Diff line number	Diff line change
`@@ -1140,7 +1140,7 @@ impl ExecutionPlan for StatisticsExec {`
`1140`	`1140`	`unimplemented!("This plan only serves for testing statistics")`
`1141`	`1141`	`}`
`1142`	`1142`
`1143`		`- fn partition_statistics(`
	`1143`	`+ fn partition_statistics_with_context(`
`1144`	`1144`	`&self,`
`1145`	`1145`	`partition: Option<usize>,`
`1146`	`1146`	`_ctx: &StatisticsContext,`
Original file line number	Diff line number	Diff line change
`@@ -983,7 +983,7 @@ impl ExecutionPlan for TestScan {`
`983`	`983`	`internal_err!("TestScan is for testing optimizer only, not for execution")`
`984`	`984`	`}`
`985`	`985`
`986`		`- fn partition_statistics(`
	`986`	`+ fn partition_statistics_with_context(`
`987`	`987`	`&self,`
`988`	`988`	`_partition: Option<usize>,`
`989`	`989`	`_ctx: &StatisticsContext,`
Original file line number	Diff line number	Diff line change
`@@ -452,7 +452,7 @@ impl ExecutionPlan for DataSourceExec {`
`452`	`452`	`Some(metrics)`
`453`	`453`	`}`
`454`	`454`
`455`		`- fn partition_statistics(`
	`455`	`+ fn partition_statistics_with_context(`
`456`	`456`	`&self,`
`457`	`457`	`partition: Option<usize>,`
`458`	`458`	`_ctx: &StatisticsContext,`
Original file line number	Diff line number	Diff line change
`@@ -243,7 +243,7 @@ impl ExecutionPlan for OutputRequirementExec {`
`243`	`243`	`unreachable!();`
`244`	`244`	`}`
`245`	`245`
`246`		`- fn partition_statistics(`
	`246`	`+ fn partition_statistics_with_context(`
`247`	`247`	`&self,`
`248`	`248`	`_partition: Option<usize>,`
`249`	`249`	`ctx: &StatisticsContext,`
Original file line number	Diff line number	Diff line change
`@@ -245,7 +245,7 @@ impl ExecutionPlan for BufferExec {`
`245`	`245`	`Some(self.metrics.clone_inner())`
`246`	`246`	`}`
`247`	`247`
`248`		`- fn partition_statistics(`
	`248`	`+ fn partition_statistics_with_context(`
`249`	`249`	`&self,`
`250`	`250`	`_partition: Option<usize>,`
`251`	`251`	`ctx: &StatisticsContext,`
Original file line number	Diff line number	Diff line change
`@@ -223,7 +223,7 @@ impl ExecutionPlan for CoalesceBatchesExec {`
`223`	`223`	`Some(self.metrics.clone_inner())`
`224`	`224`	`}`
`225`	`225`
`226`		`- fn partition_statistics(`
	`226`	`+ fn partition_statistics_with_context(`
`227`	`227`	`&self,`
`228`	`228`	`_partition: Option<usize>,`
`229`	`229`	`ctx: &StatisticsContext,`