refactor(proto): destructure repartition, union and coalesce plans in serde hooks

Exhaustive destructures for `RepartitionExec`, `UnionExec`, `InterleaveExec`,
`CoalesceBatchesExec` and `CoalescePartitionsExec` on both sides.

Note `RepartitionExec` keeps its output partitioning inside `cache`, so `cache`
is bound (not `_`) and read for the serialized `partitioning` field. Wire
format unchanged.

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

Author: adriangb

datafusion/physical-plan/src/coalesce_batches.rs

@@ -297,14 +297,26 @@ impl ExecutionPlan for CoalesceBatchesExec {
         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
+        // `CoalesceBatchesExec` is a compile error here until it is either
+        // serialized or explicitly documented as not needing to be.
+        let Self {
+            input,
+            target_batch_size,
+            fetch,
+            // Runtime metrics, not part of the plan shape.
+            metrics: _,
+            // Derived plan properties, recomputed on decode.
+            cache: _,
+        } = self;
+        let input = ctx.encode_child(input)?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::CoalesceBatches(
                     Box::new(protobuf::CoalesceBatchesExecNode {
                         input: Some(Box::new(input)),
-                        target_batch_size: self.target_batch_size() as u32,
-                        fetch: self.fetch().map(|n| n as u32),
+                        target_batch_size: *target_batch_size as u32,
+                        fetch: fetch.map(|n| n as u32),
                     }),
                 ),
             ),
@@ -335,14 +347,19 @@ impl CoalesceBatchesExec {
             protobuf::physical_plan_node::PhysicalPlanType::CoalesceBatches,
             "CoalesceBatchesExec",
         );
-        let input = ctx.decode_required_child(
-            coalesce_batches.input.as_deref(),
-            "CoalesceBatchesExec",
-            "input",
-        )?;
+        // Destructure exhaustively so that a new field on
+        // `CoalesceBatchesExecNode` is a compile error here rather than a
+        // silently dropped field.
+        let protobuf::CoalesceBatchesExecNode {
+            input,
+            target_batch_size,
+            fetch,
+        } = &**coalesce_batches;
+        let input =
+            ctx.decode_required_child(input.as_deref(), "CoalesceBatchesExec", "input")?;
         Ok(Arc::new(
-            CoalesceBatchesExec::new(input, coalesce_batches.target_batch_size as usize)
-                .with_fetch(coalesce_batches.fetch.map(|f| f as usize)),
+            CoalesceBatchesExec::new(input, *target_batch_size as usize)
+                .with_fetch(fetch.map(|f| f as usize)),
         ))
     }
 }

datafusion/physical-plan/src/coalesce_partitions.rs

@@ -353,13 +353,24 @@ impl ExecutionPlan for CoalescePartitionsExec {
         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
+        // `CoalescePartitionsExec` is a compile error here until it is either
+        // serialized or explicitly documented as not needing to be.
+        let Self {
+            input,
+            // Runtime metrics, not part of the plan shape.
+            metrics: _,
+            // Derived plan properties, recomputed on decode.
+            cache: _,
+            fetch,
+        } = self;
+        let input = ctx.encode_child(input)?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::Merge(Box::new(
                     protobuf::CoalescePartitionsExecNode {
                         input: Some(Box::new(input)),
-                        fetch: self.fetch().map(|f| f as u32),
+                        fetch: fetch.map(|f| f as u32),
                     },
                 )),
             ),
@@ -386,14 +397,17 @@ impl CoalescePartitionsExec {
             protobuf::physical_plan_node::PhysicalPlanType::Merge,
             "CoalescePartitionsExec",
         );
+        // Destructure exhaustively so that a new field on
+        // `CoalescePartitionsExecNode` is a compile error here rather than a
+        // silently dropped field.
+        let protobuf::CoalescePartitionsExecNode { input, fetch } = &**merge;
         let input = ctx.decode_required_child(
-            merge.input.as_deref(),
+            input.as_deref(),
             "CoalescePartitionsExec",
             "input",
         )?;
         Ok(Arc::new(
-            CoalescePartitionsExec::new(input)
-                .with_fetch(merge.fetch.map(|f| f as usize)),
+            CoalescePartitionsExec::new(input).with_fetch(fetch.map(|f| f as usize)),
         ))
     }
 }

datafusion/physical-plan/src/repartition/mod.rs

@@ -1711,17 +1711,33 @@ impl ExecutionPlan for RepartitionExec {
     ) -> 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
+        // `RepartitionExec` is a compile error here until it is either
+        // serialized or explicitly documented as not needing to be.
+        let Self {
+            input,
+            // Execution-time channel state, created on `execute()`.
+            state: _,
+            // Runtime metrics, not part of the plan shape.
+            metrics: _,
+            preserve_order,
+            // Derived plan properties. The output partitioning lives here (it is
+            // the plan's own `partitioning`) and *is* serialized below; the rest
+            // is recomputed on decode.
+            cache,
+        } = self;
 
-        let partitioning = self.partitioning().try_to_proto(&ctx.expr_ctx())?;
+        let input = ctx.encode_child(input)?;
+
+        let partitioning = cache.partitioning.try_to_proto(&ctx.expr_ctx())?;
 
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::Repartition(Box::new(
                     protobuf::RepartitionExecNode {
                         input: Some(Box::new(input)),
                         partitioning: Some(partitioning),
-                        preserve_order: self.preserve_order(),
+                        preserve_order: *preserve_order,
                     },
                 )),
             ),
@@ -1743,15 +1759,19 @@ impl RepartitionExec {
             protobuf::physical_plan_node::PhysicalPlanType::Repartition,
             "RepartitionExec",
         );
-        let input = ctx.decode_required_child(
-            repart.input.as_deref(),
-            "RepartitionExec",
-            "input",
-        )?;
+        // Destructure exhaustively so that a new field on
+        // `RepartitionExecNode` is a compile error here rather than a silently
+        // dropped field.
+        let protobuf::RepartitionExecNode {
+            input,
+            partitioning,
+            preserve_order,
+        } = &**repart;
+        let input =
+            ctx.decode_required_child(input.as_deref(), "RepartitionExec", "input")?;
         let input_schema = input.schema();
 
-        let partitioning = repart
-            .partitioning
+        let partitioning = partitioning
             .as_ref()
             .map(|partitioning| {
                 Partitioning::try_from_proto(
@@ -1768,7 +1788,7 @@ impl RepartitionExec {
             })?;
 
         let mut repart_exec = RepartitionExec::try_new(input, partitioning)?;
-        if repart.preserve_order {
+        if *preserve_order {
             repart_exec = repart_exec.with_preserve_order();
         }
         Ok(Arc::new(repart_exec))

datafusion/physical-plan/src/union.rs

@@ -558,7 +558,17 @@ impl ExecutionPlan for UnionExec {
         ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
     ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
         use datafusion_proto_models::protobuf;
-        let inputs = ctx.encode_children(self.inputs())?;
+        // Destructure exhaustively (no `..`) so that adding a field to
+        // `UnionExec` is a compile error here until it is either serialized or
+        // explicitly documented as not needing to be.
+        let Self {
+            inputs,
+            // Runtime metrics, not part of the plan shape.
+            metrics: _,
+            // Derived plan properties, recomputed on decode.
+            cache: _,
+        } = self;
+        let inputs = ctx.encode_children(inputs)?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::Union(
@@ -581,8 +591,10 @@ impl UnionExec {
             protobuf::physical_plan_node::PhysicalPlanType::Union,
             "UnionExec",
         );
-        let inputs = union
-            .inputs
+        // Destructure exhaustively so that a new field on `UnionExecNode` is a
+        // compile error here rather than a silently dropped field.
+        let protobuf::UnionExecNode { inputs } = union;
+        let inputs = inputs
             .iter()
             .map(|input| ctx.decode_child(input))
             .collect::<Result<Vec<_>>>()?;
@@ -805,7 +817,17 @@ impl ExecutionPlan for InterleaveExec {
         ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
     ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
         use datafusion_proto_models::protobuf;
-        let inputs = ctx.encode_children(self.inputs())?;
+        // Destructure exhaustively (no `..`) so that adding a field to
+        // `InterleaveExec` is a compile error here until it is either
+        // serialized or explicitly documented as not needing to be.
+        let Self {
+            inputs,
+            // Runtime metrics, not part of the plan shape.
+            metrics: _,
+            // Derived plan properties, recomputed on decode.
+            cache: _,
+        } = self;
+        let inputs = ctx.encode_children(inputs)?;
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(
                 protobuf::physical_plan_node::PhysicalPlanType::Interleave(
@@ -828,8 +850,10 @@ impl InterleaveExec {
             protobuf::physical_plan_node::PhysicalPlanType::Interleave,
             "InterleaveExec",
         );
-        let inputs = interleave
-            .inputs
+        // Destructure exhaustively so that a new field on `InterleaveExecNode`
+        // is a compile error here rather than a silently dropped field.
+        let protobuf::InterleaveExecNode { inputs } = interleave;
+        let inputs = inputs
             .iter()
             .map(|input| ctx.decode_child(input))
             .collect::<Result<Vec<_>>>()?;