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

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.

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 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 reparititoning 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 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 ‘ParquetExec’.

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>

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

fn static_name() -> &'static str

where Self: Sized,

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

fn schema(&self) -> SchemaRef

Get the schema for this execution plan

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<Vec<PhysicalSortRequirement>>>

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

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