Struct datafusion::physical_plan::repartition::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 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 RepartionExec 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   ;
     ╲               ╱           ╲               ╱
      '─.         ,─'             '─.         ,─'
         `───────'                   `───────'

Output Ordering

No guarantees are made about the order of the resulting partitions unless preserve_order is set.

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 RepartitionExecutor true means SortPreservingRepartitionExec, false means RepartitionExec

pub fn name(&self) -> &str

Get name of the Executor

impl RepartitionExec

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

Create a new RepartitionExec

pub fn with_preserve_order(self, preserve_order: bool) -> Self

Set Order preserving flag

Trait Implementations

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 schema(&self) -> SchemaRef

Get the schema for this execution plan

fn unbounded_output(&self, children: &[bool]) -> Result<bool>

Specifies whether this plan generates an infinite stream of records. If the plan does not support pipelining, but its input(s) are infinite, returns an error to indicate this.

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

Get a list of child execution plans that provide the input for this plan. The returned list will be empty for leaf nodes, 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 plan where all children were replaced by new plans.

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

Specifies whether the operator benefits from increased parallelization at its input for each child. If set to true, this indicates that the operator would benefit from partitioning its corresponding child (and thus from more parallelism). For operators that do very little work the overhead of extra parallelism may outweigh any benefits

fn output_partitioning(&self) -> Partitioning

Specifies the output partitioning scheme of this plan

fn output_ordering(&self) -> Option<&[PhysicalSortExpr]>

If the output of this operator within each partition is sorted, returns Some(keys) with the description of how it was sorted.

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

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

fn equivalence_properties(&self) -> EquivalenceProperties

Get the EquivalenceProperties within the plan

fn ordering_equivalence_properties(&self) -> OrderingEquivalenceProperties

Get the OrderingEquivalenceProperties within the plan

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

creates an iterator

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

Return a snapshot of the set of Metrics for this ExecutionPlan.

fn statistics(&self) -> Statistics

Returns the global output statistics for this ExecutionPlan node.

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

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

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

Specifies the ordering requirements for all of the children For each child, it’s the local ordering requirement within each partition rather than the global ordering

fn file_scan_config(&self) -> Option<&FileScanConfig>

Returns the FileScanConfig in case this is a data source scanning execution plan or None otherwise.