#2021 [Ideas] Anser: Folding a segment's parallel workers stat before the coordinator sees them New discussion created by leborchuk Description Folding a segment's workers before the coordinator sees them Status: idea, not implemented. No wire-protocol change. Original PR with Anser - #1942 The problem A parallel slice runs

numsegments × parallel_workers

processes and every one of them publishes its own part. The coordinator folds them all and then returns the merged payload to every consumer, one blocking write at a time. Both halves scale with the process count rather than the segment count, and the payload does not shrink along the way: a bloom union is a bitwise OR, so each part is a whole bitset. Measured on 4 segments with a 1 MB filter, and extrapolated to 128 segments with a 64 MB one (the ceiling

anser.max_info_size

allows): | Cluster | parallel_workers | Parts folded | Deliveries | Through the coordinator | | -------------- | ---------------- | ------------ | ---------- | ----------------------- | | 4 seg, 1 MB | 1 | 4 | 4 | 9.6 MB | | 4 seg, 1 MB | 4 | 16 | 16 | 38 MB | | 128 seg, 64 MB | 1 | 128 | 128 | 18.7 GiB | | 128 seg, 64 MB | 4 | 512 | 512 | 74.7 GiB | | 128 seg, 64 MB | 8 | 1024 | 1024 | 149.3 GiB | Per join, over one coordinator NIC, single-threaded, with a base64 decode and a CRC per part. At 4 segments and 16 workers the fan-out already took 1140 ms, which overran the old 1 s consumer deadline and left every consumer unfiltered. Raising the deadline (now 100 s by default) makes the filter arrive but does not make it cheap: the first subscriber is served in under a millisecond and the last over a second later. The idea The workers of one segment are processes on one host. They do not need the coordinator to combine their parts with each other — only to combine them with other segments'. So fold locally first, and let one process per segment speak to the coordinator on behalf of the rest.

now:      16 workers ──16 parts──> QD ──16 pushes──> 16 workers
proposed: 16 workers ──local fold──> 4 leaders ──4 parts──> QD
                                     ──4 pushes──> 4 leaders ──local──> 16 workers

Effect on the numbers above: parts folded and deliveries both drop by a factor of

parallel_workers

, back to the rows where it equals 1. The 128-segment, 8-worker case falls from 149.3 GiB to 18.7 GiB — the coordinator stops paying for parallelism it gains nothing from. What it does not fix: the remaining

numsegments

parts still cross the coordinator, so a 128-segment cluster still moves ~19 GiB per join for a 64 MB filter. A full fix is an all-reduce over the interconnect, where per-node traffic is ~2 × payload regardless of cluster size. Local folding is the cheaper first step and is a prerequisite for that anyway — the interconnect exchange wants one participant per segment, not one per worker. Why this fits the existing design • The wire protocol does not change. A leader publishes the same

ANSER_WIRE_KIND_PART

message it publishes now, and the coordinator still folds N parts without caring what N is. Only

ANSER_RF_PRIV_N_PRODUCERS

gets a different value:

numsegments

instead of

numsegments × parallel_workers

. •

fold()

already exists in

AnserPayloadOps

(

anserpayload.c

) and is the same bitwise OR the coordinator uses. The local fold calls the same function on the same payload type; nothing type-specific is duplicated. • The count stays authoritative. A gang is all-or-nothing —

cdbgang_createGang_async()

either returns a full gang or raises — so

numsegments

is exactly the number of leaders that will run, just as

numsegments × parallel_workers

is exactly the number of processes today. What has to be built 1. A per-segment rendezvous. The workers of a slice need somewhere to deposit a part and somewhere to read the merged result. A DSM segment keyed by

(gp_session_id, gp_command_count, condition_id)

is the obvious shape; the QEs of one gang share a host but not a

ParallelContext

, so nothing existing can be reused directly. 2. A local barrier. The leader publishes once its workers have all deposited. This is where the design earns its keep or fails: a worker that never arrives — squelched, or erroring — must not strand the segment. The barrier needs the same fail-open discipline the consumer deadline has, and the leader should publish what it has rather than wait indefinitely. 3. Who is the leader. One process per segment per slice has to be distinguishable.

Gp_is_writer

is not it (all the QEs in the log are writers); this needs an explicit answer. 4. Local redistribution. After the leader receives the merged filter, its workers need it. Same shared segment, in the other direction. Risks • A barrier is a new way to hang. Today a lost part costs one part; with a barrier it can cost a whole segment's progress. Every wait must have a deadline and a fail-open path, and the leader must never block on a worker that has already finished or died. • Shared memory is not transaction-aware. The segment has to be released on abort as well as on completion — the same problem

dl_resource.c

solves for datalake_fdw with a resource-owner callback, and

comm/pax_resource.cc

for PAX. • It only moves the knee. 128 segments still overwhelm a star topology. If the target is large clusters with large filters, this buys time; it is not the destination. Use case/motivation No response Related issues No response Are you willing to submit a PR? • Yes I am willing to submit a PR! apache/cloudberry