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

what's difference with 6c41d27 impl?

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

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.

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

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.

Could be reviewed, thank you.

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.

Good to know we have a MVP for runtime filter of intra-slice.

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.

It's brilliant! I converted PR to a draft and try to implement it. This should solve a lot of problems - we won't need to use DSM (and its expensive mmap calls!), everything will be in the query memory context (and will be released after the query ends), there will be no need for background workers (and we will not be limited to 1 CPU; each query will process its own data using its own process). And, of course, there is no need for a cumbersome token-based authentication mechanism.

Great, another questions concerned:

1. how to support parallel scan , parallel hash join ?
2. some benchmark of TPC-DS ?

Thank you, working on it (expect show here some results within the coming week)