Struct deltalake::datafusion::physical_optimizer::enforce_distribution::EnforceDistribution

pub struct EnforceDistribution {}

The EnforceDistribution rule ensures that distribution requirements are met. In doing so, this rule will increase the parallelism in the plan by introducing repartitioning operators to the physical plan.

For example, given an input such as:

┌─────────────────────────────────┐
│                                 │
│          ExecutionPlan          │
│                                 │
└─────────────────────────────────┘
            ▲         ▲
            │         │
      ┌─────┘         └─────┐
      │                     │
      │                     │
      │                     │
┌───────────┐         ┌───────────┐
│           │         │           │
│ batch A1  │         │ batch B1  │
│           │         │           │
├───────────┤         ├───────────┤
│           │         │           │
│ batch A2  │         │ batch B2  │
│           │         │           │
├───────────┤         ├───────────┤
│           │         │           │
│ batch A3  │         │ batch B3  │
│           │         │           │
└───────────┘         └───────────┘

     Input                 Input
       A                     B

This rule will attempt to add a RepartitionExec to increase parallelism (to 3, in this case) and create the following arrangement:

    ┌─────────────────────────────────┐
    │                                 │
    │          ExecutionPlan          │
    │                                 │
    └─────────────────────────────────┘
              ▲      ▲       ▲            Input now has 3
              │      │       │             partitions
      ┌───────┘      │       └───────┐
      │              │               │
      │              │               │
┌───────────┐  ┌───────────┐   ┌───────────┐
│           │  │           │   │           │
│ batch A1  │  │ batch A3  │   │ batch B3  │
│           │  │           │   │           │
├───────────┤  ├───────────┤   ├───────────┤
│           │  │           │   │           │
│ batch B2  │  │ batch B1  │   │ batch A2  │
│           │  │           │   │           │
└───────────┘  └───────────┘   └───────────┘
      ▲              ▲               ▲
      │              │               │
      └─────────┐    │    ┌──────────┘
                │    │    │
                │    │    │
    ┌─────────────────────────────────┐   batches are
    │       RepartitionExec(3)        │   repartitioned
    │           RoundRobin            │
    │                                 │
    └─────────────────────────────────┘
                ▲         ▲
                │         │
          ┌─────┘         └─────┐
          │                     │
          │                     │
          │                     │
    ┌───────────┐         ┌───────────┐
    │           │         │           │
    │ batch A1  │         │ batch B1  │
    │           │         │           │
    ├───────────┤         ├───────────┤
    │           │         │           │
    │ batch A2  │         │ batch B2  │
    │           │         │           │
    ├───────────┤         ├───────────┤
    │           │         │           │
    │ batch A3  │         │ batch B3  │
    │           │         │           │
    └───────────┘         └───────────┘


     Input                 Input
       A                     B

The EnforceDistribution rule

• is idempotent; i.e. it can be applied multiple times, each time producing the same result.
• always produces a valid plan in terms of distribution requirements. Its input plan can be valid or invalid with respect to distribution requirements, but the output plan will always be valid.
• produces a valid plan in terms of ordering requirements, if its input is a valid plan in terms of ordering requirements. If the input plan is invalid, this rule does not attempt to fix it as doing so is the responsibility of the EnforceSorting rule.

Note that distribution requirements are met in the strictest way. This may result in more than strictly necessary RepartitionExecs in the plan, but meeting the requirements in the strictest way may help avoid possible data skew in joins.

For example for a hash join with keys (a, b, c), the required Distribution(a, b, c) can be satisfied by several alternative partitioning ways: (a, b, c), (a, b), (a, c), (b, c), (a), (b), (c) and ( ).

This rule only chooses the exact match and satisfies the Distribution(a, b, c) by a HashPartition(a, b, c).

Implementations

impl EnforceDistribution

pub fn new() -> EnforceDistribution

Trait Implementations

impl Default for EnforceDistribution

fn default() -> EnforceDistribution

Returns the “default value” for a type.

impl PhysicalOptimizerRule for EnforceDistribution

fn optimize( &self, plan: Arc<dyn ExecutionPlan>, config: &ConfigOptions ) -> Result<Arc<dyn ExecutionPlan>, DataFusionError>

Rewrite plan to an optimized form

fn name(&self) -> &str

A human readable name for this optimizer rule

fn schema_check(&self) -> bool

A flag to indicate whether the physical planner should valid the rule will not change the schema of the plan after the rewriting. Some of the optimization rules might change the nullable properties of the schema and should disable the schema check.