refactor(proto): destructure plan and proto structs in core plan serde hooks

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/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/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,
         )))
     }
 }