Trait ExecutionPlan 

pub trait ExecutionPlan:
    Any
    + Debug
    + DisplayAs
    + Send
    + Sync {
    // Required methods
    fn name(&self) -> &str;
    fn properties(&self) -> &Arc<PlanProperties> ;
    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>>;
    fn apply_expressions(
        &self,
        f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
    ) -> Result<TreeNodeRecursion>;
    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>>;
    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream>;

    // Provided methods
    fn static_name() -> &'static str
       where Self: Sized { ... }
    fn downcast_delegate(&self) -> Option<&dyn ExecutionPlan> { ... }
    fn schema(&self) -> SchemaRef { ... }
    fn check_invariants(&self, check: InvariantLevel) -> Result<()> { ... }
    fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>> { ... }
    fn required_input_distribution(&self) -> Vec<Distribution> { ... }
    fn input_distribution_requirements(&self) -> InputDistributionRequirements { ... }
    fn required_input_ordering(&self) -> Vec<Option<OrderingRequirements>> { ... }
    fn maintains_input_order(&self) -> Vec<bool> { ... }
    fn benefits_from_input_partitioning(&self) -> Vec<bool> { ... }
    fn replace_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
        options: ReplaceChildrenOptions,
    ) -> Result<Arc<dyn ExecutionPlan>> { ... }
    fn with_new_children_and_same_properties(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> { ... }
    fn reset_state(self: Arc<Self>) -> Result<Arc<dyn ExecutionPlan>> { ... }
    fn repartitioned(
        &self,
        _target_partitions: usize,
        _config: &ConfigOptions,
    ) -> Result<Option<Arc<dyn ExecutionPlan>>> { ... }
    fn metrics(&self) -> Option<MetricsSet> { ... }
    fn partition_statistics(
        &self,
        partition: Option<usize>,
    ) -> Result<Arc<Statistics>> { ... }
    fn statistics_from_inputs(
        &self,
        _input_stats: &[Arc<Statistics>],
        args: &StatisticsArgs,
    ) -> Result<Arc<Statistics>> { ... }
    fn child_stats_requests(&self, _partition: Option<usize>) -> Vec<ChildStats> { ... }
    fn supports_limit_pushdown(&self) -> bool { ... }
    fn with_fetch(
        &self,
        _limit: Option<usize>,
    ) -> Option<Arc<dyn ExecutionPlan>> { ... }
    fn fetch(&self) -> Option<usize> { ... }
    fn cardinality_effect(&self) -> CardinalityEffect { ... }
    fn try_swapping_with_projection(
        &self,
        _projection: &ProjectionExec,
    ) -> Result<Option<Arc<dyn ExecutionPlan>>> { ... }
    fn gather_filters_for_pushdown(
        &self,
        _phase: FilterPushdownPhase,
        parent_filters: Vec<Arc<dyn PhysicalExpr>>,
        _config: &ConfigOptions,
    ) -> Result<FilterDescription> { ... }
    fn handle_child_pushdown_result(
        &self,
        _phase: FilterPushdownPhase,
        child_pushdown_result: ChildPushdownResult,
        _config: &ConfigOptions,
    ) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> { ... }
    fn with_new_state(
        &self,
        _state: Arc<dyn Any + Send + Sync>,
    ) -> Option<Arc<dyn ExecutionPlan>> { ... }
    fn try_pushdown_sort(
        &self,
        _order: &[PhysicalSortExpr],
    ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> { ... }
    fn with_preserve_order(
        &self,
        _preserve_order: bool,
    ) -> Option<Arc<dyn ExecutionPlan>> { ... }
    fn try_to_proto(
        &self,
        _ctx: &ExecutionPlanEncodeCtx<'_>,
    ) -> Result<Option<PhysicalPlanNode>> { ... }
}

Represent nodes in the DataFusion Physical Plan.

Calling execute produces an async SendableRecordBatchStream of RecordBatch that incrementally computes a partition of the ExecutionPlan’s output from its input. See Partitioning for more details on partitioning.

Methods such as Self::schema and Self::properties communicate properties of the output to the DataFusion optimizer, and methods such as required_input_distribution and required_input_ordering express requirements of the ExecutionPlan from its input.

ExecutionPlan can be displayed in a simplified form using the return value from displayable in addition to the (normally quite verbose) Debug output.

Examples

See datafusion-examples for examples, including memory_pool_execution_plan.rs which shows how to implement a custom ExecutionPlan with memory tracking and spilling support.

Required Methods

fn name(&self) -> &str

Short name for the ExecutionPlan, such as ‘DataSourceExec’.

Implementation note: this method can just proxy to static_name if no special action is needed. It doesn’t provide a default implementation like that because this method doesn’t require the Sized constrain to allow a wilder range of use cases.

fn properties(&self) -> &Arc<PlanProperties>

Return properties of the output of the ExecutionPlan, such as output ordering(s), partitioning information etc.

This information is available via methods on ExecutionPlanProperties trait, which is implemented for all ExecutionPlans.

fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>>

Get a list of children ExecutionPlans that act as inputs to this plan. The returned list will be empty for leaf nodes such as scans, will contain a single value for unary nodes, or two values for binary nodes (such as joins).

fn apply_expressions( &self, f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>, ) -> Result<TreeNodeRecursion>

Apply a closure f to each root expression that this node owns and uses during execution, either by evaluating it or updating it dynamically.

An expression must not be visited solely because it describes an input or output property, such as cached ordering, partitioning, or equivalence metadata. However, these may be traversed indirectly. For example, RepartitionExec visits the partitioning expressions it evaluates and SortExec visits the sort expressions it evaluates to order rows.

This method is shallow: it must not visit expression children or expressions owned by child execution plans.

Similarly to other TreeNode APIs, the closure can return TreeNodeRecursion::Stop to stop iteration, otherwise iteration should continue. Note that TreeNodeRecursion::Continue and TreeNodeRecursion::Jump are equivalent because this method is not recursive.

Example Usage

// Count the number of expressions
let mut count = 0;
plan.apply_expressions(&mut |_expr| {
    count += 1;
    Ok(TreeNodeRecursion::Continue)
})?;

Implementation Examples

Node with expressions (e.g., FilterExec, ProjectionExec)

Use apply_expression_roots to implement this method. It abstracts away the TreeNodeRecursion iteration from implementors.

fn apply_expressions(
    &self,
    f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
) -> Result<TreeNodeRecursion> {
    apply_expression_roots([&self.predicate], f)
}

Node with no expressions (e.g., EmptyExec, MemoryExec)

fn apply_expressions(
    &self,
    _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
) -> Result<TreeNodeRecursion> {
    Ok(TreeNodeRecursion::Continue)
}

fn with_new_children( self: Arc<Self>, children: Vec<Arc<dyn ExecutionPlan>>, ) -> Result<Arc<dyn ExecutionPlan>>

👎Deprecated since 55.0.0:

Use ExecutionPlan::replace_children with ReplaceChildrenOptions

Deprecated.

DataFusion will remove this method in the future in favor of ExecutionPlan::replace_children.

Note that this method is still required by the trait; implementations should delegate to ExecutionPlan::replace_children with ChildrenPropertiesMode::Recompute.

Example Implementation

impl ExecutionPlan for MyExec {
    // ...
    fn replace_children(
        self: Arc<Self>,
        mut children: Vec<Arc<dyn ExecutionPlan>>,
        _options: ReplaceChildrenOptions,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        Ok(Arc::new(MyExec {
            input: children.swap_remove(0),
        }))
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        // call into `replace_children` with `ReplaceChildrenOptions`
        self.replace_children(
            children,
            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
        )
    }
}

fn execute( &self, partition: usize, context: Arc<TaskContext>, ) -> Result<SendableRecordBatchStream>

Begin execution of partition, returning a Stream of RecordBatches.

Notes

The execute method itself is not async but it returns an async futures::stream::Stream. This Stream should incrementally compute the output, RecordBatch by RecordBatch (in a streaming fashion). Most ExecutionPlans should not do any work before the first RecordBatch is requested from the stream.

RecordBatchStreamAdapter can be used to convert an async Stream into a SendableRecordBatchStream.

Using async Streams allows for network I/O during execution and takes advantage of Rust’s built in support for async continuations and crate ecosystem.

Error handling

Any error that occurs during execution is sent as an Err in the output stream.

ExecutionPlan implementations in DataFusion cancel additional work immediately once an error occurs. The rationale is that if the overall query will return an error, any additional work such as continued polling of inputs will be wasted as it will be thrown away.

Cancellation / Aborting Execution

The Stream that is returned must ensure that any allocated resources are freed when the stream itself is dropped. This is particularly important for spawned tasks or threads. Unless care is taken to “abort” such tasks, they may continue to consume resources even after the plan is dropped, generating intermediate results that are never used. Thus, spawn is disallowed, and instead use SpawnedTask.

To enable timely cancellation, the Stream that is returned must not block the CPU indefinitely and must yield back to the tokio runtime regularly. In a typical ExecutionPlan, this automatically happens unless there are special circumstances; e.g. when the computational complexity of processing a batch is superlinear. See this general guideline for more context on this point, which explains why one should avoid spending a long time without reaching an await/yield point in asynchronous runtimes. This can be achieved by using the utilities from the coop module, by manually returning Poll::Pending and setting up wakers appropriately, or by calling tokio::task::yield_now() when appropriate. In special cases that warrant manual yielding, determination for “regularly” may be made using the Tokio task budget, a timer (being careful with the overhead-heavy system call needed to take the time), or by counting rows or batches.

The cancellation benchmark tracks some cases of how quickly queries can be cancelled.

For more details see SpawnedTask, JoinSet and RecordBatchReceiverStreamBuilder for structures to help ensure all background tasks are cancelled.

Implementation Examples

While async Streams have a non trivial learning curve, the futures crate provides StreamExt and TryStreamExt which help simplify many common operations.

Here are some common patterns:

Return Precomputed RecordBatch

We can return a precomputed RecordBatch as a Stream:

struct MyPlan {
    batch: RecordBatch,
}

impl MyPlan {
    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        // use functions from futures crate to convert the batch into a stream
        let fut = futures::future::ready(Ok(self.batch.clone()));
        let stream = futures::stream::once(fut);
        Ok(Box::pin(RecordBatchStreamAdapter::new(
            self.batch.schema(),
            stream,
        )))
    }
}

Lazily (async) Compute RecordBatch

We can also lazily compute a RecordBatch when the returned Stream is polled

struct MyPlan {
    schema: SchemaRef,
}

/// Returns a single batch when the returned stream is polled
async fn get_batch() -> Result<RecordBatch> {
    todo!()
}

impl MyPlan {
    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        let fut = get_batch();
        let stream = futures::stream::once(fut);
        Ok(Box::pin(RecordBatchStreamAdapter::new(
            self.schema.clone(),
            stream,
        )))
    }
}

Lazily (async) create a Stream

If you need to create the return Stream using an async function, you can do so by flattening the result:

struct MyPlan {
    schema: SchemaRef,
}

/// async function that returns a stream
async fn get_batch_stream() -> Result<SendableRecordBatchStream> {
    todo!()
}

impl MyPlan {
    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        // A future that yields a stream
        let fut = get_batch_stream();
        // Use TryStreamExt::try_flatten to flatten the stream of streams
        let stream = futures::stream::once(fut).try_flatten();
        Ok(Box::pin(RecordBatchStreamAdapter::new(
            self.schema.clone(),
            stream,
        )))
    }
}

Provided Methods

fn static_name() -> &'static str

where Self: Sized,

Short name for the ExecutionPlan, such as ‘DataSourceExec’. Like name but can be called without an instance.

fn downcast_delegate(&self) -> Option<&dyn ExecutionPlan>

Returns the plan that provides this plan’s public ExecutionPlan downcast identity.

This hook is for wrapper nodes that delegate their public downcast identity to another plan while adding cross-cutting behavior such as instrumentation. The default implementation returns None, meaning this plan’s concrete type is used for type introspection.

Most ExecutionPlan implementations should use the default None; override this only for wrapper plans that intentionally delegate their public downcast identity to another plan.

The is and downcast_ref helpers follow the returned delegate instead of checking the current concrete type, making intermediate delegating wrappers invisible to normal downcast-based inspection.

Implementations that opt in should return the delegate plan, not self.

This is independent from Self::children and should not be used for plan traversal or optimizer rewrites.

fn schema(&self) -> SchemaRef

Get the schema for this execution plan

fn check_invariants(&self, check: InvariantLevel) -> Result<()>

Returns an error if this individual node does not conform to its invariants. These invariants are typically only checked in debug mode.

A default set of invariants is provided in the check_default_invariants function. The default implementation of check_invariants calls this function. Extension nodes can provide their own invariants.

fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>>

Returns the dynamic expressions produced by this plan node.

A dynamic expression is produced when this node updates or completes its runtime state during execution. Expressions that this node only consumes must not be returned. This method is shallow and does not include dynamic expressions produced by child plans.

Each returned expression must have a PhysicalExpr::expression_id since all dynamic expressions such as DynamicFilterPhysicalExpr have an expression id.

fn required_input_distribution(&self) -> Vec<Distribution>

👎Deprecated since 55.0.0:

Use input_distribution_requirements

Specifies simple per-child input distribution requirements.

Deprecated: override Self::input_distribution_requirements instead.

By default, each child has Distribution::UnspecifiedDistribution.

fn input_distribution_requirements(&self) -> InputDistributionRequirements

Specifies the input distribution requirements for this plan.

The default implementation wraps Self::required_input_distribution. Override this method for richer requirements, such as allowing alternate satisfaction policies or requiring multiple children to be co-partitioned. See InputDistributionRequirements for details.

fn required_input_ordering(&self) -> Vec<Option<OrderingRequirements>>

Specifies the ordering required for all of the children of this ExecutionPlan.

For each child, it’s the local ordering requirement within each partition rather than the global ordering

NOTE that checking !is_empty() does not check for a required input ordering. Instead, the correct check is that at least one entry must be Some

fn maintains_input_order(&self) -> Vec<bool>

Returns false if this ExecutionPlan’s implementation may reorder rows within or between partitions.

For example, Projection, Filter, and Limit maintain the order of inputs – they may transform values (Projection) or not produce the same number of rows that went in (Filter and Limit), but the rows that are produced go in the same way.

DataFusion uses this metadata to apply certain optimizations such as automatically repartitioning correctly.

The default implementation returns false

WARNING: if you override this default, you MUST ensure that the ExecutionPlan’s maintains the ordering invariant or else DataFusion may produce incorrect results.

fn benefits_from_input_partitioning(&self) -> Vec<bool>

Specifies whether the ExecutionPlan benefits from increased parallelization at its input for each child.

If returns true, the ExecutionPlan would benefit from partitioning its corresponding child (and thus from more parallelism). For ExecutionPlan that do very little work the overhead of extra parallelism may outweigh any benefits

The default implementation returns true unless this ExecutionPlan has signalled it requires a single child input partition.

fn replace_children( self: Arc<Self>, children: Vec<Arc<dyn ExecutionPlan>>, options: ReplaceChildrenOptions, ) -> Result<Arc<dyn ExecutionPlan>>

Returns a clone of the existing plan with the children replaced, skipping recomputation of plan properties when the options indicate the new children’s properties are unchanged.

Callers should typically call replace_children_if_necessary and not invoke this method directly.

fn with_new_children_and_same_properties( self: Arc<Self>, children: Vec<Arc<dyn ExecutionPlan>>, ) -> Result<Arc<dyn ExecutionPlan>>

👎Deprecated since 55.0.0:

Use ExecutionPlan::replace_children with ReplaceChildrenOptions

Deprecated. Implement ExecutionPlan::replace_children instead.

fn reset_state(self: Arc<Self>) -> Result<Arc<dyn ExecutionPlan>>

Reset any internal state within this ExecutionPlan.

This method is called when an ExecutionPlan needs to be re-executed, such as in recursive queries. Unlike ExecutionPlan::replace_children, this method ensures that any stateful components (e.g., DynamicFilterPhysicalExpr) are reset to their initial state.

The default implementation simply calls ExecutionPlan::replace_children with the existing children, effectively creating a new instance of the ExecutionPlan with the same children but without necessarily resetting any internal state. Implementations that require resetting of some internal state should override this method to provide the necessary logic.

This method should not reset state recursively for children, as it is expected that it will be called from within a walk of the execution plan tree so that it will be called on each child later or was already called on each child.

Note to implementers: unlike ExecutionPlan::replace_children this method does not accept new children as an argument, thus it is expected that any cached plan properties will remain valid after the reset.

fn repartitioned( &self, _target_partitions: usize, _config: &ConfigOptions, ) -> Result<Option<Arc<dyn ExecutionPlan>>>

If supported, attempt to increase the partitioning of this ExecutionPlan to produce target_partitions partitions.

If the ExecutionPlan does not support changing its partitioning, returns Ok(None) (the default).

If the ExecutionPlan can increase its partitioning, but not to target_partitions, it may return an ExecutionPlan with fewer partitions. This might happen, for example, if each new partition would be too small to be efficiently processed individually.

The DataFusion optimizer attempts to use as many threads as possible by repartitioning its inputs to match the target number of threads available (target_partitions). Some data sources, such as the built in CSV and Parquet readers, implement this method as they are able to read from their input files in parallel, regardless of how the source data is split amongst files.

fn metrics(&self) -> Option<MetricsSet>

Return a snapshot of the set of Metrics for this ExecutionPlan. If no Metrics are available, return None.

While the values of the metrics in the returned MetricsSets may change as execution progresses, the specific metrics will not.

Once self.execute() has returned (technically the future is resolved) for all available partitions, the set of metrics should be complete. If this function is called prior to execute() new metrics may appear in subsequent calls.

fn partition_statistics( &self, partition: Option<usize>, ) -> Result<Arc<Statistics>>

👎Deprecated since 55.0.0:

Use StatisticsContext::compute instead

Returns statistics for a specific partition of this ExecutionPlan node.

Deprecated: use StatisticsContext::compute instead.

fn statistics_from_inputs( &self, _input_stats: &[Arc<Statistics>], args: &StatisticsArgs, ) -> Result<Arc<Statistics>>

Returns statistics for a specific partition of this ExecutionPlan node, given pre-computed child statistics.

If statistics are not available, should return Statistics::new_unknown (the default), not an error. If args.partition() is None, it returns statistics for all partitions.

Implementations should not call StatisticsContext::compute from within this method; child statistics are provided via input_stats.

Use StatisticsContext::compute to initiate a full plan-tree walk.

fn child_stats_requests(&self, _partition: Option<usize>) -> Vec<ChildStats>

Returns, per child, which statistics the StatisticsContext should resolve before calling Self::statistics_from_inputs.

One entry per child (same order as Self::children): ChildStats::At requests the child’s statistics at a partition (None = overall); ChildStats::Skip omits a child whose statistics this node does not need (a Statistics::new_unknown placeholder fills its input_stats slot).

The default skips every child, so a node that derives nothing from its children (for example one that only overrides the deprecated Self::partition_statistics) triggers no child traversal. A node that reads input_stats in Self::statistics_from_inputs must override this to declare the children it uses.

fn supports_limit_pushdown(&self) -> bool

Returns true if a limit can be safely pushed down through this ExecutionPlan node.

If this method returns true, and the query plan contains a limit at the output of this node, DataFusion will push the limit to the input of this node.

fn with_fetch(&self, _limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>>

Returns a fetching variant of this ExecutionPlan node, if it supports fetch limits. Returns None otherwise.

See physical optimizer rule limit_pushdown for details.

fn fetch(&self) -> Option<usize>

Gets the fetch count for the operator, None means there is no fetch.

fn cardinality_effect(&self) -> CardinalityEffect

Gets the effect on cardinality, if known

fn try_swapping_with_projection( &self, _projection: &ProjectionExec, ) -> Result<Option<Arc<dyn ExecutionPlan>>>

Attempts to push down the given projection into the input of this ExecutionPlan.

If the operator supports this optimization, the resulting plan will be: self_new <- projection <- source, starting from projection <- self <- source. Otherwise, it returns the current ExecutionPlan as-is.

Returns Ok(Some(...)) if pushdown is applied, Ok(None) if it is not supported or not possible, or Err on failure.

fn gather_filters_for_pushdown( &self, _phase: FilterPushdownPhase, parent_filters: Vec<Arc<dyn PhysicalExpr>>, _config: &ConfigOptions, ) -> Result<FilterDescription>

Collect filters that this node can push down to its children. Filters that are being pushed down from parents are passed in, and the node may generate additional filters to push down. For example, given the plan FilterExec -> HashJoinExec -> DataSourceExec, what will happen is that we recurse down the plan calling ExecutionPlan::gather_filters_for_pushdown:

1. FilterExec::gather_filters_for_pushdown is called with no parent filters so it only returns that FilterExec wants to push down its own predicate.
2. HashJoinExec::gather_filters_for_pushdown is called with the filter from FilterExec, which it only allows to push down to one side of the join (unless it’s on the join key) but it also adds its own filters (e.g. pushing down a bloom filter of the hash table to the scan side of the join).
3. DataSourceExec::gather_filters_for_pushdown is called with both filters from HashJoinExec and FilterExec, however DataSourceExec::gather_filters_for_pushdown doesn’t actually do anything since it has no children and no additional filters to push down. It’s only once ExecutionPlan::handle_child_pushdown_result is called on DataSourceExec as we recurse up the plan that DataSourceExec can actually bind the filters.

The default implementation bars all parent filters from being pushed down and adds no new filters. This is the safest option, making filter pushdown opt-in on a per-node basis.

There are two different phases in filter pushdown, which some operators may handle the same and some differently. Depending on the phase the operator may or may not be allowed to modify the plan. See FilterPushdownPhase for more details.

Implementations must preserve the order of parent_filters in the returned child FilterDescription: each child parent-filter result is matched back to the corresponding input parent filter by position. Unsupported filters should therefore be marked unsupported in place, rather than removed or appended after supported filters.

fn handle_child_pushdown_result( &self, _phase: FilterPushdownPhase, child_pushdown_result: ChildPushdownResult, _config: &ConfigOptions, ) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>>

Handle the result of a child pushdown.

This method is called as we recurse back up the plan tree after pushing filters down to child nodes via ExecutionPlan::gather_filters_for_pushdown. It allows the current node to process the results of filter pushdown from its children, deciding whether to absorb filters, modify the plan, or pass filters back up to its parent.

Purpose and Context: Filter pushdown is a critical optimization in DataFusion that aims to reduce the amount of data processed by applying filters as early as possible in the query plan. This method is part of the second phase of filter pushdown, where results are propagated back up the tree after being pushed down. Each node can inspect the pushdown results from its children and decide how to handle any unapplied filters, potentially optimizing the plan structure or filter application.

Behavior in Different Nodes:

• For a DataSourceExec, this often means absorbing the filters to apply them during the scan phase (late materialization), reducing the data read from the source.
• A FilterExec may absorb any filters its children could not handle, combining them with its own predicate. If no filters remain (i.e., the predicate becomes trivially true), it may remove itself from the plan altogether. It typically marks parent filters as supported, indicating they have been handled.
• A HashJoinExec might ignore the pushdown result if filters need to be applied during the join operation. It passes the parent filters back up wrapped in FilterPushdownPropagation::if_any, discarding any self-filters from children.

Example Walkthrough: Consider a query plan: FilterExec (f1) -> HashJoinExec -> DataSourceExec.

1. Downward Phase (gather_filters_for_pushdown): Starting at FilterExec, the filter f1 is gathered and pushed down to HashJoinExec. HashJoinExec may allow f1 to pass to one side of the join or add its own filters (e.g., a min-max filter from the build side), then pushes filters to DataSourceExec. DataSourceExec, being a leaf node, has no children to push to, so it prepares to handle filters in the upward phase.
2. Upward Phase (handle_child_pushdown_result): Starting at DataSourceExec, it absorbs applicable filters from HashJoinExec for late materialization during scanning, marking them as supported. HashJoinExec receives the result, decides whether to apply any remaining filters during the join, and passes unhandled filters back up to FilterExec. FilterExec absorbs any unhandled filters, updates its predicate if necessary, or removes itself if the predicate becomes trivial (e.g., lit(true)), and marks filters as supported for its parent.

The default implementation is a no-op that passes the result of pushdown from the children to its parent transparently, ensuring no filters are lost if a node does not override this behavior.

Notes for Implementation: When returning filters via FilterPushdownPropagation, the order of filters need not match the order they were passed in via child_pushdown_result. However, preserving the order is recommended for debugging and ease of reasoning about the resulting plans.

Helper Methods for Customization: There are various helper methods to simplify implementing this method:

• FilterPushdownPropagation::if_any: Marks all parent filters as supported as long as at least one child supports them.
• FilterPushdownPropagation::if_all: Marks all parent filters as supported as long as all children support them.
• FilterPushdownPropagation::with_parent_pushdown_result: Allows adding filters to the propagation result, indicating which filters are supported by the current node.
• FilterPushdownPropagation::with_updated_node: Allows updating the current node in the propagation result, used if the node has modified its plan based on the pushdown results.

Filter Pushdown Phases: There are two different phases in filter pushdown (Pre and others), which some operators may handle differently. Depending on the phase, the operator may or may not be allowed to modify the plan. See FilterPushdownPhase for more details on phase-specific behavior.

fn with_new_state( &self, _state: Arc<dyn Any + Send + Sync>, ) -> Option<Arc<dyn ExecutionPlan>>

Injects arbitrary run-time state into this execution plan, returning a new plan instance that incorporates that state if it is relevant to the concrete node implementation.

This is a generic entry point: the state can be any type wrapped in Arc<dyn Any + Send + Sync>. A node that cares about the state should down-cast it to the concrete type it expects and, if successful, return a modified copy of itself that captures the provided value. If the state is not applicable, the default behaviour is to return None so that parent nodes can continue propagating the attempt further down the plan tree.

For example, WorkTableExec down-casts the supplied state to an Arc<WorkTable> in order to wire up the working table used during recursive-CTE execution. Similar patterns can be followed by custom nodes that need late-bound dependencies or shared state.

fn try_pushdown_sort( &self, _order: &[PhysicalSortExpr], ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>>

Try to push down sort ordering requirements to this node.

This method is called during sort pushdown optimization to determine if this node can optimize for a requested sort ordering. Implementations should:

• Return SortOrderPushdownResult::Exact if the node can guarantee the exact ordering (allowing the Sort operator to be removed)
• Return SortOrderPushdownResult::Inexact if the node can optimize for the ordering but cannot guarantee perfect sorting (Sort operator is kept)
• Return SortOrderPushdownResult::Unsupported if the node cannot optimize for the ordering

For transparent nodes (that preserve ordering), implement this to delegate to children and wrap the result with a new instance of this node.

Default implementation returns Unsupported.

fn with_preserve_order( &self, _preserve_order: bool, ) -> Option<Arc<dyn ExecutionPlan>>

Returns a variant of this ExecutionPlan that is aware of order-sensitivity.

This is used to signal to data sources that the output ordering must be preserved, even if it might be more efficient to ignore it (e.g. by skipping some row groups in Parquet).

fn try_to_proto( &self, _ctx: &ExecutionPlanEncodeCtx<'_>, ) -> Result<Option<PhysicalPlanNode>>

Available on crate feature proto only.

Serialize this plan to its protobuf representation, if it knows how.

This is the ExecutionPlan analog of PhysicalExpr::try_to_proto.

• Ok(None) (the default) — “I don’t serialize myself”; the caller (datafusion-proto) falls back to the central downcast chain. Every un-migrated plan keeps its existing behavior.
• Ok(Some(node)) — fully serialized; the caller must not fall back.
• Err(_) — a real failure (e.g. a child failed to serialize).

Only self-contained plans should override this — see crate::proto for the session-dependency boundary.

Implementations

impl dyn ExecutionPlan

pub fn is<T: ExecutionPlan>(&self) -> bool

Returns true if the plan is of type T.

If this plan provides a ExecutionPlan::downcast_delegate, delegates to it.

Prefer this over downcast_ref::<T>().is_some(). Works correctly when called on Arc<dyn ExecutionPlan> via auto-deref.

pub fn downcast_ref<T: ExecutionPlan>(&self) -> Option<&T>

Attempts to downcast this plan to a concrete type T, returning None if the plan is not of that type.

If this plan provides a ExecutionPlan::downcast_delegate, delegates to it.

Works correctly when called on Arc<dyn ExecutionPlan> via auto-deref, unlike (&arc as &dyn Any).downcast_ref::<T>() which would attempt to downcast the Arc itself.

Trait Implementations

impl DynTreeNode for dyn ExecutionPlan

fn arc_children(&self) -> Vec<&Arc<Self>>

Returns all children of the specified TreeNode.

fn with_new_arc_children( &self, arc_self: Arc<Self>, new_children: Vec<Arc<Self>>, ) -> Result<Arc<Self>>

Constructs a new node with the specified children.

impl ExecutionPlanProperties for &dyn ExecutionPlan

fn output_partitioning(&self) -> &Partitioning

Specifies how the output of this ExecutionPlan is split into partitions.

fn output_ordering(&self) -> Option<&LexOrdering>

If the output of this ExecutionPlan within each partition is sorted, returns Some(keys) describing the ordering. A None return value indicates no assumptions should be made on the output ordering.

fn boundedness(&self) -> Boundedness

Boundedness information of the stream corresponding to this ExecutionPlan. For more details, see Boundedness.

fn pipeline_behavior(&self) -> EmissionType

Indicates how the stream of this ExecutionPlan emits its results. For more details, see EmissionType.

fn equivalence_properties(&self) -> &EquivalenceProperties

Get the EquivalenceProperties within the plan.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl ExecutionPlan for AggregateExec

impl ExecutionPlan for AnalyzeExec

impl ExecutionPlan for AsyncFuncExec

impl ExecutionPlan for BarrierExec

impl ExecutionPlan for BlockingExec

impl ExecutionPlan for BoundedWindowAggExec

impl ExecutionPlan for BufferExec

impl ExecutionPlan for CoalesceBatchesExec

impl ExecutionPlan for CoalescePartitionsExec

impl ExecutionPlan for CooperativeExec

impl ExecutionPlan for CrossJoinExec

impl ExecutionPlan for EmptyExec

impl ExecutionPlan for ErrorExec

impl ExecutionPlan for ExplainExec

impl ExecutionPlan for FilterExec

impl ExecutionPlan for GlobalLimitExec

impl ExecutionPlan for HashJoinExec

impl ExecutionPlan for InterleaveExec

impl ExecutionPlan for LazyMemoryExec

impl ExecutionPlan for LocalLimitExec

impl ExecutionPlan for MockExec

impl ExecutionPlan for NestedLoopJoinExec

impl ExecutionPlan for PanicExec

impl ExecutionPlan for PartialSortExec

impl ExecutionPlan for PartitionedTopKExec

impl ExecutionPlan for PiecewiseMergeJoinExec

impl ExecutionPlan for PlaceholderRowExec

impl ExecutionPlan for ProjectionExec

impl ExecutionPlan for RecursiveQueryExec

impl ExecutionPlan for RepartitionExec

impl ExecutionPlan for ScalarSubqueryExec

impl ExecutionPlan for SortExec

impl ExecutionPlan for SortMergeJoinExec

impl ExecutionPlan for SortPreservingMergeExec

impl ExecutionPlan for StatisticsExec

impl ExecutionPlan for StreamingTableExec

impl ExecutionPlan for SymmetricHashJoinExec

impl ExecutionPlan for TestMemoryExec

impl ExecutionPlan for UnionExec

impl ExecutionPlan for UnnestExec

impl ExecutionPlan for WindowAggExec

impl ExecutionPlan for WorkTableExec