GitHub user leborchuk edited a discussion: [Ideas] Add instrumentation and
latency metrics to the Anser subsystem
# 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:
| t (ms) | Event |
| --- | --- |
| 0.0 | 3 producers build their filters |
| 8.7 – 10.0 | all 3 published (`sent=1`) |
| 10.2 | 3 consumers subscribe and block |
| 16.8 / 24.0 / 31.4 | coordinator folds part 1 / 2 / 3 |
| 31.4 | channel complete, delivering |
| 33.2 – 34.0 | all 3 consumers hold the filter |
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.
| Vantage point | What it can measure | Where |
| --- | --- | --- |
| **Producer node** (segment) | build, serialize, base64 encode, send. There is
**no wait**: publishing is fire-and-forget since the dispatch-transport
migration | `AnserProducePublishPart()` (`src/anserbloomproduce.c:60`),
`AnserSidebandPublish()` (`src/ansersideband.c:94`) |
| **Consumer node** (segment) | subscribe → payload in hand; how many messages
it read while waiting; the outcome | `ExecAnserBloomFilterConsumeSideband()`
(`src/anserbloomconsume.c:100`), `AnserSidebandConsumeWait()`
(`src/ansersideband.c:137`) |
| **Coordinator** (QD backend) | first part arrival → complete → delivered, and
fold time per part | `anser_disp_apply_part()` (`src/anserdispatch.c:302`),
`anser_disp_deliver()` (`:372`), `anser_disp_push()` (`:394`) |
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
https://github.com/apache/cloudberry/pull/1942
### Are you willing to submit a PR?
- [ ] Yes I am willing to submit a PR!
GitHub link: https://github.com/apache/cloudberry/discussions/1958
----
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]
