Description

Anser (gpcontrib/anser/) is a runtime pub/sub facility: producer nodes on the
segments publish a bloom filter over a join-build key, the coordinator unions
the parts, and consumer nodes on the segments receive it and prune probe rows.
Everything travels over the dispatch connection the coordinator already holds
open to each segment — there is no shared memory and no background worker. See
gpcontrib/anser/README.md for the architecture before starting.

Today we can see what the filter did (Rows Removed by Bloom Filter in
EXPLAIN ANALYZE) but nothing about what it cost. A consumer blocks until the
coordinator delivers the merged filter; if that wait is expensive the filter is
a net loss, and we currently have no way to tell from a plan.

We know the wait is not negligible. A traced exchange on a 3-segment demo
cluster with a 1 MB filter took 34 ms end to end, and only ~13 ms of that
was work:

The parts were sent within 1.3 ms of each other but folded ~7.2 ms apart —
so roughly 21 ms, 60% of the exchange, is the coordinator picking parts up one
at a time, not doing work (a 1 MB fold is ~0.1 ms, a 1.4 MB base64 decode
~1–2 ms). That pickup latency is gated by the interconnect wait loop, it scales
with segment count, and nothing in EXPLAIN shows it. Attributing it is the
main reason this issue exists.

Goal

Make every wait and every queueing delay in Anser visible in EXPLAIN, on both
sides of the exchange, so a filter that costs more than it saves can be
identified from a plan alone.

Definition of done

• EXPLAIN (ANALYZE, VERBOSE) shows producer publish cost and consumer wait
  time, including for segment-executed nodes.
• The coordinator's side is attributable: per-channel first-part arrival,
  completion, delivery, and per-part fold time.
• A consumer that fails open says why — delivered, cancelled or timed out.
• Regression tests pass and contain no timing-dependent expected values.
• README.md gains a "Metrics" section: every field, its unit, and where it
  is measured.
• No new allocation or locking on the data path; no per-tuple timing calls.
• Core (src/) is untouched.

What to measure

Three vantage points, each with its own clock. Keeping them separate is the
single most important thing to get right — do not subtract a timestamp taken on
a segment from one taken on the coordinator.

The pickup latency that dominates the trace above is between vantage points
(segment send → coordinator fold). Deriving it exactly needs a send timestamp on
the wire, which means comparable clocks — see "Optional: exact pickup latency".

Note this issue does not ask for cluster-wide counters or a
anser.stats() view. The subsystem has no shared memory any more, and it
creates no catalog objects (CREATE EXTENSION is not part of installing it), so
there is nowhere for cross-backend totals to live and no SQL surface to expose
them. Per-query numbers in EXPLAIN are the deliverable. If cross-query
aggregation is wanted later it needs its own design discussion — reintroducing
either shared memory or an extension is a bigger decision than instrumentation.

Implementation

Three changes, in this order. Each is independently reviewable and testable.

Change 1 — per-node numbers in EXPLAIN

• Time the publish in the producer node and the receive in the consumer
  node. Record per node: number of waits, total wait time, longest
  single wait, and for the consumer the outcome.
  • Producer: wrap AnserProducePublishPart(); the accumulators belong in
    AnserBloomProduceScanState (src/anserplanexec.c:80), and the publish
    is triggered from anser_produce_next() (:356) when the child is
    exhausted.
  • Consumer: wrap ExecAnserBloomFilterConsumeSideband(); accumulators in
    AnserBloomConsumeScanState (src/anserplanexec.c:94), driven from
    anser_consume_receive() (:498).
• Print them from anser_produce_explain() (src/anserplanexec.c:435) and
  anser_consume_explain() (:648), gated on es->analyze && es->verbose.
  Use ExplainPropertyInteger / ExplainPropertyFloat so JSON/YAML/XML
  output works for free — never appendStringInfo into the plan text.

The trap you must handle. A field you add to the node state on a segment
does not reach the QD. Read the comment at src/anserplanexec.c above the
InstrCountFiltered1/2 calls in the consumer's exec loop: the existing code
uses those counters deliberately, because only the fixed fields of
CdbExplain_StatInst (src/backend/commands/explain_gp.c:44) travel back. Your
new timers are not in that struct.

The supported escape hatch is the per-node extra text channel, which is how
Hash reports Extra Text: (seg2) Hash chain length ...:

• PlanState.cdbexplainbuf and PlanState.cdbexplainfun
  (src/include/nodes/execnodes.h:1152-1153)
• collected on the segment by cdbexplain_collectExtraText()
  (src/backend/commands/explain_gp.c:1308) and shipped to the QD
• copy the pattern from src/backend/executor/nodeRuntimeFilter.c:174-177 —
  the closest existing analogue, a runtime-filter node doing exactly this

Allocate cdbexplainbuf in *_begin when estate->es_instrument is set,
install a cdbexplainfun that appends your numbers, and the QD shows them as
Extra Text per segment. No core change is needed — if you find yourself
editing anything under src/, stop and re-read this paragraph.

Change 2 — the coordinator's side

The QD merges in the backend running the query, so its numbers can ride out on
the plan rather than into a stats table.

• Add per-channel timing to AnserDispChannel (src/anserdispatch.c):
  first-part arrival, completion, delivery, accumulated fold time, and part
  count. All from one clock, all in memory already owned by the query.
• Surface them on the consumer node's EXPLAIN output — the QD is where
  that node's plan output is assembled, so no transport is needed. One line
  of key=value pairs, matching the shape of the Extra Text lines.
• Use instr_time throughout (src/include/portability/instr_time.h:
  INSTR_TIME_SET_CURRENT, INSTR_TIME_SUBTRACT,
  INSTR_TIME_GET_MICROSEC). Do not use GetCurrentTimestamp() for
  durations.

Accuracy caveats to document in the code. The consumer sleeps in slices
(ANSER_SIDEBAND_POLL_MS = 100 ms in src/ansersideband.c), so measure wall
time across the whole wait call, never by counting loop iterations, and note
that sub-poll-interval waits are quantized. On the coordinator side, note that
fold time and pickup latency are different things and the fold is the small one.

Change 3 — build on the existing trace, don't duplicate it

anser.debug (ANSER_DEBUG(), include/anser.h:78) already logs every step of
the exchange with the sender's identity. Reuse its call sites rather than adding
a parallel set:

• Where a trace line already exists, extend it with the new measurement
  instead of adding a second line.
• Keep the log format one line of key=value pairs — it is grep-and-awk
  material and people already have scripts.
• The README's "Tracing an exchange" section documents the current output;
  update it in the same commit.

Optional: exact pickup latency

The 21 ms above can only be attributed exactly by comparing a segment's send
time with the coordinator's fold time. If you want that:

• Add a send timestamp to the QE→QD wire header (anser1 ... in
  include/ansersideband.h) — there is room, and the parser takes fields
  positionally, so bump ANSER_WIRE_TAG if you change the layout.
• Report it as a separate, clearly-labelled field and document that it is
  meaningless without synchronised clocks across hosts. Do not fold it
  into any other total.

If that is more than you want to take on, skip it: reporting each side's own
durations still narrows the gap to "time spent between publish and fold", which
is the actionable finding.

Testing

• Extend gpcontrib/anser/sql/anser_test.sql (+ expected/anser_test.out).
• Never put a measured time in expected output — the tests would fail
  randomly. Assert properties instead: a counter is > 0, a wait total
  increased after a known round trip, an outcome field says delivered.
  Return booleans from C test helpers (src/anser_test.c) rather than
  printing numbers.
• For the EXPLAIN part, prefer asserting on plan shape — a raw
  EXPLAIN (ANALYZE, VERBOSE) in expected output is unstable. If you must,
  filter it through a query that only checks the property is present.
• Run: make -C gpcontrib/anser install && make -C gpcontrib/anser installcheck
  (the installcheck target arms the cluster itself). Requires
  shared_preload_libraries='anser' and anser.enable=on; no
  CREATE EXTENSION is needed for the subsystem itself.

What I expect from code

EXPLAIN (ANALYZE, VERBOSE) — the normal case

Three segments, anser_rf_build (200 rows) joined to anser_rf_probe
(2000 rows), the filter working as intended:

 Gather Motion 3:1 (slice1; segments: 3) (actual time=38.412..39.104 rows=200 loops=1)
 Output: b.name, p.payload
 -> Hash Join (actual time=35.881..36.402 rows=67 loops=1)
 Output: b.name, p.payload
 Hash Cond: (p.id = b.id)
 -> Custom Scan (Anser Bloom Consumer) (actual time=31.902..33.114 rows=73 loops=1)
 Output: p.id, p.payload
 Bloom Filter Size: 1048576 bytes
 Bloom Filter Stats: memory=1024kB checked=667 rejected=594
 Rows Removed by Bloom Filter: 594
 Anser Waits: 1
 Anser Wait Time: 23.774 ms
 Anser Max Wait: 23.774 ms
 Anser Result: delivered
 Anser Coordinator: parts=3 first=16.840ms complete=31.361ms delivered=31.389ms fold=0.31ms
 Extra Text: (seg0) Anser consumer: waits=1 total=23.774ms max=23.774ms reads=1 result=delivered
 -> Seq Scan on public.anser_rf_probe p (actual time=0.021..0.204 rows=667 loops=1)
 Output: p.id, p.payload
 -> Hash (actual time=2.510..2.511 rows=67 loops=1)
 Output: b.name, b.id
 Buckets: 262144 Batches: 1 Memory Usage: 2049kB
 -> Redistribute Motion 3:3 (slice2; segments: 3) (actual time=1.884..2.301 rows=67 loops=1)
 Output: b.name, b.id
 Hash Key: b.id
 -> Custom Scan (Anser Bloom Producer) (actual time=0.031..10.002 rows=67 loops=1)
 Output: b.name, b.id
 Bloom Filter Size: 1048576 bytes
 Bloom Filter Stats: memory=1024kB
 Anser Publish Time: 1.310 ms
 Extra Text: (seg1) Anser producer: parts=1 bytes=1048592 encode=1.021ms send=0.289ms
 -> Seq Scan on public.anser_rf_build b (actual time=0.014..0.098 rows=67 loops=1)
 Output: b.name, b.id
 Optimizer: Postgres query optimizer
 Execution Time: 39.884 ms

Read this carefully — it encodes several requirements:

• The producer has no wait, only a cost. It publishes fire-and-forget, so
  report Anser Publish Time (encode + send), not a wait count. Anything
  labelled "wait" on the producer is a leftover from the old libpq transport,
  where it blocked for an acknowledgement.
• Anser Coordinator is the line that makes the trace above legible:
  first/complete/delivered are offsets on the QD clock, fold is real
  work. In this example complete - first = 14.5ms for fold=0.31ms, which
  says the cost is pickup, not merging.
• Anser Result must distinguish delivered / cancelled / timeout.
  Without it, a fast failure and a successful delivery look identical.
• reads= on the consumer — messages taken off the socket while waiting.
  reads=0 on a timeout means nothing ever arrived; a non-zero count means
  something arrived that was not ours, which is a different bug.
• The Extra Text line is the per-segment breakdown shipped through
  cdbexplainbuf. The summary properties above it are the winning segment's
  values, like every other per-node MPP statistic. One line, key=value, no
  wrapping.
• Where the time lands is not symmetric. The producer publishes after its
  child is exhausted, so its cost shows in the node's last tuple time
  (0.031..10.002). The consumer must receive before returning anything, so its
  wait shows in the first tuple time (31.902..). Do not "fix" this; it is
  the truth about when each side blocks.
• The consumer emits 73 rows but the join emits 67: those 6 are bloom false
  positives. The filter may pass rows that do not join — it must never reject
  one that does.

EXPLAIN (ANALYZE) without VERBOSE — unchanged

The new lines are VERBOSE-only. Plain ANALYZE keeps exactly today's output:

 -> Custom Scan (Anser Bloom Consumer) (actual time=31.902..33.114 rows=73 loops=1)
 Bloom Filter Size: 1048576 bytes
 Bloom Filter Stats: memory=1024kB checked=667 rejected=594
 Rows Removed by Bloom Filter: 594

EXPLAIN (ANALYZE, VERBOSE) — the case this feature exists for

A producer never published (squelched, or a segment whose slice was abandoned),
so the channel never completed and the consumer waited anser.timeout_ms for
nothing before failing open. The query is still correct — just slower than with
the feature off, which is precisely what we cannot see today:

 -> Custom Scan (Anser Bloom Consumer) (actual time=1002.774..1003.918 rows=667 loops=1)
 Output: p.id, p.payload
 Bloom Filter Size: 1048576 bytes
 Bloom Filter Stats: memory=1024kB (no filter received)
 Anser Waits: 1
 Anser Wait Time: 1000.118 ms
 Anser Max Wait: 1000.118 ms
 Anser Result: timeout
 Anser Coordinator: parts=2 first=8.204ms complete=- delivered=- fold=0.21ms
 Extra Text: (seg2) Anser consumer: waits=1 total=1000.118ms max=1000.118ms reads=0 result=timeout

rows=667 — every probe row passed, no pruning happened, and a second was spent
waiting. Note parts=2 with no completion: the coordinator's line shows exactly
why it timed out, which is the diagnosis this whole issue is for.

Use case/motivation

No response

Related issues

#1942

Are you willing to submit a PR?

• Yes I am willing to submit a PR!