Lavkesh/network shuffle experiments

[ologlogn]

optimize_shuffle_partitioning_ratio

Workload Guide

What the ratio controls

At each shuffle boundary, the planner applies a clamp formula to determine how many tasks
the downstream stage should have:

• ratio >= 1.0: clamp(cardinality, p, ceil(p * ratio))
  • Lifts small cardinality up to p (enabling optimized 1:1 mapping)
  • Caps large cardinality at p * ratio (limiting N×M fanout)
  • ratio=1 always produces exactly p tasks → fully optimized every boundary
  • ratio=2 allows up to 2p tasks → moderate parallelism with controlled fanout
• ratio < 1.0: cardinality.min(ceil(p * ratio))
  • Scales task count below p — useful for reducing parallelism on light workloads

Where p is the output_partitions of the NetworkShuffleExec at the boundary (determined
by target_partitions session config), and cardinality is the inherited task count from
upstream stages, adjusted by cardinality-effect factors from reducing operations.

The use_optimized flag on NetworkShuffleExec is set per-boundary when task_count == p.
When optimized, each consumer task reads exactly one partition stream (1:1 mapping) instead
of reading all N upstream streams and filtering by hash.

---

Ratio selection by workload type

ratio=1 (default, fully optimized everywhere)

Best for:

• Uniform data distribution where parallelism is already well-matched to target_partitions
• Light or medium workloads where network coordination overhead is significant relative to compute
• Deep pipelines (5+ stages) — ratio=1 keeps stream counts flat at p per boundary,
  preventing compounding fanout across many stages
• Cases where you want predictable, low-overhead behavior as a baseline

What it does: every shuffle boundary becomes optimized (1:1 task-to-partition mapping).
Stream counts stay constant at p across all stages regardless of source cardinality.

ratio=2–3 (moderate, recommended general-purpose default)

Best for:

• Mixed workloads: heavy source stages followed by cardinality-reducing aggregations
• Queries where the source stage benefits from full parallelism but downstream stages
  are cheaper because aggregation has already reduced the data volume
• Production environments where you want parallelism headroom without aggressive tuning
• Situations where data distribution is mostly uniform but occasionally uneven

What it does: source and early stages can run at up to 2–3× the partition count, preserving
parallelism for heavy compute. Once a cardinality-reducing operation appears (partial
aggregate, filter with high selectivity), subsequent stages naturally inherit smaller
task counts, making the high ratio less impactful downstream.

ratio=4+ (high parallelism)

Best for:

• Massive, uniformly distributed data with few pipeline stages
• Workloads where every stage is compute-bound and data is evenly split across workers
• Clusters with many workers where you want to utilize all available parallelism

Avoid when:

• Pipeline has many stages — stream count compounds per boundary. With N=15 tasks and p=4
  partitions, each boundary has 60 streams. Across 5 stages this means each worker
  simultaneously maintains many gRPC connections, increasing memory pressure proportional
  to record_batch_buffer_size × connections.
• Source data is skewed — see section below.

---

When ratio does not help: skewed data

The ratio controls how many consumers drain upstream tasks. It does not control how evenly
data is distributed across those upstream tasks.

If one upstream task processes 167× more data than another (as seen in EventStore queries
over 6-month time ranges, where certain months have 27M events and others have near-zero),
every consumer stage will exhibit HOL (head-of-line) blocking — consumers finish their light
tasks quickly but then sit idle waiting for the single heavy upstream task to complete.

This manifests in plans as:

• High network_latency_sum on NetworkShuffleExec consumers (e.g., 180s, 69s, 236s)
• send_time on the heavy source task orders of magnitude above the others
• fetch_time uniformly high across all downstream tasks (all waiting for the same bottleneck)

Ratio cannot fix this. More consumers just means more workers waiting on the same slow
upstream task. The solution is a bytes-based or rows-based task estimator at the source stage
that splits heavy time-range partitions into multiple tasks before the first shuffle boundary.

---

Multi-stage cardinality propagation

In a pipeline with multiple stages, the ratio affects early stages most significantly.
Later stages self-correct because:

1. Cardinality-reducing operations (partial aggregates, filters) scale down the inherited
   task count via the cardinality_task_count_factor configuration.
2. Once the task count drops below p, ratio=1 behavior kicks in anyway (lifting back to p).
3. Final stages producing small output (e.g., LIMIT 10000, top-N window) end up with
   p tasks regardless of what happened upstream.

This means ratio matters most for the first 1–2 heavy stages. Tuning ratio for the
whole query is really tuning for the source-side parallelism. The rest of the pipeline
adjusts naturally.

Typical multi-stage behavior with ratio=2 (p=4, source cardinality=15):

Stage 1: 15 tasks  (source, cardinality=15)
Stage 2: 15 tasks  (join/filter, cardinality inherited)
Stage 3: 8 tasks   (clamp(15, 4, 8) = 8, partial aggregate)
Stage 4: 4 tasks   (cardinality reduced by aggregate, matches p → optimized)
Stage 5: 4 tasks   (final output, optimized 1:1)

With ratio=4:

Stage 1: 15 tasks  (source, cardinality=15)
Stage 2: 15 tasks  (join/filter, cardinality inherited)
Stage 3: 15 tasks  (clamp(15, 4, 16) = 15, fits within window)
Stage 4: 5 tasks   (cardinality partially reduced, slight reduction from aggregate)
Stage 5: 4 tasks   (optimized, 4==p)

---

Quick reference

Scenario	Recommended ratio
Uniform data, any pipeline depth	1
Uniform data, source is heavy compute	2–3
Massive uniform data, few stages, many workers	4+
Deep pipeline (5+ stages)	1–2
Skewed data (any case)	Fix source estimator; ratio won't help
Light analytic queries	1
Heavy joins followed by aggregation	2–3
Window functions with skewed partition keys	Fix partition key distribution

---

Relationship to N×M fanout

Original behavior without optimize_shuffle_partitioning: every boundary is N×M, where
N is the upstream task count and M is output_partitions. Each consumer reads N streams.

With optimization enabled:

• ratio=1: always 1×M (optimized). M total streams per consumer.
• ratio=r: up to r×M streams per consumer at boundaries where cardinality > p.
• ratio=∞: full N×M restored everywhere (equivalent to no optimization).

The ratio is therefore a continuous knob between fully optimized (low fanout, low overhead,
requires uniform data) and fully N×M (maximum parallelism, maximum stream count, tolerates
any distribution).

For most production workloads, ratio=2 represents a reasonable tradeoff: it preserves
optimized behavior for light-to-medium stages while allowing heavy stages to utilize
more parallelism than a strict 1:1 mapping would permit.