refactor(proto): migrate single-child plans

datafusion/physical-plan/src/buffer.rs

@@ -298,6 +298,48 @@ impl ExecutionPlan for BufferExec {
             Ok(Arc::new(Self::new(new_input, self.capacity)) as Arc<dyn ExecutionPlan>)
         })
     }
+
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
+        use datafusion_proto_models::protobuf;
+        let input = ctx.encode_child(self.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,
+                    },
+                )),
+            ),
+        }))
+    }
+}
+
+#[cfg(feature = "proto")]
+impl BufferExec {
+    /// Reconstruct a [`BufferExec`] from its protobuf representation.
+    ///
+    /// The exact inverse of [`ExecutionPlan::try_to_proto`].
+    ///
+    /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
+        ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        use datafusion_proto_models::protobuf;
+        let buffer = crate::expect_plan_variant!(
+            node,
+            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)))
+    }
 }
 
 /// Represents anything that occupies a capacity in a [MemoryBufferedStream].

datafusion/physical-plan/src/coalesce_batches.rs

@@ -290,6 +290,61 @@ impl ExecutionPlan for CoalesceBatchesExec {
             ) as Arc<dyn ExecutionPlan>)
         })
     }
+
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
+        use datafusion_proto_models::protobuf;
+        let input = ctx.encode_child(self.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),
+                    }),
+                ),
+            ),
+        }))
+    }
+}
+
+#[cfg(feature = "proto")]
+#[expect(deprecated)]
+impl CoalesceBatchesExec {
+    /// Reconstruct a [`CoalesceBatchesExec`] from its protobuf representation.
+    ///
+    /// The exact inverse of [`ExecutionPlan::try_to_proto`]: it takes the whole
+    /// [`PhysicalPlanNode`] so every plan's `try_from_proto` shares one
+    /// signature. The child plan is decoded recursively via the
+    /// [`ExecutionPlanDecodeCtx`].
+    ///
+    /// [`PhysicalPlanNode`]: datafusion_proto_models::protobuf::PhysicalPlanNode
+    /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+    /// [`ExecutionPlanDecodeCtx`]: crate::proto::ExecutionPlanDecodeCtx
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
+        ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        use datafusion_proto_models::protobuf;
+        let coalesce_batches = crate::expect_plan_variant!(
+            node,
+            protobuf::physical_plan_node::PhysicalPlanType::CoalesceBatches,
+            "CoalesceBatchesExec",
+        );
+        let input = ctx.decode_required_child(
+            coalesce_batches.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)),
+        ))
+    }
 }
 
 /// Stream for [`CoalesceBatchesExec`]. See [`CoalesceBatchesExec`] for more details.

datafusion/physical-plan/src/coalesce_partitions.rs

@@ -346,6 +346,56 @@ impl ExecutionPlan for CoalescePartitionsExec {
                 }
             })
     }
+
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
+        use datafusion_proto_models::protobuf;
+        let input = ctx.encode_child(self.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),
+                    },
+                )),
+            ),
+        }))
+    }
+}
+
+#[cfg(feature = "proto")]
+impl CoalescePartitionsExec {
+    /// Reconstruct a [`CoalescePartitionsExec`] from its protobuf representation.
+    ///
+    /// The exact inverse of [`ExecutionPlan::try_to_proto`]. Note the protobuf
+    /// variant is named `Merge` (node [`CoalescePartitionsExecNode`]).
+    ///
+    /// [`CoalescePartitionsExecNode`]: datafusion_proto_models::protobuf::CoalescePartitionsExecNode
+    /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
+        ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        use datafusion_proto_models::protobuf;
+        let merge = crate::expect_plan_variant!(
+            node,
+            protobuf::physical_plan_node::PhysicalPlanType::Merge,
+            "CoalescePartitionsExec",
+        );
+        let input = ctx.decode_required_child(
+            merge.input.as_deref(),
+            "CoalescePartitionsExec",
+            "input",
+        )?;
+        Ok(Arc::new(
+            CoalescePartitionsExec::new(input)
+                .with_fetch(merge.fetch.map(|f| f as usize)),
+        ))
+    }
 }
 
 #[cfg(test)]

datafusion/physical-plan/src/coop.rs

@@ -369,6 +369,50 @@ impl ExecutionPlan for CooperativeExec {
             }
         }
     }
+
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
+        use datafusion_proto_models::protobuf;
+        let input = ctx.encode_child(self.input())?;
+        Ok(Some(protobuf::PhysicalPlanNode {
+            physical_plan_type: Some(
+                protobuf::physical_plan_node::PhysicalPlanType::Cooperative(Box::new(
+                    protobuf::CooperativeExecNode {
+                        input: Some(Box::new(input)),
+                    },
+                )),
+            ),
+        }))
+    }
+}
+
+#[cfg(feature = "proto")]
+impl CooperativeExec {
+    /// Reconstruct a [`CooperativeExec`] from its protobuf representation.
+    ///
+    /// The exact inverse of [`ExecutionPlan::try_to_proto`].
+    ///
+    /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
+        ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        use datafusion_proto_models::protobuf;
+        let cooperative = crate::expect_plan_variant!(
+            node,
+            protobuf::physical_plan_node::PhysicalPlanType::Cooperative,
+            "CooperativeExec",
+        );
+        let input = ctx.decode_required_child(
+            cooperative.input.as_deref(),
+            "CooperativeExec",
+            "input",
+        )?;
+        Ok(Arc::new(CooperativeExec::new(input)))
+    }
 }
 
 /// Creates a [`CooperativeStream`] wrapper around the given [`RecordBatchStream`].