#1962 [Ideas] Adaptive join: correct a wrong Broadcast decision at runtime New discussion created by leborchuk 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

): 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 inne… apache/cloudberry