[Ideas] Adaptive join: correct a wrong Broadcast decision at runtime

[leborchuk]

Description

The planner picks Broadcast Motion for the inner side of a join when it
estimates that side is small. When the 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 (gpcontrib/anser/) already establishes a rendezvous at exactly the right
moment: the build side is fully scanned before probing starts, producers
publish, the coordinator aggregates, consumers receive. This issue is about
using that channel to carry cardinality instead of (or as well as) a bloom
filter, and deciding — with numbers — when the difference is large enough to act
on.

The requirement that drives the whole design: the switch boundary must be a
region where one algorithm is unambiguously better. Not "they are within 10%
and the trend favours one" — a zone where switching is obviously right, with a
deliberate no-switch band around it.

What we can reuse

An Anser channel is a palloc'd accumulator in the coordinator backend running
the query, fed by parts arriving over the dispatch connection and folded by
anser_disp_apply_part() (src/anserdispatch.c:302). Only the fold is
bloom-specific — AnserBloomFoldPartInPlace() (src/anserfilter.c:126) checks
the ABF1 magic. 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 / Misra-Gries top-K	mergeable summary union	skew (milestone 3)

So milestone 1 needs a second payload type, not a new transport. The wire
already carries a kind and a flags field (ANSER_WIRE_KIND_PART,
ANSER_WIRE_F_CANCELLED in include/ansersideband.h:59-63) and the parser
takes fields positionally, so adding a payload-type tag plus a fold callback is
a contained change — bump ANSER_WIRE_TAG if the header layout changes. Design
that generalization deliberately, because milestone 3 depends on it.

One thing already learned the hard way. The coordinator originally took
expected_parts from the first part that arrived, and a part claiming a
different count could lower it — completing a channel early and delivering a
filter missing another segment's keys. That is a false negative, i.e. silently
dropped join rows. It now takes the maximum any producer claims and logs
disagreement (anser_disp_apply_part()). Any new payload type inherits that
hazard: a merge that completes early is a correctness bug, not a performance
bug.

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 (scanned from a
table distributed by that key, or a motion earlier in the plan already put 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 note.

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

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
(src/backend/utils/misc/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.

Required experiments

Use the Level-2 harness from the bloom-performance issue (bench_build /
bench_probe), with the inner side made deliberately misestimated — e.g.
ANALYZE, then insert 100× more rows without re-analyzing, so the planner still
believes the old estimate.

Sweep, and plot broadcast vs redistribute as two curves:

Variable	Values
actual inner rows	10 K, 100 K, 1 M, 10 M, 100 M
misestimation factor	1×, 10×, 100× (estimate ÷ actual)
segments N	3, 8, 24 (or whatever the cluster allows — N is in the formula, so at least two values)
inner row width	32 B, 512 B
work_mem	default, and one value that moves the spill knee

Per point record: query time (median of 5), bytes moved, peak memory per
segment, whether the hash join spilled (EXPLAIN ANALYZE batches / workfile
lines), and the Anser publish/deliver timings (anser.debug gives these today).

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 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
  (anser_disp_push(), src/anserdispatch.c:394).
• 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.

A note on timing. The measured Anser exchange on a 3-segment cluster is
~34 ms for a 1 MB payload, of which ~60% is the coordinator picking parts up one
at a time. A row count is a handful of bytes, so the payload cost vanishes but
the rendezvous latency does not — budget on the order of tens of milliseconds
for the round trip, and check that against the saving you are chasing. A switch
that saves 20 ms is not worth a 30 ms barrier.

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 / Misra-Gries top-K, or a Count-Min sketch); the coordinator
merges them — both merge by addition, so they fit the 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 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 the bloom-performance issue
  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.
• Setup: shared_preload_libraries='anser', anser.enable=on. No
  CREATE EXTENSION needed.

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. See the expected_parts history above
  for how easily this happens.
• 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.
  Changing a planner constant is a different (and much cheaper) change; if your
  measurements show the default 100 000 is simply wrong, that is a valuable
  one-line finding — report it separately rather than folding it in here.
• 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

#1942

Are you willing to submit a PR?

• Yes I am willing to submit a PR!