Anser - runtime instrumentation - MVP with one bloomfilter by leborchuk · Pull Request #1942 · apache/cloudberry

leborchuk · 2026-08-31T16:38:02Z

That's the MVP for the Anser https://vldb.org/pvldb/vol16/p3636-wu.pdf

Here we covered only scenario with adding bloomfilters to the query execution plan.

The main idea is as a follow

we have a distributed plan with motions and hash join, when one table is joined with another one, and only a small subset of rows meets the join criteria. So it'd be useful to filter out rows from big one table in advance.
we could use bloom filters to filter out rows. To do so we should gather it on segments, combine in one on master, redistribute to the segments and filter out rows before hash join (after seq scan)

The overall execution plan should looks like (see Custom Scan nodes and their stat)

postgres=# explain analyze select
  aef.name
  ,aef.value
  ,aef.id
  ,a.*
from
  applications_extra_fields as aef
  left join applications as a
    on aef.id = a.id;
                                                                           QUERY PLAN

------------------------------------------------------------------------------------------------------------------------------------------------------------
----
 Gather Motion 3:1  (slice1; segments: 3)  (cost=0.00..1689.57 rows=199266 width=386) (actual time=79.799..416.211 rows=100000 loops=1)
   ->  Hash Right Join  (cost=0.00..1402.93 rows=66422 width=386) (actual time=78.964..299.241 rows=33850 loops=1)
         Hash Cond: (a.id = aef.id)
         Extra Text: (seg1)   Hash chain length 10.1 avg, 20 max, using 3365 of 262144 buckets.
         ->  Custom Scan (Anser Bloom Consumer)  (cost=0.00..499.02 rows=333334 width=353) (actual time=21.440..224.105 rows=3385 loops=1)
               Bloom Filter Size: 1048576 bytes
               Bloom Filter Stats: memory=1024kB checked=332791 rejected=329406
               Rows Removed by Bloom Filter: 329406
               ->  Seq Scan on applications a  (cost=0.00..499.02 rows=333334 width=353) (actual time=2.663..193.493 rows=334042 loops=1)
         ->  Hash  (cost=437.42..437.42 rows=33334 width=33) (actual time=56.086..56.088 rows=33850 loops=1)
               Buckets: 262144  Batches: 1  Memory Usage: 4233kB
               ->  Redistribute Motion 3:3  (slice2; segments: 3)  (cost=0.00..437.42 rows=33334 width=33) (actual time=0.176..50.171 rows=33850 loops=1)
                     Hash Key: aef.id
                     ->  Custom Scan (Anser Bloom Producer)  (cost=0.00..431.94 rows=33334 width=33) (actual time=0.361..41.874 rows=50000 loops=1)
                           Bloom Filter Size: 1048576 bytes
                           Bloom Filter Stats: memory=1024kB
                           ->  Seq Scan on applications_extra_fields aef  (cost=0.00..431.94 rows=33334 width=33) (actual time=0.360..6.863 rows=50000 loops
=1)
 Planning Time: 15.377 ms
   (slice0)    Executor memory: 1143K bytes.
   (slice1)    Executor memory: 8605K bytes avg x 3x(0) workers, 8637K bytes max (seg0).  Work_mem: 4233K bytes max.
   (slice2)    Executor memory: 2370K bytes avg x 3x(0) workers, 2370K bytes max (seg2).
 Memory used:  128000kB
 Optimizer: GPORCA
 Execution Time: 427.266 ms

The main architecture overview

See detailed description in [src/backend/cdb/anser/README.md]

All Anser state lives in fixed coordinator shared memory, allocated once at
postmaster start. Producers and consumers are ordinary query backends
(coordinator-resident, or on segments reaching the coordinator over libpq); they
never talk to each other directly and never own the shared state — they only
hand work to, or wait on, two background workers that do.

Three shared structures, two hand-off points:

Channel map — the hash of channels (one per runtime condition per query),
holding each channel's state, accounting, and payload. The single source of
truth.
Submission queue — the producer → gather hand-off. A producer copies its
serialized part into a free queue entry, signals the gather worker, and blocks
for an ACK; it never touches the channel payload itself.
Wait table — the send → consumer hand-off, an array of slots. A slot
is one consumer's reservation on a channel: it records the consumer's key, a
pointer to that backend's latch, and a place for the send worker to stamp the
delivered payload (or a cancel). A blocked consumer owns one slot and sleeps on
its latch until the send worker flips it.

The two background workers exist because the shared state has to keep moving
independent of any one transient/blocked backend:

Gather service — drains the submission queue: for each part it folds
(bitwise-OR unions) the data into the target channel's single payload, advances
the channel toward READY, and ACKs the producer. It also runs periodic
maintenance: time out stragglers (gp_anser_timeout_ms) and sweep terminal or
orphaned channels.
Send service — delivers: once a channel is READY it copies the combined
payload into every waiting slot and wakes those consumers' latches; when all
expected consumers are served it recycles the channel to CONSUMED.

Both workers sleep on a latch and wake on demand — a producer's submission sets
the gather latch, a publish/registration sets the send latch — plus a periodic
timeout so maintenance runs even when idle. Concurrency is guarded by two
LWLocks, always taken in the order AnserChannelLock → AnserRingLock.

Why MVP

For some queries using bloom filters leads to performance degradation. Usually it happens when there are no significant row dataset reduction after bloom filtering. We will address these issue in future researches/PRs.

Also I haven't checked all the cases where bloomfilters could be used.

So for now we support only one simple type of hashjoin, see details in anser_hashjoin_keys and anser_resolve_build_scan functions.

Open questions

We use Custom Nodes, do not create our own Anser nodes. The main reason here is to make PG rebase process easier. Is it OK?
We use libpq protocol to send/get data to/from master. Since we added new functions to work with Anser those functions were registered in a catalog with oid 8195-8197, and catversion was increased. It it Ok or we should create our own protocol?
Anser is runtime instrumentation to improve execution time without changes in current planning flow. So PlannedStmt * planner was modified and special hook AnserApplyRuntimeFilters was added. We generate execution plan and after that add to it special steps for gathering/redistribution runtime data. Is it Ok or we should fix the planner too?
Anser consumer does not have timeout. So when step executes it just opens connection an wait for data. Do not use timeout, we cannot use it here, just open connection and wait data. If something goes wrong consumer will be wait forever. It'd be better to limit waits somehow, but I cannot understand how to do it.

How to enable && test it

SET gp_anser_enable=on
SET gp_anser_runtime_filter=on;

make -C src/test/modules/anser
make -C src/test/modules/anser install
make -C src/test/modules/anser installcheck

Anser is a runtime pub/sub facility for MPP execution: producers publish a small piece of query information (today: a bloom filter over a join build key), the coordinator unions the per-segment parts into one global payload (bitwise OR), and consumers receive it to prune work. This commit adds the subsystem core: the coordinator-resident shared-memory channel map with a five-state channel lifecycle, the producer submission queue and consumer wait table, the gather/send background services that own the data path, the bloom part wire format and fold helpers, and the bloom filter library accessors they need. Includes postmaster/shmem/ LWLock/GUC wiring and the subsystem README. Everything is fail-open: any failure degrades to unfiltered execution, never wrong results. Disabled by default via gp_anser_enable.

Segment producers/consumers cannot touch the coordinator-resident channel map directly, so they open a libpq connection back to the QD (address from gp_qd_hostname/gp_qd_port) and drive the new gp_anser_producer_begin / gp_anser_publish / gp_anser_consume_wait builtins. Since stock clusters never grant segment hosts in the coordinator's pg_hba.conf, these connections authenticate with a per-session random token instead, following the parallel-retrieve cursor model: the QD registers a 128-bit token in a shared-memory session token hash, a gp_anser_conn startup marker selects token authentication in ClientAuthentication before pg_hba is consulted, and the token is presented as the connection password. Channels stay bound to their creator role. All paths fail open to unfiltered execution.

Add the planner integration that puts Anser to work: a post-plan hook in planner() (covers both the Postgres planner and ORCA) recognizes a single-column equijoin over a hash build and injects two pass-through CustomScan providers. The producer above the build scan feeds join keys into a bloom filter and publishes it to the channel; the consumer above the probe scan receives the unioned filter and prunes rows that cannot join. Executor helper library selects the transport by role (coordinator: direct shmem; segment: libpq to the QD with the session token carried in the plan). EXPLAIN reports planned filter size and, under ANALYZE, winner-segment lookup/prune counters via instrumentation. Adds the src/test/modules/anser regression module (channel map, services, bloom protocol, client loopback, session token, plan injection) and wires it into CI as the ic-anser test with gp_anser_enable=on.

Copilot

Pull request overview

This PR introduces the Anser runtime instrumentation subsystem for Cloudberry/Greenplum-style MPP execution, delivering an MVP runtime Bloom filter that is produced on segments, unioned on the coordinator, and consumed on segments to prune probe-side rows before hash joins—without changing the existing planning flow (post-plan tree injection).

Changes:

Adds coordinator-resident Anser shared-memory channel map plus gather/send background workers, and a libpq-based segment→QD transport with token authentication.
Implements plan-tree injection and executor support via CustomScan “Anser Bloom Producer/Consumer” nodes plus Bloom payload serialization/union helpers.
Adds a comprehensive regression test module (src/test/modules/anser) and wires it into Meson/Make and CI.

Reviewed changes

Copilot reviewed 44 out of 44 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
src/test/modules/meson.build	Adds `anser` test module to Meson build.
src/test/modules/Makefile	Adds `anser` test module to make-based test build.
src/test/modules/anser/test_anser.control	Defines `test_anser` extension for SQL-callable test helpers.
src/test/modules/anser/test_anser--1.0.sql	Registers SQL functions implemented by `anser_test.c`.
src/test/modules/anser/sql/test_anser.sql	Regression tests for channel lifecycle, payload correctness, libpq transport, and maintenance behavior.
src/test/modules/anser/sql/anser_runtime_filter.sql	Plan-shape and correctness regression for runtime filter injection and results stability.
src/test/modules/anser/meson.build	Builds/installs the `test_anser` shared module and regression schedule (Meson).
src/test/modules/anser/Makefile	Builds/installs the `test_anser` shared module and regression schedule (Make).
src/test/modules/anser/expected/test_anser.out	Expected output for `test_anser` regression.
src/test/modules/anser/expected/anser_runtime_filter.out	Expected output for runtime-filter plan/correctness regression.
src/test/modules/anser/anser_test.c	SQL-callable C helpers driving Anser APIs and libpq loopback tests.
src/include/utils/unsync_guc_name.h	Marks Anser GUCs as unsynchronized.
src/include/postmaster/postmaster.h	Increases auxiliary background worker count to accommodate Anser workers.
src/include/lib/bloomfilter.h	Adds bitset accessors and constructor-from-bitset API for deserialization.
src/include/executor/nodeAnserBloomFilter.h	Declares executor helper APIs for Bloom producer/consumer.
src/include/cdb/anserplan.h	Declares post-plan runtime-filter injection and CustomScan builders.
src/include/cdb/anserfilter.h	Declares Bloom part framing, serialization, deserialization, and fold API.
src/include/cdb/anserclient.h	Declares libpq client helpers for segment↔QD Anser transport.
src/include/cdb/anser.h	Introduces Anser shared-memory channel map API, GUCs, and service hooks.
src/include/catalog/pg_proc.dat	Adds built-in `gp_anser_*` functions for producer/publish/consume_wait transport.
src/include/catalog/catversion.h	Bumps catalog version for new built-ins.
src/backend/utils/misc/guc_gp.c	Adds Anser GUC definitions (enable/runtime_filter/limits/timeout/marker).
src/backend/utils/init/postinit.c	Registers CustomScan providers on backend init so dispatched plans resolve methods.
src/backend/storage/lmgr/lwlocknames.txt	Adds Anser LWLock names.
src/backend/storage/ipc/ipci.c	Accounts for and initializes Anser shared memory at postmaster start.
src/backend/postmaster/postmaster.c	Adds Anser gather/send background workers.
src/backend/postmaster/bgworker.c	Registers Anser worker entrypoints.
src/backend/optimizer/plan/planner.c	Calls `AnserApplyRuntimeFilters()` as a post-plan hook (ORCA + PG planner).
src/backend/libpq/auth.c	Adds `gp_anser_conn` startup marker parsing and token-based auth branch.
src/backend/lib/bloomfilter.c	Implements bitset accessors and `bloom_create_from_bitset`.
src/backend/executor/nodeAnserBloomFilterProduce.c	Implements executor helper for building/publishing per-producer Bloom parts.
src/backend/executor/nodeAnserBloomFilterConsume.c	Implements executor helper for consuming and reconstructing the merged Bloom filter.
src/backend/executor/Makefile	Links new executor helper objects.
src/backend/cdb/Makefile	Adds `anser` backend subdir to cdb build.
src/backend/cdb/anser/README.md	Documents architecture, transport/auth, GUCs, and channel state machine.
src/backend/cdb/anser/Makefile	Builds Anser subsystem objects.
src/backend/cdb/anser/anserservice.c	Implements coordinator-local gather/send worker main loops and error recovery.
src/backend/cdb/anser/anserplanexec.c	CustomScan execution providers for producer/consumer plan nodes + EXPLAIN stats.
src/backend/cdb/anser/anserplan.c	Plan-tree recognition and injection logic (hash join shape) + sizing and token registration.
src/backend/cdb/anser/anserfuncs.c	Implements built-in SQL functions backing the transport (producer/publish/consume_wait).
src/backend/cdb/anser/anserfilter.c	Implements Bloom payload framing, serialization/deserialization, and fold-in-place union.
src/backend/cdb/anser/anserclient.c	Implements libpq client transport used by segments to reach coordinator services.
.github/workflows/build-cloudberry.yml	Adds CI job to run `src/test/modules/anser` installcheck with `gp_anser_enable=on`.

Suppressed comments (1)

src/test/modules/anser/anser_test.c:1076

anser_test_dsm_free_on_cancel() disables the global Anser sweep (AnserSetSweepEnabled(false)) but never re-enables it in PG_FINALLY. Because sweep_enabled is global shared memory state, this can leak into later tests/sessions.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

The query-tuning GUC list test now also returns gp_anser_runtime_filter (registered under QUERY_TUNING_METHOD by the plan-injection commit); add it to the expected output. This commit collects CI/review fixes for the Anser PR series; squash later fixes into it.

yjhjstz · 2026-09-01T13:40:59Z

what's difference with 6c41d27 impl?

The query-tuning GUC list test now also returns gp_anser_runtime_filter (registered under QUERY_TUNING_METHOD by the plan-injection commit); add it to the expected output. This commit collects CI/review fixes for the Anser PR series; squash later fixes into it.

leborchuk · 2026-09-01T17:58:14Z

what's difference with 6c41d27 impl?

Runtime filter pushdown is intra-slice only. The RuntimeFilter executor node reaches the HashJoin's in-memory hash table via a plain executor pointer (node->hjstate->hj_HashTable, nodeRuntimeFilter.c:83-86); the Hash variant hands bloom scankeys to a registered SeqScan/DynamicSeqScanState in the same process. Nothing ever crosses a Motion or the network. Use the same hash function.

Anser is cross-slice, cross-segment. Per-segment blooms are unioned on the coordinator into a global filter; any consumer anywhere can use it. That's the general MPP case from the paper — non-colocated joins, producer and consumer in different slices, even different joins sharing an equivalence-class condition_key. Could use different hash function. The built-in structurally cannot do any of that.

But let's return to the example. For the proposed example you are right, existing approach is better.

Explain with gp_enable_runtime_filter_pushdown TO on:

 Gather Motion 3:1  (slice1; segments: 3)  (cost=0.00..1689.57 rows=199266 width=386) (actual time=24.519..254.783 rows=100000 loops=1)
   ->  Hash Right Join  (cost=0.00..1402.93 rows=66422 width=386) (actual time=24.110..198.894 rows=33850 loops=1)
         Hash Cond: (a.id = aef.id)
         Extra Text: (seg1)   Hash chain length 10.1 avg, 20 max, using 3365 of 262144 buckets.
         ->  Seq Scan on applications a  (cost=0.00..499.02 rows=333334 width=353) (actual time=1.823..154.653 rows=3386 loops=1)
               Rows Removed by Pushdown Runtime Filter: 329405
         ->  Hash  (cost=437.42..437.42 rows=33334 width=33) (actual time=21.867..21.868 rows=33850 loops=1)
               Buckets: 262144  Batches: 1  Memory Usage: 4233kB
               ->  Redistribute Motion 3:3  (slice2; segments: 3)  (cost=0.00..437.42 rows=33334 width=33) (actual time=1.316..17.193 rows=33850 loops=1)
                     Hash Key: aef.id
                     ->  Seq Scan on applications_extra_fields aef  (cost=0.00..431.94 rows=33334 width=33) (actual time=0.299..6.707 rows=50000 loops=1)
 Planning Time: 6.577 ms
 Optimizer: GPORCA
 Execution Time: 258.942 ms

Explain with gp_anser_runtime_filter=on:

 Gather Motion 3:1  (slice1; segments: 3)  (cost=0.00..1689.57 rows=199266 width=386) (actual time=88.531..358.959 rows=100000 loops=1)
   ->  Hash Right Join  (cost=0.00..1402.93 rows=66422 width=386) (actual time=88.154..310.096 rows=33850 loops=1)
         Hash Cond: (a.id = aef.id)
         Extra Text: (seg1)   Hash chain length 10.1 avg, 20 max, using 3365 of 262144 buckets.
         ->  Custom Scan (Anser Bloom Consumer)  (cost=0.00..499.02 rows=333334 width=353) (actual time=21.028..224.871 rows=3385 loops=1)
               Bloom Filter Size: 1048576 bytes
               Bloom Filter Stats: memory=1024kB checked=332791 rejected=329406
               Rows Removed by Bloom Filter: 329406
               ->  Seq Scan on applications a  (cost=0.00..499.02 rows=333334 width=353) (actual time=1.252..158.640 rows=334042 loops=1)
         ->  Hash  (cost=437.42..437.42 rows=33334 width=33) (actual time=66.655..66.657 rows=33850 loops=1)
               Buckets: 262144  Batches: 1  Memory Usage: 4233kB
               ->  Redistribute Motion 3:3  (slice2; segments: 3)  (cost=0.00..437.42 rows=33334 width=33) (actual time=1.271..63.106 rows=33850 loops=1)
                     Hash Key: aef.id
                     ->  Custom Scan (Anser Bloom Producer)  (cost=0.00..431.94 rows=33334 width=33) (actual time=0.357..49.022 rows=50000 loops=1)
                           Bloom Filter Size: 1048576 bytes
                           Bloom Filter Stats: memory=1024kB
                           ->  Seq Scan on applications_extra_fields aef  (cost=0.00..431.94 rows=33334 width=33) (actual time=0.355..8.658 rows=50000 loops
=1)
 Planning Time: 5.853 ms
 Optimizer: GPORCA
 Execution Time: 369.783 ms

The interesting detail is that Rows Removed by is practically equal for both optimizations. It means that we could cross-check new approach with existing one. I did it and fixed a couple of bugs )

Another idea is why use separate step for filter out rows? We could push down all filters close to AM-level. And use it, for example in PAX or in the future iceberg approach. I want to implement it in the future but since we are talking about it here, add push-down to seq scan.

The true meaning this PR is to add Anser, bloomfilters here just the tool for check how whole system works. I'm going to address all issues in other PR's, where I could just use working system. Here we have for about 8500 lines of the new code ...

The query-tuning GUC list test now also returns gp_anser_runtime_filter (registered under QUERY_TUNING_METHOD by the plan-injection commit); add it to the expected output. This commit collects CI/review fixes for the Anser PR series; squash later fixes into it.

leborchuk · 2026-09-01T18:20:24Z

Pushed down filters to seq scan. Now execution plan looks like:

 Gather Motion 3:1  (slice1; segments: 3)  (cost=0.00..1689.57 rows=199266 width=386) (actual time=79.213..317.383 rows=100000 loops=1)
   ->  Hash Right Join  (cost=0.00..1402.93 rows=66422 width=386) (actual time=79.679..251.165 rows=33850 loops=1)
         Hash Cond: (a.id = aef.id)
         Extra Text: (seg1)   Hash chain length 10.1 avg, 20 max, using 3365 of 262144 buckets.
         ->  Custom Scan (Anser Bloom Consumer)  (cost=0.00..499.02 rows=333334 width=353) (actual time=18.552..173.863 rows=3385 loops=1)
               Bloom Filter Size: 1048576 bytes
               Bloom Filter Stats: memory=1024kB checked=0 rejected=0
               Rows Removed by Bloom Filter: 0
               ->  Seq Scan on applications a  (cost=0.00..499.02 rows=333334 width=353) (actual time=1.079..154.374 rows=3385 loops=1)
                     Rows Removed by Pushdown Runtime Filter: 329406
         ->  Hash  (cost=437.42..437.42 rows=33334 width=33) (actual time=60.732..60.734 rows=33850 loops=1)
               Buckets: 262144  Batches: 1  Memory Usage: 4233kB
               ->  Redistribute Motion 3:3  (slice2; segments: 3)  (cost=0.00..437.42 rows=33334 width=33) (actual time=1.260..57.224 rows=33850 loops=1)
                     Hash Key: aef.id
                     ->  Custom Scan (Anser Bloom Producer)  (cost=0.00..431.94 rows=33334 width=33) (actual time=0.440..44.779 rows=50000 loops=1)
                           Bloom Filter Size: 1048576 bytes
                           Bloom Filter Stats: memory=1024kB
                           ->  Seq Scan on applications_extra_fields aef  (cost=0.00..431.94 rows=33334 width=33) (actual time=0.437..6.862 rows=50000 loops
=1)
 Planning Time: 7.103 ms
   (slice0)    Executor memory: 1143K bytes.
   (slice1)    Executor memory: 8604K bytes avg x 3x(0) workers, 8636K bytes max (seg0).  Work_mem: 4233K bytes max.
   (slice2)    Executor memory: 2370K bytes avg x 3x(0) workers, 2370K bytes max (seg2).
 Memory used:  128000kB
 Optimizer: GPORCA
 Execution Time: 327.988 ms

I know right now it's better not use Anser, but it's the subject for future improvements.

leborchuk added 3 commits August 31, 2026 19:19

Copilot AI lite review requested due to automatic review settings August 31, 2026 16:38

leborchuk marked this pull request as draft August 31, 2026 16:38

Merge branch 'main' into anser-prs

217e846

Copilot started reviewing on behalf of leborchuk August 31, 2026 16:38 View session

Copilot AI reviewed Aug 31, 2026

View reviewed changes

Comment thread src/backend/executor/nodeAnserBloomFilterProduce.c

Comment thread src/test/modules/anser/anser_test.c

Comment thread src/test/modules/anser/anser_test.c

Comment thread src/test/modules/anser/anser_test.c

Comment thread src/backend/executor/nodeAnserBloomFilterConsume.c

leborchuk and others added 3 commits August 31, 2026 23:14

Merge branch 'main' into anser-prs

6844627

leborchuk added 2 commits September 1, 2026 19:58

Navigation Menu

Uh oh!

Anser - runtime instrumentation - MVP with one bloomfilter - #1942

Anser - runtime instrumentation - MVP with one bloomfilter#1942
leborchuk wants to merge 10 commits into
apache:mainfrom
leborchuk:anser-prs

leborchuk commented Aug 31, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

yjhjstz commented Sep 1, 2026

Uh oh!

leborchuk commented Sep 1, 2026 •

edited

Loading

Uh oh!

leborchuk commented Sep 1, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

leborchuk commented Aug 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

The main idea is as a follow

The main architecture overview

Why MVP

Open questions

How to enable && test it

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

yjhjstz commented Sep 1, 2026

Uh oh!

leborchuk commented Sep 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

leborchuk commented Sep 1, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

leborchuk commented Aug 31, 2026 •

edited

Loading

leborchuk commented Sep 1, 2026 •

edited

Loading