Struct RepartitionExec 

pub struct RepartitionExec { /* private fields */ }

Maps N input partitions to M output partitions based on a Partitioning scheme.

Background

DataFusion, like most other commercial systems, with the notable exception of DuckDB, uses the “Exchange Operator” based approach to parallelism which works well in practice given sufficient care in implementation.

DataFusion’s planner picks the target number of partitions and then RepartitionExec redistributes RecordBatches to that number of output partitions.

For example, given target_partitions=3 (trying to use 3 cores) but scanning an input with 2 partitions, RepartitionExec can be used to get 3 even streams of RecordBatches

        ▲                  ▲                  ▲
        │                  │                  │
        │                  │                  │
        │                  │                  │
 ┌───────────────┐  ┌───────────────┐  ┌───────────────┐
 │    GroupBy    │  │    GroupBy    │  │    GroupBy    │
 │   (Partial)   │  │   (Partial)   │  │   (Partial)   │
 └───────────────┘  └───────────────┘  └───────────────┘
        ▲                  ▲                  ▲
        └──────────────────┼──────────────────┘
                           │
              ┌─────────────────────────┐
              │     RepartitionExec     │
              │   (hash/round robin)    │
              └─────────────────────────┘
                         ▲   ▲
             ┌───────────┘   └───────────┐
             │                           │
             │                           │
        .─────────.                 .─────────.
     ,─'           '─.           ,─'           '─.
    ;      Input      :         ;      Input      :
    :   Partition 0   ;         :   Partition 1   ;
     ╲               ╱           ╲               ╱
      '─.         ,─'             '─.         ,─'
         `───────'                   `───────'

Error Handling

If any of the input partitions return an error, the error is propagated to all output partitions and inputs are not polled again.

Output Ordering

If more than one stream is being repartitioned, the output will be some arbitrary interleaving (and thus unordered) unless Self::with_preserve_order specifies otherwise.

Spilling Architecture

RepartitionExec uses SpillPool channels to handle memory pressure during repartitioning. Each (input partition, output partition) pair gets its own SpillPool channel for FIFO ordering.

Input Partitions (N)          Output Partitions (M)
────────────────────          ─────────────────────

   Input 0 ──┐                      ┌──▶ Output 0
             │  ┌──────────────┐    │
             ├─▶│ SpillPool    │────┤
             │  │ [In0→Out0]   │    │
   Input 1 ──┤  └──────────────┘    ├──▶ Output 1
             │                       │
             │  ┌──────────────┐    │
             ├─▶│ SpillPool    │────┤
             │  │ [In1→Out0]   │    │
   Input 2 ──┤  └──────────────┘    ├──▶ Output 2
             │                      │
             │       ... (N×M SpillPools total)
             │                      │
             │  ┌──────────────┐    │
             └─▶│ SpillPool    │────┘
                │ [InN→OutM]   │
                └──────────────┘

Each SpillPool maintains FIFO order for its (input, output) pair.
See `RepartitionBatch` for details on the memory/spill decision logic.

Footnote

The “Exchange Operator” was first described in the 1989 paper Encapsulation of parallelism in the Volcano query processing system Paper which uses the term “Exchange” for the concept of repartitioning data across threads.

Implementations

impl RepartitionExec

pub fn input(&self) -> &Arc<dyn ExecutionPlan>

Input execution plan

pub fn partitioning(&self) -> &Partitioning

Partitioning scheme to use

pub fn preserve_order(&self) -> bool

Get preserve_order flag of the RepartitionExec true means SortPreservingRepartitionExec, false means RepartitionExec

pub fn name(&self) -> &str

Get name used to display this Exec

impl RepartitionExec

pub fn try_new( input: Arc<dyn ExecutionPlan>, partitioning: Partitioning, ) -> Result<Self>

Create a new RepartitionExec, that produces output partitioning, and does not preserve the order of the input (see Self::with_preserve_order for more details)

pub fn with_preserve_order(self) -> Self

Specify if this repartitioning operation should preserve the order of rows from its input when producing output. Preserving order is more expensive at runtime, so should only be set if the output of this operator can take advantage of it.

If the input is not ordered, or has only one partition, this is a no op, and the node remains a RepartitionExec.

Trait Implementations

impl Clone for RepartitionExec

fn clone(&self) -> RepartitionExec

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for RepartitionExec

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl DisplayAs for RepartitionExec

fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter<'_>) -> Result

Format according to DisplayFormatType, used when verbose representation looks different from the default one

impl ExecutionPlan for RepartitionExec

fn as_any(&self) -> &dyn Any

Return a reference to Any that can be used for downcasting

fn name(&self) -> &'static str

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

fn properties(&self) -> &PlanProperties

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

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 with_new_children( self: Arc<Self>, children: Vec<Arc<dyn ExecutionPlan>>, ) -> Result<Arc<dyn ExecutionPlan>>

Returns a new ExecutionPlan where all existing children were replaced by the children, in order

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

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

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

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

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

Begin execution of partition, returning a Stream of RecordBatches.

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

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

fn statistics(&self) -> Result<Statistics>

👎Deprecated since 48.0.0: Use partition_statistics method instead

Returns statistics for this ExecutionPlan node. If statistics are not available, should return Statistics::new_unknown (the default), not an error.

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

Returns statistics for a specific partition of this ExecutionPlan node. If statistics are not available, should return Statistics::new_unknown (the default), not an error. If partition is None, it returns statistics for the entire plan.

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.

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:

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.

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.

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

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

Specifies the data distribution requirements for all the children for this ExecutionPlan, By default it’s [Distribution::UnspecifiedDistribution] for each child,

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

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

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

Reset any internal state within this ExecutionPlan.

fn supports_limit_pushdown(&self) -> bool

Returns true if a limit can be safely pushed down through this ExecutionPlan 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.

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

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

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.