fix: remove WorkTableExec special case in reset_plan_states (#18803)

## Which issue does this PR close?

This PR fixes an oversight in
#16469 that broke wrapper nodes
in recursive queries.

When `with_new_state()` was added as a generic state injection
mechanism, `reset_plan_states()` kept a concrete type check for
`WorkTableExec`. This means wrapper nodes that delegate `as_any()` to
their inner node won't be recognized, causing them to keep stale state
across iterations. This breaks recursive queries for external crates
that wrap execution plans (like tracing or monitoring tools).

## Rationale for this change

The `reset_plan_states()` function uses
`plan.as_any().is::<WorkTableExec>()` to decide which nodes to skip
resetting. This works fine for bare `WorkTableExec` but fails when it's
wrapped by custom nodes as the type check can't see through the wrapper.

This defeats the whole point of `with_new_state()`, which was designed
to let wrapper and third-party nodes participate in recursive queries
without concrete type checks.

The special case was added to save one `Arc::clone()` per iteration, but
it's not worth it since `WorkTableExec::with_new_children()` already
just returns `Arc::clone(&self)` anyway. The shared `WorkTable` state is
preserved through the `Arc<WorkTable>` reference, so the optimization
doesn't buy us much while breaking wrapper nodes.

## What changes are included in this PR?

- Remove the `WorkTableExec` type check in `reset_plan_states()` - just
reset all nodes uniformly via `reset_state()`.
- Simplify the function from 9 lines to 4 lines.
- The shared `WorkTable` state stays correct because
`WorkTableExec::with_new_children()` returns `Arc::clone(&self)`.

## Are these changes tested?

Yes, covered by existing tests:

- All recursive query tests pass, so backward compatibility is
maintained.
- The behavior for bare `WorkTableExec` is identical.
- This fixes wrapper nodes that were broken before.

Since this restores functionality for wrapper nodes rather than changing
core DataFusion behavior, it fixes test failures in external crates
(like `datafusion-tracing`) without needing new tests here.

## Are there any user-facing changes?

**No Breaking Changes:**
- Everything that worked before still works.
- This is purely a bug fix.

**Benefits:**
- External crates can now wrap `WorkTableExec` and implement
`with_new_state()` without breaking recursive queries.
- Tracing, monitoring, and instrumentation wrappers now work correctly
with recursive queries.
- Restores the extensibility that
#16469 was designed to provide.

Author: geoffreyclaude

datafusion/physical-plan/src/recursive_query.rs

@@ -21,7 +21,7 @@ use std::any::Any;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
-use super::work_table::{ReservedBatches, WorkTable, WorkTableExec};
+use super::work_table::{ReservedBatches, WorkTable};
 use crate::execution_plan::{Boundedness, EmissionType};
 use crate::{
     metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet},
@@ -378,13 +378,8 @@ fn assign_work_table(
 /// as the work table changes. When the next iteration executes this plan again, we must clear the left table.
 fn reset_plan_states(plan: Arc<dyn ExecutionPlan>) -> Result<Arc<dyn ExecutionPlan>> {
     plan.transform_up(|plan| {
-        // WorkTableExec's states have already been updated correctly.
-        if plan.as_any().is::<WorkTableExec>() {
-            Ok(Transformed::no(plan))
-        } else {
-            let new_plan = Arc::clone(&plan).reset_state()?;
-            Ok(Transformed::yes(new_plan))
-        }
+        let new_plan = Arc::clone(&plan).reset_state()?;
+        Ok(Transformed::yes(new_plan))
     })
     .data()
 }