GitHub user leborchuk created a discussion: [Ideas] Adaptive join: correct a
wrong Broadcast decision at runtime
### Description
The planner picks `Broadcast Motion` for the inner side when it estimates the
inner side is small. When that estimate is wrong by 10× or 100×, every segment
receives *all* inner rows and builds *the entire* hash table — N× the network
traffic, N× the build CPU, N× the memory, and often a spill to disk that
redistributing would have avoided. The query does not fail; it just runs far
slower than the alternative plan, and today nothing notices.
Anser could be used to switch from broadcast to redistribute (or vice versa -
to discuss).
**The user-facing requirement, which drives the whole design:**
boundary must be a region where one algorithm is *unambiguously* better. Not
"they are within 10% and the trend favours one" — a zone where
obviously right, with a deliberate no-switch band around it.
## What we can reuse
Anser's channel payload is opaque bytes; only the *fold* is bloom-specific
(`AnserBloomFoldPartInPlace()` checks the `ABF1` magic in
`gpcontrib/anser/src/anserfilter.c`). Statistics fold just as easily:
| Payload | Fold operation | Use |
| --- | --- | --- |
| bloom filter (today) | bitwise OR | pruning |
| row count / byte count | integer addition | this issue, milestone 1 |
| Count-Min sketch | element-wise addition | skew (milestone 3) |
| Space-Saving top-K | mergeable summary union | skew (milestone 3) |
So milestone 1 needs a second payload type with its own magic a
not a new transport. Design that generalization deliberately — a payload-type
tag in the header plus a fold callback — because milestone 3 de
## Milestone 0 — feasibility and correctness (do this first, do not skip)
**Before any code or benchmarks, answer this in a design note.** It is entirely
possible that the honest answer to part of this issue is "not feasible without
planner changes", and finding that out in week 1 is a success.
Facts to start from:
- `MOTIONTYPE_HASH` vs `MOTIONTYPE_BROADCAST`
(`src/include/nodes/plannodes.h:1880-1886`)
differ, at the sender, only in **how each tuple's destination is chosen**. The
receiving slice is the same set of processes either way, and the gang is
already running. So flipping the sender's routing is not obviously impossible.
- **But the join above the motion is only correct if both inputs are placed
compatibly.** Broadcast-inner works with the outer left wherever it is.
Redistribute-inner requires the outer to *also* be hashed on the join key. If
the plan broadcast the inner precisely so the outer would not need a motion,
you cannot flip one side in isolation — you would silently produce **wrong
results**, which is far worse than being slow.
That yields one case where a runtime switch is *provably* safe:
> **The outer side is already distributed on the join key** (it was scanned from
> a table distributed by that key, or a motion earlier in the p
> it there). Then broadcast-inner and redistribute-inner are *both* correct, and
> the choice is purely about cost.
**Milestone 0 deliverable — a note answering:**
1. In which plan shapes is the outer side already hash-distributed on the join
key? How often does that happen in TPC-DS at SF100? (Count it — this bounds
the value of the whole issue.)
2. Can all senders in a slice agree on a routing switch, and what synchronizes
them? (They must agree, or tuples for the same key land on different
segments and rows are lost.)
3. What happens to inner tuples already sent before the switch? Does the inner
side need re-scanning, and if so what does that cost?
4. Where would the decision be taken and enforced — in `nodeMotion.c` at the
sender, or by choosing between two pre-planned alternatives at slice start?
5. For the general (unsafe) case, which is more realistic: planning both
alternatives and choosing at execution start, or aborting and re-dispatching
with corrected cardinality (what Spark AQE does at shuffle boundaries)?
Do not begin milestone 2 until a reviewer has agreed with this
## Milestone 1 — the cost model and where the boundary is (the core research)
### What the system already believes
The Postgres-planner motion cost (`cdbpath_cost_motion()`,
`src/backend/cdb/cdbpath.c`):
```c
cost_per_row = (gp_motion_cost_per_row > 0) ? gp_motion_cost_per_row
: 2 * cpu_tuple_cost; /* = 0.02 */
motioncost = cost_per_row * 0.5 * (sendrows + recvrows);
```
With R inner rows and N segments: redistribute has `recvrows ≈ R`, broadcast has
`recvrows = R × N`. So the model says broadcast-inner beats redistributing both
sides when, roughly:
```
0.5 · R_in · (1 + N) < R_in + R_out
⇒ R_out > R_in · (N − 1) / 2
```
i.e. at N=3 the outer must merely exceed the inner; at N=48 it must exceed it
**23×**. ORCA carries a separate, blunter rule:
`optimizer_penalize_broadcast_threshold` = **100 000 rows** by default
(`guc_gp.c:4531`).
**Your first job is to find out whether this linear model is true.** It has no
term for any of the following, and at least two of them are non-linear.
### The four cost components — measure each
| # | Component | Broadcast | Redistribute | Why it may dominate |
| --- | --- | --- | --- | --- |
| 1 | bytes on the wire | R × N received | R received | linear, but
interconnect flow control is not |
| 2 | receive-side CPU (deserialize) | R × N | R | linear |
| 3 | **hash table build** | every segment builds all R rows | each builds R/N
| N× CPU *and* N× memory |
| 4 | **spill to disk** | R × width vs `work_mem` on *every* segment | (R/N) ×
width | **step function, 2-10×** |
### The strongest candidate for an unambiguous boundary
Component 4 is a **step function**, and steps are exactly what "no doubt" looks
like:
> If broadcasting makes the hash table exceed `work_mem` and spill, while
> redistributing keeps it in memory, redistribute wins by a large, verifiable
> margin — regardless of the row count.
That reframes the trigger from "how many rows" to "which side of the spill
boundary each alternative lands on", with the row count as the input to that
test. ORCA's flat 100 000 rows is a crude proxy for the same thing that ignores
row width, `work_mem`, and segment count.
**Verify this before building on it.** Measure query time for broadcast and
redistribute while sweeping inner rows across the point where broadcast starts
spilling, at fixed `work_mem`. If the curve shows a sharp knee, that is your
boundary and you can defend it. If it degrades smoothly, say so — the
recommendation then has to come from the K-factor rule below alone.
### The switch rule you must propose
Express it as a **rule with a deliberate dead zone**, and state every constant
with the measurement that produced it:
```
switch broadcast → redistribute iff
cost_broadcast > K × cost_redistribute (K ≥ 2, measured)
AND estimated absolute saving > S (e.g. > 1 s or > 1 GB
moved)
AND cost_of_switching_now < estimated saving
```
- **K is the no-doubt factor.** Within K× we do not switch, on purpose. Justify
the value you pick from the spread of your own measurements: K must be larger
than your measurement noise by a comfortable margin.
- **S prevents churn** on queries too small for any of this to matter.
- **The switch itself is not free** — item 3 of the milestone-0 note. If the
inner side must be re-scanned, the switch cost is a full inner scan, and the
threshold must exceed it. Include this term with a measured value, not a
guess.
Deliver the rule as: formula, every constant with its measurement, the dead
zone drawn on the crossover plot, and the predicted vs actual outcome for at
least 10 points (does the rule fire when it should, and stay quiet when it
should not?). **Report false positives** — cases where the rule would switch and
be wrong. Those matter more than the wins.
## Milestone 2 — implement the safe case
Only the case established in milestone 0: **outer already hash-distributed on
the join key**. Even then:
- Anser publishes the actual inner row count (and byte count) per segment; the
coordinator sums them; the decision is taken once, centrally, and delivered to
all senders through the existing consumer delivery path.
- If any segment cannot be reached or the deadline expires, **keep the planned
motion** — the existing fail-open discipline. Never let an adaptation failure
change results or raise an error.
- `EXPLAIN ANALYZE` must show that the switch happened, what the estimate was,
and what the actual was. An invisible adaptation is undebuggable.
- Behind a GUC (`anser.adaptive_join`), default **off**.
## Milestone 3 — skewed join (likely its own issue)
Once cardinality feedback works, the same channel can carry a skew profile.
**Detection.** Each segment builds a mergeable heavy-hitter summary of the join
key (Space-Saving top-K, or a Count-Min sketch); the coordinator
merges them — both merge by addition, so they fit Anser's fold model directly.
A key is "heavy" when its frequency exceeds roughly `total_rows / N` (one
segment would receive more than its fair share); measure the right multiple.
**Routing.** Heavy keys broadcast, everything else redistributes.
`MOTIONTYPE_EXPLICIT` (destination taken from a column) already provides
per-tuple routing and is worth studying as the mechanism.
**Correctness argument you must write down before coding:** for a hash join,
if the *build* rows of a heavy key are broadcast to every segment while the
*probe* rows of that key are spread arbitrarily, each probe row still meets
every build row for its key, so no match is lost and none is duplicated. Prove
the same for the non-heavy keys and for outer joins (the null-extended side is
where this usually breaks).
**Threshold**, same discipline as milestone 1: only act when skew is extreme
enough that the imbalance is unambiguous — e.g. the heaviest key alone exceeds
some multiple of the per-segment average — and quantify the cost of being wrong.
## Benchmarks
- **Simple benchmark:** the Issue 2 Level-2 harness with the misestimation knob
above, plus a skew generator for milestone 3 (e.g. a Zipf key distribution;
state the parameter, and include a case where one key is 30% of all rows).
- **TPC-DS at SF100 minimum**, reported exactly as Issue 2 requires (injected
subset in detail, non-injected regression check, geomean). TPC-DS is
especially relevant here: several queries have known cardinality
misestimations, and **count how many queries broadcast an inner side that
turns out large** — that count is the business case for this whole issue, so
measure it early and report it even before any implementation exists.
- Report from both optimizers (`optimizer=on`/`off`) — the motion decisions,
and their mistakes, differ.
## Definition of done
- [ ] Milestone 0 design note, reviewed and agreed, including the TPC-DS count
of how often the safe case occurs.
- [ ] Crossover measurements for all four cost components, with plots.
- [ ] A switch rule with every constant justified by measurement, an explicit
dead zone, the switch-cost term, and predicted-vs-actual for ≥10 points
including false positives.
- [ ] Correctness argument written down before any routing change.
- [ ] If implemented: GUC-gated and off by default, visible in `EXPLAIN
ANALYZE`,
fail-open on any adaptation failure, results verified identical to the
non-adaptive plan on the full regression suite plus TPC-DS row counts.
## Out of scope / traps
- **Do not change results, ever.** A performance feature that returns a
different row set is a data-corruption bug. Every routing change needs the
correctness argument first, and a test that compares full result sets (not
counts) against the non-adaptive plan.
- Do not start with the general case (both sides needing motion changes). It
requires planner-level alternatives or re-dispatch and is a much bigger
project; scope it only after milestone 2 works.
- Do not tune `optimizer_penalize_broadcast_threshold` and call it adaptive.
We do not want to replan query.
- Do not adapt on a single segment's statistics. Skew means segments disagree;
the decision has to be made from the merged picture at the coordinator.
### Use case/motivation
_No response_
### Related issues
_No response_
### Are you willing to submit a PR?
- [ ] Yes I am willing to submit a PR!
GitHub link: https://github.com/apache/cloudberry/discussions/1962
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
