fix(proto): honor ExecutionPlan downcast_delegate during serialization (apache#23154)

## Which issue does this PR close?

N/A. This is a follow-up to apache#22559 and apache#21263, but there is no dedicated
issue for this specific bug.

## Rationale for this change

apache#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 apache#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.

Author: geoffreyclaude

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);

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![