refactor(proto): destructure leaf and pass-through plans in serde hooks

Exhaustive destructures for `CooperativeExec`, `BufferExec`, `EmptyExec`,
`PlaceholderRowExec`, `ExplainExec` and `ScalarSubqueryExec` on both sides,
plus `ScalarSubqueryLink` on the encode side (its `index` is positional).

`EmptyExec`/`PlaceholderRowExec` now read the serialized partition count from
the `partitions` field instead of via `cache`; the two are kept in sync by
`with_partitions`, so the encoded value is unchanged. Wire format unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: adriangb

datafusion/physical-plan/src/buffer.rs

@@ -305,13 +305,24 @@ impl ExecutionPlan for BufferExec {
         ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
     ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
         use datafusion_proto_models::protobuf;
-        let input = ctx.encode_child(self.input())?;
+        // Destructure exhaustively (no `..`) so that adding a field to
+        // `BufferExec` is a compile error here until it is either serialized or
+        // explicitly documented as not needing to be.
+        let Self {
+            input,
+            // Derived from the input's properties at construction time.
+            properties: _,
+            capacity,
+            // Runtime metrics, not part of the plan shape.
+            metrics: _,
+        } = self;
+        let input = ctx.encode_child(input)?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::Buffer(Box::new(
                     protobuf::BufferExecNode {
                         input: Some(Box::new(input)),
-                        capacity: self.capacity() as u64,
+                        capacity: *capacity as u64,
                     },
                 )),
             ),
@@ -336,9 +347,11 @@ impl BufferExec {
             protobuf::physical_plan_node::PhysicalPlanType::Buffer,
             "BufferExec",
         );
-        let input =
-            ctx.decode_required_child(buffer.input.as_deref(), "BufferExec", "input")?;
-        Ok(Arc::new(BufferExec::new(input, buffer.capacity as usize)))
+        // Destructure exhaustively so that a new field on `BufferExecNode` is a
+        // compile error here rather than a silently dropped field.
+        let protobuf::BufferExecNode { input, capacity } = &**buffer;
+        let input = ctx.decode_required_child(input.as_deref(), "BufferExec", "input")?;
+        Ok(Arc::new(BufferExec::new(input, *capacity as usize)))
     }
 }
 

datafusion/physical-plan/src/coop.rs

@@ -376,7 +376,15 @@ impl ExecutionPlan for CooperativeExec {
         ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
     ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
         use datafusion_proto_models::protobuf;
-        let input = ctx.encode_child(self.input())?;
+        // Destructure exhaustively (no `..`) so that adding a field to
+        // `CooperativeExec` is a compile error here until it is either
+        // serialized or explicitly documented as not needing to be.
+        let Self {
+            input,
+            // Derived from the input's properties at construction time.
+            properties: _,
+        } = self;
+        let input = ctx.encode_child(input)?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::Cooperative(Box::new(
@@ -406,11 +414,12 @@ impl CooperativeExec {
             protobuf::physical_plan_node::PhysicalPlanType::Cooperative,
             "CooperativeExec",
         );
-        let input = ctx.decode_required_child(
-            cooperative.input.as_deref(),
-            "CooperativeExec",
-            "input",
-        )?;
+        // Destructure exhaustively so that a new field on
+        // `CooperativeExecNode` is a compile error here rather than a silently
+        // dropped field.
+        let protobuf::CooperativeExecNode { input } = &**cooperative;
+        let input =
+            ctx.decode_required_child(input.as_deref(), "CooperativeExec", "input")?;
         Ok(Arc::new(CooperativeExec::new(input)))
     }
 }

datafusion/physical-plan/src/empty.rs

@@ -192,16 +192,22 @@ impl ExecutionPlan for EmptyExec {
         _ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
     ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
         use datafusion_proto_models::protobuf;
-        let schema = self.schema().as_ref().try_into()?;
+        // Destructure exhaustively (no `..`) so that adding a field to
+        // `EmptyExec` is a compile error here until it is either serialized or
+        // explicitly documented as not needing to be.
+        let Self {
+            schema,
+            partitions,
+            // Derived from `schema` and `partitions`, recomputed on decode.
+            cache: _,
+        } = self;
+        let schema = schema.as_ref().try_into()?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::Empty(
                     protobuf::EmptyExecNode {
                         schema: Some(schema),
-                        partitions: self
-                            .properties()
-                            .output_partitioning()
-                            .partition_count() as u32,
+                        partitions: *partitions as u32,
                     },
                 ),
             ),
@@ -222,15 +228,18 @@ impl EmptyExec {
             protobuf::physical_plan_node::PhysicalPlanType::Empty,
             "EmptyExec",
         );
-        let schema = empty.schema.as_ref().ok_or_else(|| {
+        // Destructure exhaustively so that a new field on `EmptyExecNode` is a
+        // compile error here rather than a silently dropped field.
+        let protobuf::EmptyExecNode { schema, partitions } = empty;
+        let schema = schema.as_ref().ok_or_else(|| {
             datafusion_common::internal_datafusion_err!(
                 "EmptyExec is missing required field 'schema'"
             )
         })?;
         let schema = Arc::new(arrow::datatypes::Schema::try_from(schema)?);
         // A zero (absent) partition count comes from a plan encoded before the
         // field existed, which always meant a single partition.
-        let partitions = empty.partitions.max(1) as usize;
+        let partitions = (*partitions).max(1) as usize;
         Ok(Arc::new(EmptyExec::new(schema).with_partitions(partitions)))
     }
 }

datafusion/physical-plan/src/explain.rs

@@ -193,17 +193,26 @@ impl ExecutionPlan for ExplainExec {
     ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
         use datafusion_proto_models::protobuf;
 
+        // Destructure exhaustively (no `..`) so that adding a field to
+        // `ExplainExec` is a compile error here until it is either serialized
+        // or explicitly documented as not needing to be.
+        let Self {
+            schema,
+            stringified_plans,
+            verbose,
+            // Derived from `schema`, recomputed on decode.
+            cache: _,
+        } = self;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::Explain(
                     protobuf::ExplainExecNode {
-                        schema: Some(self.schema().as_ref().try_into()?),
-                        stringified_plans: self
-                            .stringified_plans()
+                        schema: Some(schema.as_ref().try_into()?),
+                        stringified_plans: stringified_plans
                             .iter()
                             .map(stringified_plan_to_proto)
                             .collect(),
-                        verbose: self.verbose(),
+                        verbose: *verbose,
                     },
                 ),
             ),
@@ -225,19 +234,25 @@ impl ExplainExec {
             protobuf::physical_plan_node::PhysicalPlanType::Explain,
             "ExplainExec",
         );
-        let schema = explain.schema.as_ref().ok_or_else(|| {
+        // Destructure exhaustively so that a new field on `ExplainExecNode` is
+        // a compile error here rather than a silently dropped field.
+        let protobuf::ExplainExecNode {
+            schema,
+            stringified_plans,
+            verbose,
+        } = explain;
+        let schema = schema.as_ref().ok_or_else(|| {
             datafusion_common::internal_datafusion_err!(
                 "ExplainExec is missing required field 'schema'"
             )
         })?;
         Ok(Arc::new(ExplainExec::new(
             Arc::new(arrow::datatypes::Schema::try_from(schema)?),
-            explain
-                .stringified_plans
+            stringified_plans
                 .iter()
                 .map(stringified_plan_from_proto)
                 .collect(),
-            explain.verbose,
+            *verbose,
         )))
     }
 }

datafusion/physical-plan/src/placeholder_row.rs

@@ -193,16 +193,22 @@ impl ExecutionPlan for PlaceholderRowExec {
         _ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
     ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
         use datafusion_proto_models::protobuf;
-        let schema = self.schema().as_ref().try_into()?;
+        // Destructure exhaustively (no `..`) so that adding a field to
+        // `PlaceholderRowExec` is a compile error here until it is either
+        // serialized or explicitly documented as not needing to be.
+        let Self {
+            schema,
+            partitions,
+            // Derived from `schema` and `partitions`, recomputed on decode.
+            cache: _,
+        } = self;
+        let schema = schema.as_ref().try_into()?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::PlaceholderRow(
                     protobuf::PlaceholderRowExecNode {
                         schema: Some(schema),
-                        partitions: self
-                            .properties()
-                            .output_partitioning()
-                            .partition_count() as u32,
+                        partitions: *partitions as u32,
                     },
                 ),
             ),
@@ -223,15 +229,19 @@ impl PlaceholderRowExec {
             protobuf::physical_plan_node::PhysicalPlanType::PlaceholderRow,
             "PlaceholderRowExec",
         );
-        let schema = placeholder.schema.as_ref().ok_or_else(|| {
+        // Destructure exhaustively so that a new field on
+        // `PlaceholderRowExecNode` is a compile error here rather than a
+        // silently dropped field.
+        let protobuf::PlaceholderRowExecNode { schema, partitions } = placeholder;
+        let schema = schema.as_ref().ok_or_else(|| {
             datafusion_common::internal_datafusion_err!(
                 "PlaceholderRowExec is missing required field 'schema'"
             )
         })?;
         let schema = Arc::new(Schema::try_from(schema)?);
         // A zero (absent) partition count comes from a plan encoded before the
         // field existed, which always meant a single partition.
-        let partitions = placeholder.partitions.max(1) as usize;
+        let partitions = (*partitions).max(1) as usize;
         Ok(Arc::new(
             PlaceholderRowExec::new(schema).with_partitions(partitions),
         ))

datafusion/physical-plan/src/scalar_subquery.rs

@@ -262,10 +262,29 @@ impl ExecutionPlan for ScalarSubqueryExec {
     ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
         use datafusion_proto_models::protobuf;
 
-        let input = ctx.encode_child(self.input())?;
+        // Destructure exhaustively (no `..`) so that adding a field to
+        // `ScalarSubqueryExec` is a compile error here until it is either
+        // serialized or explicitly documented as not needing to be.
+        let Self {
+            input,
+            subqueries,
+            // Execution-time one-shot future, created on `execute()`.
+            subquery_future: _,
+            // Runtime results container, rebuilt (empty) on decode and shared
+            // with the input's `ScalarSubqueryExpr` nodes.
+            results: _,
+            // Copied from the input's properties, recomputed on decode.
+            cache: _,
+        } = self;
+        let input = ctx.encode_child(input)?;
         // Subquery indices are positional and recovered during decoding.
-        let subqueries =
-            ctx.encode_children(self.subqueries().iter().map(|subquery| &subquery.plan))?;
+        let subqueries = ctx.encode_children(subqueries.iter().map(
+            |ScalarSubqueryLink {
+                 plan,
+                 // Positional: recovered from the element's position on decode.
+                 index: _,
+             }| plan,
+        ))?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::ScalarSubquery(Box::new(
@@ -293,17 +312,20 @@ impl ScalarSubqueryExec {
             protobuf::physical_plan_node::PhysicalPlanType::ScalarSubquery,
             "ScalarSubqueryExec",
         );
-        let results = ScalarSubqueryResults::new(scalar_subquery.subqueries.len());
-        let input_node = scalar_subquery.input.as_deref().ok_or_else(|| {
+        // Destructure exhaustively so that a new field on
+        // `ScalarSubqueryExecNode` is a compile error here rather than a
+        // silently dropped field.
+        let protobuf::ScalarSubqueryExecNode { input, subqueries } = &**scalar_subquery;
+        let results = ScalarSubqueryResults::new(subqueries.len());
+        let input_node = input.as_deref().ok_or_else(|| {
             datafusion_common::internal_datafusion_err!(
                 "ScalarSubqueryExec is missing required field 'input'"
             )
         })?;
         // The input's ScalarSubqueryExpr nodes must share this results container.
         let input =
             ctx.decode_child_with_scalar_subquery_results(input_node, results.clone())?;
-        let subqueries = scalar_subquery
-            .subqueries
+        let subqueries = subqueries
             .iter()
             .enumerate()
             .map(|(index, plan)| {