GitHub user leborchuk created a discussion: [Ideas] Add instrumentation and
latency metrics to the Anser subsystem
### Description
# 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.
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 producer blocks until the
coordinator acknowledges its part; a consumer blocks until the payload is
delivered. If those waits are expensive, the filter can be a net loss and we
currently have no way to tell.
## Goal
Make every wait and every queueing delay in Anser measurable, both
per query (in `EXPLAIN`) and cluster-wide (in shared memory).
## Definition of done
- [ ] `EXPLAIN (ANALYZE, VERBOSE)` shows producer and consumer wait count and
wait time, **including for segment-executed nodes**.
- [ ] `SELECT * FROM anser.stats()` shows queue time, service-cycle latency,
and the histograms; `anser.stats_reset()` works.
- [ ] `SELECT * FROM anser.session_stats()` reflects the current session only.
- [ ] 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 LWLock acquisitions on the data path; no per-tuple timing calls.
- [ ] Core (`src/`) is untouched.
## What to measure
There are two distinct vantage points. Do not mix them up — this is the single
most important thing to understand before writing code.
| Vantage point | Who waits | Where the wait actually happens |
| --- | --- | --- |
| **Executor node** (segment or coordinator) | the query | segment: the libpq
round trip in `AnserClientPublish` / `AnserClientConsumeWait`
(`src/anserclient.c`); coordinator: the direct calls `AnserPublish` /
`AnserWaitProducersRegistered` + `AnserWaitReady` + `AnserConsumeReady`
(`src/anser.c`) |
| **Coordinator internals** | the QD backend serving a segment, and the two
background workers | `AnserWaitSubmissionAck` (`src/anser.c:906`),
`AnserWaitSlotResult` (`src/anser.c:1110`), `AnserGatherServiceCycle`
(`src/anser.c:1235`), `AnserSendServiceCycle` (`src/anser.c:1433`) |
The first belongs in `EXPLAIN`. The second belongs in shared-memory counters,
because no single backend sees it.
## Implementation
Suggest three changes in this order. Each is independently reviewable and
testable.
### Change 1 — per-node wait time in EXPLAIN
- [ ] Time the publish call in the producer node and the receive call in the
consumer node. Record, per node: **number of waits**, **total wait time**,
and **longest single wait**.
- Producer: `AnserProducePublishPart()` in `src/anserbloomproduce.c` — it
picks
the transport by role, so wrapping it covers both segment and coordinator.
- Consumer: `ExecAnserBloomFilterConsume()` in `src/anserbloomconsume.c`.
- Store the accumulators in `AnserBloomProduceScanState` /
`AnserBloomConsumeScanState` (`src/anserplanexec.c:81` and `:95`).
- [ ] Print them from `anser_produce_explain()` (`src/anserplanexec.c:422`) and
`anser_consume_explain()` (`:636`), gated on `es->analyze` (there is
nothing to print without execution) and shown only at
`EXPLAIN (ANALYZE, VERBOSE)` — i.e. `es->analyze && es->verbose`.
Use `ExplainPropertyFloat("Anser Wait Time", "ms", ...)` and
`ExplainPropertyInteger("Anser Waits", NULL, ...)`.
**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:583-588`:
the existing code deliberately uses `InstrCountFiltered1/2` 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`** —
it is the closest existing analogue (a runtime-filter node doing exactly this)
So: allocate `cdbexplainbuf` in `*_begin` when `estate->es_instrument` is set,
install a `cdbexplainfun` that appends your numbers, and the QD will show 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 — subsystem-wide counters and latency histograms in shared memory
- [ ] Add a stats block to Anser's shared memory. Either extend `AnserControl`
(`include/anser.h`) or add a separate struct; account for it in
`AnserShmemSize()` and initialize it in `AnserShmemInit()`
(`src/anser.c`), exactly as the existing structures do.
- [ ] Use `pg_atomic_uint64` (`src/include/storage/atomics.h`) for every counter
and bucket, and update them **without** taking `AnserChannelLock` /
`AnserRingLock`. These are on the data path; adding lock traffic for
metrics would change what we are trying to measure.
- [ ] Measure, at minimum:
| Metric | Where |
| --- | --- |
| time a part waits in the submission queue (enqueue → gather picks it up) |
stamp the time in `AnserEnqueueSubmission()` (`src/anser.c:860`), read it in
`AnserGatherServiceCycle()` (`:1235`) — add a field to `AnserSubmissionEntry` |
| time a consumer slot waits for delivery (register → send stamps payload) |
stamp in `AnserRegisterWaitSlot()` (`:1077`), read in `AnserSendServiceCycle()`
(`:1433`) — add a field to `AnserWaitSlot` |
| producer ack wait, on the QD side | `AnserWaitSubmissionAck()` (`:906`) |
| consumer delivery wait, on the QD side | `AnserWaitSlotResult()` (`:1110`) |
| background worker cycle duration, per service | around the
`AnserGatherServiceCycle()` / `AnserSendServiceCycle()` calls in
`AnserServiceLoop()` (`src/anserservice.c`) |
| counts: parts submitted / accepted / rejected, channels cancelled by
timeout, consumers delivered / cancelled | the same call sites |
- [ ] Histogram design: fixed power-of-two microsecond buckets, e.g. 20 buckets
covering `<1µs, <2µs, ... , <512ms, >=512ms`. Compute the bucket with
`pg_leftmost_one_pos64()` (`src/include/port/pg_bitutils.h`) — no loops,
no
floating point. Keep the bucket count a compile-time constant; the array
lives in fixed shared memory sized at postmaster start.
- [ ] Use `instr_time` for all timing (`src/include/portability/instr_time.h`:
`INSTR_TIME_SET_CURRENT`, `INSTR_TIME_SUBTRACT`,
`INSTR_TIME_GET_MICROSEC`).
Do **not** use `GetCurrentTimestamp()` for durations.
**Accuracy caveat to document in the code:** the wait loops sleep in slices
(`ANSER_WAIT_POLL_INTERVAL_MS` = 10 ms, `ANSER_WAIT_LATCH_TIMEOUT_MS` = 1000 ms,
`ANSER_SERVICE_WAKEUP_INTERVAL_MS` = 1000 ms — see `include/anser.h`). Measure
wall time across the *whole* wait call, never by counting loop iterations, and
note in a comment that sub-poll-interval waits are quantized.
### Change 3 — per-session stats and the user-facing view
- [ ] Accumulate this backend's Anser activity in a plain `static` struct in
backend-local memory (no atomics, no locks — it is private).
- [ ] Flush it into the shared-memory totals once per query from an
`ExecutorEnd_hook` (`src/include/executor/executor.h:108`). Install it in
`_PG_init()` (`src/anserinit.c`) and **chain the previous hook** the same
way `anser_planner()` chains `prev_planner_hook` — copy that pattern.
- [ ] Expose the numbers:
- `anser.stats()` — set-returning function over the shared totals and the
histogram (one row per bucket, or one row per metric with the histogram as
an array; your call, document it), plus a matching view.
- `anser.session_stats()` — the backend-local numbers.
- `anser.stats_reset()` — zero the shared block; superuser only.
- [ ] Ship this as an **upgrade script**: add `anser--1.0--1.1.sql`, bump
`default_version` in `anser.control`, and add both to `DATA` in the
`Makefile`. Do not edit `anser--1.0.sql` in place — installed clusters
already have 1.0.
## Testing
- [ ] Extend `gpcontrib/anser/sql/anser_test.sql` (+ `expected/anser_test.out`).
- [ ] **Never put a measured time or a histogram bucket count in expected
output** — the tests would fail randomly. Assert properties instead:
`anser_test_...() > 0`, `total_waits` increased after a known round trip,
`stats_reset()` zeroes things. 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 the output through a query that checks only that 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'`, `anser.enable=on`, and
`CREATE EXTENSION anser` in the test database.
### 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:
```
EXPLAIN (ANALYZE, VERBOSE, COSTS OFF)
SELECT b.name, p.payload
FROM anser_rf_build b JOIN anser_rf_probe p ON b.id = p.id;
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: 31.774 ms
Anser Max Wait: 31.774 ms
Extra Text: (seg0) Anser consumer: waits=1 total=31.774ms
max=31.774ms registration=0.412ms delivery=31.362ms transport=libpq
-> 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..12.902 rows=67 loops=1)
Output: b.name, b.id
Bloom Filter Size: 1048576 bytes
Bloom Filter Stats: memory=1024kB
Anser Waits: 1
Anser Wait Time: 12.418 ms
Anser Max Wait: 12.418 ms
Extra Text: (seg1) Anser producer: waits=1
total=12.418ms max=12.418ms parts=1 bytes=1048640 transport=libpq
-> Seq Scan on public.anser_rf_build b (actual
time=0.014..0.098 rows=67 loops=1)
Output: b.name, b.id
Planning Time: 4.221 ms
(slice0) Executor memory: 84K bytes.
(slice1) Executor memory: 3204K bytes avg x 3 workers, 3210K bytes max
(seg0). Work_mem: 2049K bytes max.
(slice2) Executor memory: 412K bytes avg x 3 workers, 412K bytes max
(seg1).
Memory used: 128000kB
Optimizer: Postgres query optimizer
Execution Time: 39.884 ms
```
Read this example carefully — it encodes several requirements:
- **Three summary properties per node**, in this order: `Anser Waits`
(integer), `Anser Wait Time` (ms), `Anser Max Wait` (ms). Emit them via
`ExplainPropertyInteger` / `ExplainPropertyFloat` so JSON/YAML/XML output
formats work for free — never `appendStringInfo` into the plan text.
- **The `Extra Text` line is the per-segment breakdown** shipped through
`cdbexplainbuf` (see the main issue). The summary properties above it are the
winning segment's values, exactly like every other per-node MPP statistic;
`Extra Text` is where a specific segment's detail shows up. Keep it one line,
`key=value` pairs, no line wrapping.
- **The consumer's wait splits into two phases**: `registration` (waiting for
producers to announce themselves) and `delivery` (waiting for the payload
after that). They come from different code paths and behave differently
under load, so report them separately.
- **`transport=libpq` vs `transport=shmem`** tells you whether the node ran on
a segment (libpq round trip to the QD) or coordinator-local (direct shared
memory). Without it you cannot interpret the numbers at all.
- **Where the time lands is not symmetric.** The producer publishes after its
child is exhausted, so its wait shows up in the node's *last* tuple time
(`0.031..12.902`). The consumer must receive before it can return anything,
so its wait shows up 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 is allowed to 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 (crash, or a segment that produced no rows and was
cancelled), the channel hit the produce deadline, and the consumer waited
`anser.timeout_ms` for nothing before failing open. The query is still correct —
just slower than it would have been with the feature off, which is precisely
what we currently cannot see:
```
-> 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: cancelled)
Anser Waits: 1
Anser Wait Time: 1002.601 ms
Anser Max Wait: 1002.601 ms
Extra Text: (seg2) Anser consumer: waits=1 total=1002.601ms
max=1002.601ms registration=1000.118ms delivery=0.000ms result=cancelled
transport=libpq
```
Note `rows=667` — every probe row passed, no pruning happened, and 1 second was
spent waiting. A `result=` field distinguishing `delivered` / `cancelled` /
`timeout` is required for exactly this reason.
#### `anser.stats()` — the shared-memory side (PR 2/3)
One row per metric, histogram buckets as an array so a single query is readable:
```
SELECT metric, count, total_ms, max_ms, p50_ms, p99_ms FROM anser.stats();
metric | count | total_ms | max_ms | p50_ms | p99_ms
-----------------------------+-------+----------+--------+--------+--------
producer_ack_wait | 1284 | 15012.44 | 84.12 | 8.19 | 65.53
submission_queue_time | 1284 | 612.08 | 16.38 | 0.26 | 4.09
consumer_registration_wait | 428 | 4118.72 | 131.07 | 4.09 | 131.07
consumer_delivery_wait | 428 | 12844.16 | 262.14 | 16.38 | 262.14
send_slot_queue_time | 428 | 131.07 | 2.05 | 0.13 | 1.02
gather_cycle | 18422 | 921.60 | 8.19 | 0.03 | 0.52
send_cycle | 18422 | 755.20 | 4.09 | 0.02 | 0.26
(7 rows)
SELECT metric, buckets FROM anser.stats() WHERE metric =
'consumer_delivery_wait';
metric | buckets
------------------------+--------------------------------------------------
consumer_delivery_wait | {0,0,0,0,0,0,2,11,48,192,140,28,6,1,0,0,0,0,0,0}
(1 row)
SELECT * FROM anser.stats_counters();
counter | value
----------------------------+-------
parts_submitted | 1284
parts_accepted | 1281
parts_rejected | 3
channels_timed_out | 2
consumers_delivered | 426
consumers_cancelled | 2
(6 rows)
```
Because `p50_ms` / `p99_ms` are derived from the fixed power-of-two buckets they
are approximate — bucket boundaries, not interpolated values. Say so in the
`README.md` metrics table rather than pretending to millisecond precision.
`total_ms` and `max_ms` are exact.
### 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]
