The Contract Every DataFusion Operator Signs: A Tour of ExecutionPlan's Declarative Methods

TL;DR: DataFusion’s physical optimizer has no if plan is HashJoinExec branches. Instead, every operator answers a dozen declarative questions — “what distribution do your inputs need?”, “are your inputs already sorted?”, “can you absorb a LIMIT?”, “does a filter on your output reduce your output?” — and the optimizer combines the answers to decide where to put shuffles, which sorts to delete and which filters to push. It is a nice piece of API design, and it is the reason DataFusion can add a new operator without touching the optimizer.

Most query optimizers are a pile of special cases. if the plan is a HashJoin, do X; if it is a Sort, do Y. It works, and it is why those optimizers are hard to extend.

DataFusion’s physical optimizer takes the other approach. The ExecutionPlan trait has about a dozen declarative methods, and the optimizer is a small amount of logic that combines their answers.

The questions

Here is the full list of what an operator can be asked, from datafusion/physical-plan/src/execution_plan.rs:

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 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, ...) -> Result<Option<Arc<dyn ExecutionPlan>>> { ... }
fn gather_filters_for_pushdown(&self, ...) -> Result<FilterDescription> { ... }
fn try_pushdown_sort(&self, ...) -> Result<Option<Arc<dyn ExecutionPlan>>> { ... }
fn with_preserve_order(&self, ...) -> Result<Arc<dyn ExecutionPlan>> { ... }

Each one is a question the optimizer asks. Here is what they mean.

“What distribution do your inputs need?”

fn required_input_distribution(&self) -> Vec<Distribution> { ... }

Distribution::SinglePartition, Distribution::HashPartitioned(exprs, n), Distribution::RangePartitioned(..). A hash join says “I need both my inputs hash-partitioned on the join key, in the same way”. The optimizer then inserts a RepartitionExec where that is not already true.

The newer, more expressive version is input_distribution_requirements, which can express co-partitioning:

/// `InputDistributionRequirements` describes what distribution an operator
/// requires from each child.
///
/// - [`Self::new`] describes independent per-child requirements.
/// - [`Self::co_partitioned`] additionally requires child partitions with the
/// same index to cover compatible key ranges.

The difference matters: “both sides hash-partitioned on the key” is satisfied by two independent repartitions. “Both sides range-partitioned on the key with the same split points” is not — the second side has to be cut at the same boundaries, which means it needs the first side’s boundaries.

“Are your inputs already sorted?”

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

A SortExec says “my input must be sorted by these columns”. The optimizer compares that against the input’s declared output_ordering, and if the requirement is already satisfied, the sort is deleted.

And maintains_input_order answers the complementary question: “does this operator preserve the order of its input?” A FilterExec does; a HashJoinExec does not. Without that, the optimizer could not know whether it is safe to remove a sort.

“Can you absorb a LIMIT?”

fn supports_limit_pushdown(&self) -> bool { ... }
fn with_fetch(&self, _limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>> { ... }

A SortExec says yes, and with_fetch returns a copy of itself that will stop after N rows. An operator that cannot says no, and the optimizer leaves a LimitExec in place.

The default for with_fetch is None — an operator has to opt in to absorbing a limit. And an operator that says it supports limit pushdown but returns None from with_fetch is handled gracefully.

“Does a filter on your output reduce your output?”

fn cardinality_effect(&self) -> CardinalityEffect { ... }

CardinalityEffect::Equal means “the output has the same number of rows as the input” — so a filter pushed above this operator cannot reduce it. CardinalityEffect::Lower (or Unknown) means it might.

This is what stops the optimizer from pushing a filter through an operator where the filter would be a no-op. It is a small method that prevents a whole class of wrong optimisations.

“Can I swap you with a projection?”

fn try_swapping_with_projection(&self, ...) -> Result<Option<Arc<dyn ExecutionPlan>>> { ... }

This is the projection pushdown mechanism, and it is operator-specific in a good way: a ProjectionExec can say “yes, and here is the projection I would rather have” — which is how a projection gets folded into a scan, or a ColumnRewriter gets applied.

“What filters can I take?”

fn gather_filters_for_pushdown(&self, ...) -> Result<FilterDescription> { ... }
fn handle_child_pushdown_result(&self, ...) -> Result<FilterPushdownResult> { ... }

The two-phase filter pushdown negotiation. The parent offers its filters to the child; the child says which ones it can handle entirely, which ones it can handle partially, and which ones it cannot. The result is a FilterPushdownResult with the residual filters that must stay above.

The “partially” case is the interesting one: a Parquet scan can push a predicate into the reader and keep a residual filter to guarantee correctness. That is only possible because the negotiation returns “partially handled”, not just “yes” or “no”.

“Can I push a sort below you?”

fn try_pushdown_sort(&self, ...) -> Result<Option<Arc<dyn ExecutionPlan>>> { ... }

A FilterExec says “yes, put the sort below me and I will filter the sorted stream”. The optimizer then has to know that the filter does not change the order — which is what maintains_input_order is for.

The properties object

All of the declarative answers are collected once, at construction time, into a PlanProperties:

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

which holds the output partitioning, the output ordering, the boundedness and the equivalence properties. The reason is caching: these values never change for a given operator instance, so computing them once at construction beats recomputing them on every optimizer question.

Why it matters

The alternative — special-casing operators in the optimizer — has a familiar failure mode. Every new operator needs a branch in the optimizer, and every branch is a place where the special case and the general case can disagree. DataFusion’s approach means a new operator implements a trait and is immediately optimisable.

It also means the knowledge lives with the operator. A hash join knows what distribution it needs; a sort knows what ordering it needs; a filter knows what it can push down. The optimizer is a policy engine, not a catalogue of operator-specific hacks.

And the method list is a good map of what a modern query engine thinks matters: distribution, ordering, cardinality, limits, projections, filters, and the ability to preserve order. If you were writing an execution engine from scratch, that list is a good place to start.