[Ideas] Anser: Folding a segment's parallel workers stat before the coordinator sees them

[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!