refactor(physical-plan): default child_stats_requests to Skip

asolimando · asolimando · commit e4dcf39c1b2f · 2026-06-23T15:23:30.000+02:00
Make the default `child_stats_requests` skip every child, so a node that does not
derive its statistics from children (for example one that only overrides the
deprecated `partition_statistics`) triggers no child traversal. This avoids the
eager child walk, and the child-stat errors it could surface, that
`StatisticsContext::compute` previously performed even when `statistics_from_inputs`
ignored the results.

Nodes that read `input_stats` now declare the children they use via
`child_stats_requests`.
diff --git a/datafusion-examples/examples/relation_planner/table_sample.rs b/datafusion-examples/examples/relation_planner/table_sample.rs
@@ -108,7 +108,8 @@ use datafusion::{
     },
     physical_expr::EquivalenceProperties,
     physical_plan::{
-        DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, StatisticsArgs,
+        ChildStats, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+        StatisticsArgs,
         metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet, RecordOutput},
     },
     physical_planner::{DefaultPhysicalPlanner, ExtensionPlanner, PhysicalPlanner},
@@ -722,6 +723,10 @@ impl ExecutionPlan for SampleExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-optimizer/src/output_requirements.rs b/datafusion/physical-optimizer/src/output_requirements.rs
@@ -39,8 +39,8 @@ use datafusion_physical_plan::projection::{
 use datafusion_physical_plan::sorts::sort::SortExec;
 use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
 use datafusion_physical_plan::{
-    DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
-    SendableRecordBatchStream, StatisticsArgs,
+    ChildStats, DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties,
+    PlanProperties, SendableRecordBatchStream, StatisticsArgs,
 };
 
 /// This rule either adds or removes [`OutputRequirements`]s to/from the physical
@@ -242,6 +242,10 @@ impl ExecutionPlan for OutputRequirementExec {
         unreachable!();
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -33,7 +33,7 @@ use crate::filter_pushdown::{
     FilterPushdownPropagation, PushedDownPredicate,
 };
 use crate::metrics::{ExecutionPlanMetricsSet, MetricsSet};
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::{
     DisplayFormatType, Distribution, ExecutionPlan, InputOrderMode,
     SendableRecordBatchStream, Statistics, check_if_same_properties,
@@ -1765,6 +1765,10 @@ impl ExecutionPlan for AggregateExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/buffer.rs b/datafusion/physical-plan/src/buffer.rs
@@ -24,7 +24,7 @@ use crate::filter_pushdown::{
     FilterPushdownPropagation,
 };
 use crate::projection::ProjectionExec;
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::stream::RecordBatchStreamAdapter;
 use crate::{
     DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, SortOrderPushdownResult,
@@ -238,6 +238,10 @@ impl ExecutionPlan for BufferExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/coalesce_batches.rs b/datafusion/physical-plan/src/coalesce_batches.rs
@@ -24,7 +24,7 @@ use std::task::{Context, Poll};
 use super::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
 use super::{DisplayAs, ExecutionPlanProperties, PlanProperties, Statistics};
 use crate::projection::ProjectionExec;
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::stream::EmptyRecordBatchStream;
 use crate::{
     DisplayFormatType, ExecutionPlan, RecordBatchStream, SendableRecordBatchStream,
@@ -216,6 +216,10 @@ impl ExecutionPlan for CoalesceBatchesExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/coop.rs b/datafusion/physical-plan/src/coop.rs
@@ -84,7 +84,7 @@ use crate::filter_pushdown::{
     FilterPushdownPropagation,
 };
 use crate::projection::ProjectionExec;
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::{
     DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, RecordBatchStream,
     SendableRecordBatchStream, SortOrderPushdownResult, check_if_same_properties,
@@ -299,6 +299,10 @@ impl ExecutionPlan for CooperativeExec {
         Ok(make_cooperative(child_stream))
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
@@ -538,20 +538,23 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync {
         self.partition_statistics(args.partition())
     }
 
-    /// Returns the partition index to request from each child when computing
-    /// statistics for this node at `partition`.
-    ///
-    /// The returned `Vec` has one entry per child (in the same order as
-    /// [`Self::children`]). [`ChildStats::At`] with `None` requests the child's
-    /// overall (all-partitions) statistics, and [`ChildStats::Skip`] omits a child
-    /// whose statistics this node does not need.
-    ///
-    /// The default requests the same `partition` from every child.
-    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
-        self.children()
-            .iter()
-            .map(|_| ChildStats::At(partition))
-            .collect()
+    /// Returns, per child, which statistics the [`StatisticsContext`] should resolve
+    /// before calling [`Self::statistics_from_inputs`].
+    ///
+    /// One entry per child (same order as [`Self::children`]): [`ChildStats::At`]
+    /// requests the child's statistics at a partition (`None` = overall);
+    /// [`ChildStats::Skip`] omits a child whose statistics this node does not need
+    /// (a `Statistics::new_unknown` placeholder fills its `input_stats` slot).
+    ///
+    /// The default skips every child, so a node that derives nothing from its
+    /// children (for example one that only overrides the deprecated
+    /// [`Self::partition_statistics`]) triggers no child traversal. A node that reads
+    /// `input_stats` in [`Self::statistics_from_inputs`] must override this to declare
+    /// the children it uses.
+    ///
+    /// [`StatisticsContext`]: crate::statistics::StatisticsContext
+    fn child_stats_requests(&self, _partition: Option<usize>) -> Vec<ChildStats> {
+        self.children().iter().map(|_| ChildStats::Skip).collect()
     }
 
     /// Returns `true` if a limit can be safely pushed down through this
diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs
@@ -42,7 +42,7 @@ use crate::projection::{
     EmbeddedProjection, ProjectionExec, ProjectionExpr, make_with_child,
     try_embed_projection, update_expr,
 };
-use crate::statistics::{StatisticsArgs, StatisticsContext};
+use crate::statistics::{ChildStats, StatisticsArgs, StatisticsContext};
 use crate::stream::EmptyRecordBatchStream;
 use crate::{
     DisplayFormatType, ExecutionPlan,
@@ -592,6 +592,10 @@ impl ExecutionPlan for FilterExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     /// The output statistics of a filtering operation can be estimated if the
     /// predicate's selectivity value can be determined for the incoming data.
     fn statistics_from_inputs(
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
@@ -38,7 +38,7 @@ use crate::projection::{
     physical_to_column_exprs, update_join_on,
 };
 use crate::spill::spill_manager::SpillManager;
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::{
     DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, ExecutionPlanProperties,
     PlanProperties, SendableRecordBatchStream, Statistics, check_if_same_properties,
@@ -564,6 +564,10 @@ impl ExecutionPlan for SortMergeJoinExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition), ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/limit.rs b/datafusion/physical-plan/src/limit.rs
@@ -27,7 +27,7 @@ use super::{
     SendableRecordBatchStream, Statistics,
 };
 use crate::execution_plan::{Boundedness, CardinalityEffect};
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::{
     DisplayFormatType, Distribution, ExecutionPlan, Partitioning,
     check_if_same_properties,
@@ -220,6 +220,10 @@ impl ExecutionPlan for GlobalLimitExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
@@ -387,6 +391,10 @@ impl ExecutionPlan for LocalLimitExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
@@ -33,7 +33,7 @@ use crate::filter_pushdown::{
     FilterPushdownPropagation, FilterRemapper, PushedDownPredicate,
 };
 use crate::joins::utils::{ColumnIndex, JoinFilter, JoinOn, JoinOnRef};
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::{DisplayFormatType, ExecutionPlan, PhysicalExpr, check_if_same_properties};
 use std::collections::HashMap;
 use std::pin::Pin;
@@ -349,6 +349,10 @@ impl ExecutionPlan for ProjectionExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/scalar_subquery.rs b/datafusion/physical-plan/src/scalar_subquery.rs
@@ -33,7 +33,7 @@ use datafusion_expr::execution_props::{ScalarSubqueryResults, SubqueryIndex};
 
 use crate::execution_plan::{CardinalityEffect, ExecutionPlan, PlanProperties};
 use crate::joins::utils::{OnceAsync, OnceFut};
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::stream::RecordBatchStreamAdapter;
 use crate::{DisplayAs, DisplayFormatType, SendableRecordBatchStream};
 
@@ -236,6 +236,13 @@ impl ExecutionPlan for ScalarSubqueryExec {
         vec![false; self.subqueries.len() + 1]
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        // Only `self.input` (child 0) is used; the subqueries are skipped.
+        let mut requests = vec![ChildStats::Skip; 1 + self.subqueries.len()];
+        requests[0] = ChildStats::At(partition);
+        requests
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/sorts/partial_sort.rs b/datafusion/physical-plan/src/sorts/partial_sort.rs
@@ -58,7 +58,7 @@ use std::task::{Context, Poll};
 
 use crate::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
 use crate::sorts::sort::sort_batch;
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::stream::EmptyRecordBatchStream;
 use crate::{
     DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, ExecutionPlanProperties,
@@ -365,6 +365,10 @@ impl ExecutionPlan for PartialSortExec {
         Some(self.metrics_set.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs
@@ -676,6 +676,10 @@ impl ExecutionPlan for InterleaveExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition); self.inputs.len()]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs
@@ -28,7 +28,7 @@ use std::task::{Context, Poll};
 
 use super::utils::create_schema;
 use crate::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::stream::EmptyRecordBatchStream;
 use crate::windows::{
     calc_requirements, get_ordered_partition_by_indices, get_partition_by_sort_exprs,
@@ -378,6 +378,10 @@ impl ExecutionPlan for BoundedWindowAggExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/datafusion/physical-plan/src/windows/window_agg_exec.rs b/datafusion/physical-plan/src/windows/window_agg_exec.rs
@@ -24,7 +24,7 @@ use std::task::{Context, Poll};
 use super::utils::create_schema;
 use crate::execution_plan::{CardinalityEffect, EmissionType};
 use crate::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
-use crate::statistics::StatisticsArgs;
+use crate::statistics::{ChildStats, StatisticsArgs};
 use crate::stream::EmptyRecordBatchStream;
 use crate::windows::{
     calc_requirements, get_ordered_partition_by_indices, get_partition_by_sort_exprs,
@@ -281,6 +281,10 @@ impl ExecutionPlan for WindowAggExec {
         Some(self.metrics.clone_inner())
     }
 
+    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+        vec![ChildStats::At(partition)]
+    }
+
     fn statistics_from_inputs(
         &self,
         input_stats: &[Arc<Statistics>],
diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md
@@ -185,9 +185,11 @@ migration is required until the deprecated method is removed.
 **Migration guide:**
 
 For **implementations**, override `statistics_from_inputs` instead of
-`partition_statistics`. Child statistics arrive pre-computed in `input_stats`
-(one entry per child, in `children()` order); the node only expresses its local
-propagation logic. Leaf nodes that have no children can ignore `input_stats`.
+`partition_statistics`, plus `child_stats_requests` to declare which children to
+resolve. Child statistics then arrive pre-computed in `input_stats` (one entry per
+child, in `children()` order), so the node only expresses its local propagation
+logic. Leaf nodes, and nodes that derive their statistics without reading children,
+need neither override (the default `child_stats_requests` skips every child).
 
 ```rust,ignore
 // Before:
@@ -196,7 +198,11 @@ fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistic
     // ... transform child_stats ...
 }
 
-// After:
+// After: declare the child to resolve, then compute from its statistics.
+fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+    vec![ChildStats::At(partition)]
+}
+
 fn statistics_from_inputs(
     &self,
     input_stats: &[Arc<Statistics>],
@@ -207,17 +213,12 @@ fn statistics_from_inputs(
 }
 ```
 
-> **Note:** the default `child_stats_requests` requests each child at the
-> parent's partition. A partition-merging or multi-child node must override it,
-> or it will silently receive child statistics for the wrong partition:
->
-> ```rust,ignore
-> fn child_stats_requests(&self, _partition: Option<usize>) -> Vec<ChildStats> {
->     vec![ChildStats::At(None)] // request the child's overall stats
-> }
-> ```
->
-> See the `child_stats_requests` rustdoc for `ChildStats::Skip` and multi-child cases.
+> **Important:** the default `child_stats_requests` skips every child, so a node that
+> reads `input_stats` must override it to declare the children it uses, or those slots
+> are filled with `Statistics::new_unknown` placeholders. Request a child with
+> `ChildStats::At(partition)` (`None` = overall) and omit one with `ChildStats::Skip`.
+> For example, a partition-merging operator requests `ChildStats::At(None)`, and a
+> broadcast join requests its build side at `None`.
 
 For **callers**, walk a plan through `StatisticsContext::compute`. The cache is
 created with the context:
