feat: support repartitioning of FFI execution plans (apache#20449)

## Which issue does this PR close?

This is a blocker for apache#20450

## Rationale for this change

This PR introduces an important concept in the FFI work to avoids
creating wrappers upon wrappers of plans. It was discovered as part of
the work to create FFI physical optimizer rules. Suppose we have a
foreign plan. Then we attempt to turn this into an FFI plan. What we
will end up with currently is a FFI plan where the underlying private
data is a foreign plan that additionally contains a FFI plan. Instead
*any* time we are creating an FFI object we should check to see if it is
locally downcastable to a Foreign plan and if so to just access the
already existing FFI object.

This pattern is adapted across all FFI objects in this PR.

With this work in place we can also properly support repartioning via
FFI as well as `new_with_children` via FFI.

## What changes are included in this PR?

- Adds access pattern for creating new FFI objects. When they are
already a locally downcastable to a Foreign wrapper then we simply get
the underlying existing FFI object instead of creating a wrapper around
a wrapper.
- Implement repartitioning and new_with_children via FFI on execution
plans.

## Are these changes tested?

Integration tests are added.

## Are there any user-facing changes?

The one use facing change is that for some of the aggregates and
accumulators that take in closures we require these closures to be
static so that we can downcast the boxed traits.

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: timsaucer

datafusion/catalog/src/table.rs

@@ -486,7 +486,7 @@ pub trait TableProviderFactory: Debug + Sync + Send {
 }
 
 /// A trait for table function implementations
-pub trait TableFunctionImpl: Debug + Sync + Send {
+pub trait TableFunctionImpl: Debug + Sync + Send + Any {
     /// Create a table provider
     fn call(&self, args: &[Expr]) -> Result<Arc<dyn TableProvider>>;
 }

datafusion/expr-common/src/accumulator.rs

@@ -48,7 +48,7 @@ use std::fmt::Debug;
 /// [`evaluate`]: Self::evaluate
 /// [`merge_batch`]: Self::merge_batch
 /// [window function]: https://en.wikipedia.org/wiki/Window_function_(SQL)
-pub trait Accumulator: Send + Sync + Debug {
+pub trait Accumulator: Send + Sync + Debug + std::any::Any {
     /// Updates the accumulator's state from its input.
     ///
     /// `values` contains the arguments to this aggregate function.

datafusion/expr-common/src/groups_accumulator.rs

@@ -108,7 +108,7 @@ impl EmitTo {
 ///
 /// [`Accumulator`]: crate::accumulator::Accumulator
 /// [Aggregating Millions of Groups Fast blog]: https://arrow.apache.org/blog/2023/08/05/datafusion_fast_grouping/
-pub trait GroupsAccumulator: Send {
+pub trait GroupsAccumulator: Send + std::any::Any {
     /// Updates the accumulator's state from its arguments, encoded as
     /// a vector of [`ArrayRef`]s.
     ///

datafusion/expr/src/partition_evaluator.rs

@@ -90,7 +90,7 @@ use crate::window_state::WindowAggState;
 /// For more background, please also see the [User defined Window Functions in DataFusion blog]
 ///
 /// [User defined Window Functions in DataFusion blog]: https://datafusion.apache.org/blog/2025/04/19/user-defined-window-functions
-pub trait PartitionEvaluator: Debug + Send {
+pub trait PartitionEvaluator: Debug + Send + std::any::Any {
     /// When the window frame has a fixed beginning (e.g UNBOUNDED
     /// PRECEDING), some functions such as FIRST_VALUE, LAST_VALUE and
     /// NTH_VALUE do not need the (unbounded) input once they have

datafusion/ffi/src/catalog_provider.rs

@@ -250,6 +250,11 @@ impl FFI_CatalogProvider {
         runtime: Option<Handle>,
         logical_codec: FFI_LogicalExtensionCodec,
     ) -> Self {
+        if let Some(provider) = provider.as_any().downcast_ref::<ForeignCatalogProvider>()
+        {
+            return provider.0.clone();
+        }
+
         let private_data = Box::new(ProviderPrivateData { provider, runtime });
 
         Self {

datafusion/ffi/src/catalog_provider_list.rs

@@ -212,6 +212,13 @@ impl FFI_CatalogProviderList {
         runtime: Option<Handle>,
         logical_codec: FFI_LogicalExtensionCodec,
     ) -> Self {
+        if let Some(provider) = provider
+            .as_any()
+            .downcast_ref::<ForeignCatalogProviderList>()
+        {
+            return provider.0.clone();
+        }
+
         let private_data = Box::new(ProviderPrivateData { provider, runtime });
 
         Self {