Struct CoalesceRule 

pub struct CoalesceRule { /* private fields */ }

Merges partitions whose memory_bytes falls below min_partition_bytes.

When coalescing is beneficial (i.e. the advised group count is smaller than the current partition count), apply rewrites the physical plan by appending a NodeOp::CoalescePartitions node that signals downstream operators to merge the output into target_partitions partitions.

Implementations

impl CoalesceRule

pub fn new(min_partition_bytes: u64) -> Self

Create a new CoalesceRule with the given minimum partition byte threshold.

Uses the default target_partition_bytes of 128 MiB and no parallelism floor; see Self::with_min_partitions.

pub fn with_target_partition_bytes(self, target_partition_bytes: u64) -> Self

Set a custom target_partition_bytes (bytes per merged output partition).

pub fn target_partition_bytes(&self) -> u64

Return the configured target_partition_bytes.

pub fn with_min_partitions(self, min_partitions: usize) -> Self

Never coalesce below min_partitions partitions.

Sizing partitions purely by bytes answers “how big should a partition be” and never asks “how many workers are there”. A stage whose whole output is under target_partition_bytes collapses to a single group, so it runs as one task on one core — measured live on TPC-H q2 at SF100, where four stages coalesced to 1 partition and the cluster sat at one busy core per executor with eight of nine slots idle. Bytes were small; the work over them was not, and coalescing cannot see that.

Callers pass the live slot count so the floor tracks the actual cluster. This mirrors Spark’s coalescePartitions.parallelismFirst, which shrinks the advisory partition size for the same reason.

The floor is advisory in one direction only: it never raises the partition count above what the stage already has, because coalescing may only merge.

pub fn min_partitions(&self) -> usize

Return the configured parallelism floor.

pub fn advise(&self, stats: &[RuntimeStats]) -> CoalesceAdvice

Compute coalesce advice from per-partition stats, without modifying the plan.

Partitions are sorted by memory_bytes (ascending) before grouping so that all small partitions cluster together regardless of their original execution order. Without sorting, a large partition sitting between two small ones would prevent them from coalescing (Spark’s AQE sorts before coalescing for the same reason). Each group of small partitions is capped at target_partition_bytes. Large partitions are always singleton groups.

Each group contains the original partition indices (not sorted indices), so callers can map groups back to the original execution order.

Example: [small(0), big(1), small(2)] → [[0,2], [1]] (2 groups) vs. the old consecutive-only approach: [[0], [1], [2]] (3 groups, no gain)

Trait Implementations

impl AqeRule for CoalesceRule

fn apply( &self, plan: &PhysicalPlan, stats: &[RuntimeStats], ) -> Option<PhysicalPlan>

Compute coalesce advice and, when beneficial, rewrite the plan.

When advise() produces fewer groups than the current partition count, stamps coalesced_partition_count on the plan and appends a NodeOp::CoalescePartitions node carrying the computed target count.

fn name(&self) -> &str

Short, stable rule name used in explain and diagnostics output.