GitHub user leborchuk created a discussion: [Ideas] Anser: Folding a segment's parallel workers stat before the coordinator sees them
### Description # Folding a segment's workers before the coordinator sees them Status: idea, not implemented. No wire-protocol change. Original PR with Anser - https://github.com/apache/cloudberry/pull/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! GitHub link: https://github.com/apache/cloudberry/discussions/2021 ---- 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]
