Anser bloomfilter MVP

[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

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..1282.56 rows=2000 width=384) (actual time=76.511..582.788 rows=1000 loops=1)
   ->  Hash Right Join  (cost=0.00..1279.69 rows=667 width=384) (actual time=75.725..581.356 rows=380 loops=1)
         Hash Cond: (a.id = aef.id)
         Extra Text: (seg0)   Hash chain length 10.0 avg, 10 max, using 38 of 262144 buckets.
         ->  Custom Scan (Anser Bloom Consumer)  (cost=0.00..499.02 rows=333334 width=353) (actual time=30.858..528.642 rows=38 loops=1)
               Bloom Filter Size: 8192 bytes
               Rows Removed by Bloom Filter: 333129
               ->  Seq Scan on applications a  (cost=0.00..499.02 rows=333334 width=353) (actual time=1.817..452.622 rows=334042 loops=1)
         ->  Hash  (cost=431.06..431.06 rows=334 width=31) (actual time=44.433..44.435 rows=380 loops=1)
               Buckets: 262144  Batches: 1  Memory Usage: 2072kB
               ->  Redistribute Motion 3:3  (slice2; segments: 3)  (cost=0.00..431.06 rows=334 width=31) (actual time=1.731..44.301 rows=380 loops=1)
                     Hash Key: aef.id
                     ->  Custom Scan (Anser Bloom Producer)  (cost=0.00..431.01 rows=334 width=31) (actual time=0.367..42.216 rows=500 loops=1)
                           Bloom Filter Size: 8192 bytes
                           ->  Seq Scan on applications_extra_fields aef  (cost=0.00..431.01 rows=334 width=31) (actual time=0.364..0.585 rows=500 loops=1)
 Planning Time: 30.705 ms
   (slice0)    Executor memory: 140K bytes.
   (slice1)    Executor memory: 5314K bytes avg x 3x(0) workers, 5314K bytes max (seg0).  Work_mem: 2072K bytes max.
   (slice2)    Executor memory: 430K bytes avg x 3x(0) workers, 430K bytes max (seg0).
 Memory used:  128000kB
 Optimizer: GPORCA
 Execution Time: 584.787 ms

The main architecture overview

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

[Copilot]

Pull request overview

Introduces the initial “Anser” adaptive information-sharing subsystem scaffolding in the backend (shared-memory channel map + GUCs + coordinator bgworkers) along with a dedicated SQL-callable test extension and regression coverage.

Changes:

• Adds core Anser shared-memory structures/APIs and wires shmem sizing/init into postmaster startup paths.
• Registers two new coordinator-only background workers and adds new Anser-related GUCs.
• Adds src/test/modules/anser test extension + regression test, and hooks it into both Makefile and Meson test-module builds.

Reviewed changes

Copilot reviewed 27 out of 28 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
src/test/modules/meson.build	Adds the anser test module to Meson test module build.
src/test/modules/Makefile	Adds the anser test module to Makefile-based test module build.
src/test/modules/anser/test_anser.control	Declares the test_anser extension metadata.
src/test/modules/anser/test_anser--1.0.sql	Defines SQL wrappers for Anser test helper C functions.
src/test/modules/anser/sql/test_anser.sql	Adds regression SQL to exercise register/subscribe/publish/consume/cancel.
src/test/modules/anser/meson.build	Builds and installs test_anser for Meson; registers regression.
src/test/modules/anser/Makefile	Builds and installs test_anser for Makefile; registers regression.
src/test/modules/anser/expected/test_anser.out	Expected regression output for test_anser.
src/test/modules/anser/anser_test.c	Implements SQL-callable test helpers that wrap Anser APIs.
src/include/utils/unsync_guc_name.h	Registers new gp_anser_* GUC names in the “no sync” list.
src/include/postmaster/postmaster.h	Increases MaxPMAuxProc base to account for new aux workers.
src/include/cdb/anser.h	Adds public Anser API and shared-memory structs/GUC externs.
src/backend/utils/misc/guc_gp.c	Adds gp_anser_* GUC definitions and includes Anser header.
src/backend/storage/lmgr/lwlocknames.txt	Adds two LWLock names for Anser.
src/backend/storage/ipc/ipci.c	Hooks Anser shmem sizing/init into shared-memory creation paths.
src/backend/storage/ipc/ipc.c	Changes proc_exit() gprof directory chdir() handling.
src/backend/postmaster/postmaster.c	Registers two new Anser background workers in PMAuxProcList.
src/backend/cdb/Makefile	Adds anser to cdb backend subdirectories.
src/backend/cdb/anser/Makefile	Adds Makefile for building Anser backend objects.
src/backend/cdb/anser/anserservice.c	Adds (currently skeletal) gather/send background worker mains.
src/backend/cdb/anser/anser.c	Implements Anser shared-memory channel map and core APIs.
src/backend/cdb/anser/anser-pr1-plan.md	Adds in-repo design/plan documentation for PR1.

---

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

[Copilot]

Pull request overview

Copilot reviewed 48 out of 49 changed files in this pull request and generated 1 comment.

Suppressed comments (3)

src/backend/cdb/anser/anserclient.c:424

• AnserClientConsumeWait() allocates buf even when the caller passes payload == NULL, which will leak memory for every successful delivery in that call pattern (e.g., if a caller only cares about cancelled).

		int			len = PQgetlength(res, 0, 0);
		char	   *val = PQgetvalue(res, 0, 0);
		void	   *buf = NULL;

		if (len > 0)
		{
			buf = palloc(len);
			memcpy(buf, val, len);
		}
		if (payload != NULL)
			*payload = buf;
		if (payload_len != NULL)
			*payload_len = (Size) len;
		ok = true;

src/backend/cdb/anser/anser-pr1-plan.md:251

• The Anser PR plan doc says gp_anser_max_info_size defaults to 1MB, but the GUC is defined with a default of 16 * 1024 * 1024 in src/backend/utils/misc/guc_gp.c. This mismatch will mislead operators sizing shared memory.

| `gp_anser_enable` | `off` | `PGC_POSTMASTER` | Master switch. When off: **no shmem allocated, no workers registered**; enabling requires a restart. |
| `gp_anser_max_channels` | `128` | `PGC_POSTMASTER` | Channel hash size / arena slot count (sizes shmem) |
| `gp_anser_max_info_size` | `1MB` | `PGC_POSTMASTER` | Per-record memory cap, per the paper (sizes shmem) |
| `gp_anser_timeout_ms` | `1000` | `PGC_USERSET` | Weak-dependency wait budget for consumers |

src/backend/cdb/anser/anserplanexec.c:465

• Fail-open is expected to happen in benign cases (e.g., cancellation/timeouts or partial delivery). Logging it at LOG will spam server logs for any query that hits this path, potentially at high frequency when gp_anser_runtime_filter is enabled. This should be DEBUG-level or behind an explicit debug GUC.

	if (st->filter == NULL)
		elog(LOG,
			 "anser bloom consumer: no filter, failing open (cancelled=%d, received_parts=%u)",
			 ExecAnserBloomFilterConsumerWasCancelled(st->consume),
			 ExecAnserBloomFilterConsumerReceivedParts(st->consume));