(datafusion) branch main updated: fix(proto): honor ExecutionPlan downcast_delegate during serialization (#23154)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git


The following commit(s) were added to refs/heads/main by this push:
     new 96ff37acbb fix(proto): honor ExecutionPlan downcast_delegate during serialization (#23154)
96ff37acbb is described below

commit 96ff37acbb7f9238b2e44892e9e489457cdb8d40
Author: Geoffrey Claude <ge...@datadoghq.com>
AuthorDate: Wed Jun 24 13:11:01 2026 +0200

    fix(proto): honor ExecutionPlan downcast_delegate during serialization (#23154)
    
    ## Which issue does this PR close?
    
    N/A. This is a follow-up to #22559 and #21263, but there is no dedicated
    issue for this specific bug.
    
    ## Rationale for this change
    
    #21263 removed `ExecutionPlan::as_any()` now that `ExecutionPlan` has
    `Any` as a supertrait. Most call sites were migrated from code like
    this:
    
    ```rust
    plan.as_any().downcast_ref::<SomeExec>()
    ```
    
    to the new helper:
    
    ```rust
    plan.downcast_ref::<SomeExec>()
    ```
    
    That distinction matters after #22559. `ExecutionPlan::downcast_ref` is
    now the public downcast operation for execution plans: it follows
    `ExecutionPlan::downcast_delegate()` before falling back to raw `Any`.
    This lets wrapper plans keep their own concrete type private while still
    presenting the wrapped plan's normal public identity.
    
    For example, tracing or instrumentation wrappers can implement
    `downcast_delegate()` so callers that ask "is this a `FilterExec` /
    `EmptyExec` / `ProjectionExec`?" get the same answer they would have
    received without the wrapper. The wrapper remains visible only to code
    that intentionally performs a raw concrete-type `Any` check.
    
    This problem came up while trying to instrument distributed plans
    through
    [`datafusion-distributed`](https://github.com/datafusion-contrib/datafusion-distributed).
    In that setting, physical plans may be wrapped for
    tracing/instrumentation and then serialized for distributed execution.
    The wrapper is meant to be transparent for normal plan inspection, but
    serialization still needs to recognize the delegated built-in plan
    underneath it.
    
    The physical plan proto serializer still had one leftover mechanical
    migration from the old `as_any()` world:
    
    ```rust
    let plan = plan.as_ref() as &dyn Any;
    ```
    
    That line bypasses `ExecutionPlan::downcast_ref` entirely. As a result,
    a delegating wrapper around a built-in execution plan is not serialized
    as the built-in plan. Instead, the serializer sees only the wrapper's
    concrete type, fails all built-in `Exec` checks, and falls through to
    the extension codec. With the default extension codec this produces an
    unsupported-plan error, even though the wrapped plan itself is
    serializable.
    
    This is particularly relevant for transparent wrappers such as
    [`InstrumentedExec`](https://github.com/datafusion-contrib/datafusion-tracing/blob/main/datafusion-tracing/src/instrumented_exec.rs)
    in
    [`datafusion-contrib/datafusion-tracing`](https://github.com/datafusion-contrib/datafusion-tracing),
    which intentionally delegate public downcasts to the wrapped plan.
    
    ## What changes are included in this PR?
    
    - Changes physical plan proto serialization to bind the plan as `&dyn
    ExecutionPlan` instead of `&dyn Any`.
    - Leaves the existing serializer downcast chain intact, so each
    `plan.downcast_ref::<...>()` now dispatches through the `ExecutionPlan`
    helper and therefore honors `downcast_delegate()`.
    - Adds a regression test with a small wrapper execution plan that
    delegates public downcasts to an inner `EmptyExec`.
    - Audited other `as_ref() as &dyn Any` patterns. The remaining hits are
    either non-`ExecutionPlan` traits, `PhysicalExpr` checks, UDF tests, or
    the intentional raw `Any` fallback inside the `ExecutionPlan` helper
    itself.
    
    ## Are these changes tested?
    
    Yes.
    
    ```bash
    cargo fmt --all
    cargo fmt --all --check
    cargo test -p datafusion-proto --test proto_integration serialize_uses_downcast_delegate
    cargo clippy --all-targets --all-features -- -D warnings
    ```
    
    ## Are there any user-facing changes?
    
    Yes, as a bug fix. Physical plan proto serialization now honors the
    documented `ExecutionPlan::downcast_delegate()` behavior. Transparent
    wrapper plans can serialize as their delegated built-in execution plan
    when appropriate instead of requiring an extension codec for the wrapper
    itself.
---
 datafusion/proto/src/physical_plan/mod.rs          |  2 +-
 .../proto/tests/cases/roundtrip_physical_plan.rs   | 72 +++++++++++++++++++++-
 2 files changed, 71 insertions(+), 3 deletions(-)

diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs
index 9efcd25fcb..dcbb6f761e 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -442,7 +442,7 @@ pub trait PhysicalPlanNodeExt: Sized {
         proto_converter: &dyn PhysicalProtoConverterExtension,
     ) -> Result<protobuf::PhysicalPlanNode> {
         let plan_clone = Arc::clone(&plan);
-        let plan = plan.as_ref() as &dyn Any;
+        let plan = plan.as_ref();
 
         if let Some(exec) = plan.downcast_ref::<ExplainExec>() {
             return protobuf::PhysicalPlanNode::try_from_explain_exec(exec, codec);
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 2022857d4e..50c5fb4c0b 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -89,8 +89,8 @@ use datafusion::physical_plan::windows::{
 };
 use datafusion::physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, InputOrderMode, Partitioning,
-    PhysicalExpr, RangePartitioning, SendableRecordBatchStream, SplitPoint, Statistics,
-    displayable,
+    PhysicalExpr, PlanProperties, RangePartitioning, SendableRecordBatchStream,
+    SplitPoint, Statistics, displayable,
 };
 use datafusion::prelude::{ParquetReadOptions, SessionContext};
 use datafusion::scalar::ScalarValue;
@@ -229,6 +229,74 @@ fn roundtrip_empty() -> Result<()> {
     roundtrip_test(Arc::new(EmptyExec::new(Arc::new(Schema::empty()))))
 }
 
+#[derive(Debug)]
+struct DowncastDelegatingExec {
+    inner: Arc<dyn ExecutionPlan>,
+}
+
+impl DowncastDelegatingExec {
+    fn new(inner: Arc<dyn ExecutionPlan>) -> Self {
+        Self { inner }
+    }
+}
+
+impl DisplayAs for DowncastDelegatingExec {
+    fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
+        self.inner.fmt_as(t, f)
+    }
+}
+
+impl ExecutionPlan for DowncastDelegatingExec {
+    fn name(&self) -> &str {
+        self.inner.name()
+    }
+
+    fn properties(&self) -> &Arc<PlanProperties> {
+        self.inner.properties()
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+        self.inner.children()
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn ExecutionPlan>>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let inner = Arc::clone(&self.inner).with_new_children(children)?;
+        Ok(Arc::new(Self::new(inner)))
+    }
+
+    fn downcast_delegate(&self) -> Option<&dyn ExecutionPlan> {
+        Some(self.inner.as_ref())
+    }
+
+    fn execute(
+        &self,
+        partition: usize,
+        context: Arc<TaskContext>,
+    ) -> Result<SendableRecordBatchStream> {
+        self.inner.execute(partition, context)
+    }
+}
+
+#[test]
+fn serialize_uses_downcast_delegate() -> Result<()> {
+    let inner: Arc<dyn ExecutionPlan> =
+        Arc::new(EmptyExec::new(Arc::new(Schema::empty())));
+    let plan: Arc<dyn ExecutionPlan> = Arc::new(DowncastDelegatingExec::new(inner));
+    let codec = DefaultPhysicalExtensionCodec {};
+
+    let proto = PhysicalPlanNode::try_from_physical_plan(plan, &codec)?;
+
+    assert!(matches!(
+        proto.physical_plan_type,
+        Some(protobuf::physical_plan_node::PhysicalPlanType::Empty(_))
+    ));
+
+    Ok(())
+}
+
 #[test]
 fn roundtrip_date_time_interval() -> Result<()> {
     let schema = Schema::new(vec![


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org