Anser - runtime instrumentation - MVP with one bloomfilter

[leborchuk]

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]

Implementation consists of parts:

1. Add new authentication hook to core to bypass pg_hba checking. Need to connect from segments to master and send data to Anser/consume data from Anser.
2. Expand bloomfilter interfaces and create bloomfilters from paylod, just copying data
3. gpcontrib/anser extension to add Anser core - shared memory structures and background workers.

Right now working with bloomfilters is part of Anser, but maybe afterwards I will take it out to the new extension (depend of Anser). It's

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

1. 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?
2. 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?
3. 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
shared_preload_libraries = 'anser'
anser.enable=on

restart cluster

CREATE EXTENSION anser;
SET anser.runtime_filter=on;

test

cd gpcontrib/anser
make installcheck

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

[yjhjstz]

what's difference with 6c41d27 impl?

[leborchuk]

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 ...

[leborchuk]

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.

[yjhjstz]

can we use RegisterCustomScanMethods and hook to impl anser as extension ?

[leborchuk]

can we use RegisterCustomScanMethods and hook to impl anser as extension ?

Yes, thank you, really I could move it to the extension. The only tricky moment is with the authentication but I think I just could add hook.

Converted PR to draft to rewrite code and tests.

[leborchuk]

Could be reviewed, thank you.

[yjhjstz]

Alternative: reuse the dispatch connection instead of a backward libpq connection

The QD↔QE dispatch connection is already bidirectional mid-execution. nextval() on a QE does exactly this: it sends a NOTIFY to the QD over its own frontend connection and waits for the reply on the same socket (src/backend/commands/sequence.c:2117-2176). The QD handles it inside the dispatcher's receive loop and writes the answer back with a custom message type (cdbdisp_async.c:1160-1195, send_sequence_response at :972). Bloom filter exchange is the same pattern with N producers and M consumers.

Sketch:

1. Producer QE finishes its scan and sends NotifyMyFrontEnd("anser_filter", payload). Payload has to be a C string, so hex/base64 encode the filter.
2. QD merges per channel in processResults. Producer count is known from the gang size. When complete, write the merged filter to every consumer connection via pqPutMsgStart with a custom type. No libpq change is needed for QD→QE since the QE reads raw with pq_getmessage.
3. Consumer QE waits for that message on first ExecProcNode, using WaitLatchOrSocket on MyProcPort->sock rather than the busy loop nextval uses.

What this removes: the segment→QD connection, the pg_hba bypass hooks, the token hash, QD shared memory and background workers, the three catalog functions and the catversion bump. Cancellation and error propagation come for free: any QE error makes the QD cancel all gangs, and the consumer's wait loop exits via CHECK_FOR_INTERRUPTS. That also resolves open question 3 about the consumer having no timeout.