GitHub user leborchuk created a discussion: [Ideas] Implement Robust Predicate Transfer (RPT+) on top of Anser
### Description # The references, precisely | What | Where | Note | | --- | --- | --- | | **RPT paper** | Junyi Zhao et. all Debunking the Myth of Join Ordering: Toward Robust SQL Analytics | https://arxiv.org/pdf/2502.15181 | | **DuckDB RPT extension** | `https://github.com/robust-sql/robust` | this is the *baseline* RPT | | **RPT+ paper** | Yiming Qiao, Peter Boncz, Huanchen Zhang, *Robust Predicate Transfer with Dynamic Execution* | https://vldb.org/pvldb/vol19/p1278-qiao.pdf | | **DuckDB RPT+ extension** | `https://github.com/embryo-labs/dynamic-predicate-transfer` | this is RPT+ | Read both repos, but know which is which: `robust-sql/robust` implements plain Predicate Transfer (DuckDB's native bloom filter, 12 bits/key; reports 1.76× geomean on JOB), while the paper is specifically about fixing RPT's *overhead and regressions*. ⚠️ **Licensing.** Cloudberry is Apache-2.0. Check the license of both repos before reusing even a snippet, and prefer implementing from the paper. If you do reuse anything, it needs attribution plus `LICENSE`/`NOTICE` handling — raise it in the PR rather than deciding alone. ## What the paper actually proposes **RPT** (the baseline) makes Yannakakis' algorithm practical: model the query as a join graph `G(V,E)` where edge weight = number of shared join attributes, build a maximum spanning tree rooted at the **largest** table (`LargestRoot`), then propagate bloom filters over that tree in a **forward pass** (leaves→root) and a **backward pass** (root→leaves), fully reducing tables before the join phase. Its weakness: it ignores equivalence relationships between join keys, producing **redundant** and **oversized** filters, and pays heavy scan + BF-construction overhead — enough to cause real regressions. **RPT+** adds three things: 1. **Asymmetric Transfer Plan (ATP)** — different tree shapes for the two passes, because they do different jobs. Forward = *collect* information, so a **deep chaining** tree is best (filter table *i* before building the filter for *i+1*, avoiding oversized BFs). Backward = *distribute* information, so a **wide broadcast** tree is best (build the final filter **once** and reuse it for every table in the same equivalence class). Chains for different equivalence classes are linked through "bridge tables" (the root of each class's induced subtree). 2. **Cascade filters** — min/max for block-level skipping plus a bloom filter for tuple-level filtering, applied coarse-to-fine. 3. **Dynamic pipelines** — runtime selectivity monitoring that *abandons* filter construction when it will not pay off (Algorithm 2 in §5.2). Reported speedups over DuckDB v1.3.0: **1.47× JOB, 1.28× SQLStorm, 1.17× TPC-H, 1.01× Appian**, "avoid[ing] the significant performance regressions observed with the original RPT". **Constants from §5.5, use these as starting values, do not invent your own:** | Parameter | Value | Meaning | | --- | --- | --- | | γ | 100 K tuples | rows observed before deciding | | τ_sel | 0.35 | abandon if > 35% of tuples survive | | τ_prog | 0.6 | only abandon while < 60% through the scan | | M_avail | 64 GB | memory budget for materialization | | τ_stop | 0.9 | stop *probing* a filter that rejects too little | | filter | **20 bits/key, 7 hash functions** | FPR 6.1×10⁻⁵ | | probe cost | 2.48 cycles/tuple | 64-byte cache-line blocks, 32-bit sectors, `VPGATHERDD` | Sampling is **sequential, not random** — random access breaks streaming execution. On JOB, τ_sel in 0.1-0.3 gave the best speedups; τ_sel = 0 (drop almost everything) *hurt*. ## Why Anser fits — and the one place it does not | RPT+ concept | Anser today | Gap | | --- | --- | --- | | equivalence class of join attributes | `condition_key` — `anser.h` documents it as exactly "the optimizer-generated equivalence-class symbols" | naming only | | backward pass: build the final filter **once**, broadcast to every table in the class | one channel per condition, union once at the coordinator, deliver to every registered consumer | **already implemented** | | combining filters built by different workers | gather service folds parts by bitwise OR | **already implemented** (the paper is single-node; this is our addition) | | forward pass: **chaining**, table *i* reduced before it builds the filter for *i+1* | channels are independent, with no ordering between them | **the hard part — see below** | | dynamic pipeline abandonment (Alg. 2) | never abandons | overlaps Issue 2 task 2.4 | | cascade filter, blocked/sectorized BF | classic bloom: k random bitset accesses | overlaps Issue 2 tasks 2.1/2.3 | | one filter per (table, equivalence class) | plan pass injects one producer+consumer per hash join | needs join-graph analysis in `anserplan.c` | Anser's backward pass is essentially free — it is what the extension already does. **The forward chaining pass is where this project lives or dies**, for a reason that does not exist in DuckDB: - In DuckDB, a chain hop is an in-process pipeline breaker: microseconds of coordination. - In Anser, every hop is a **global barrier**: all segments publish → coordinator unions → all segments receive. One hop costs a full segment→QD→segment rendezvous. JOB queries join 5-10 tables, i.e. 4-9 sequential hops. If one hop costs ~30 ms, a 7-table chain adds ~200 ms of pure latency to a query whose *total* runtime may be under a second. That arithmetic, with a measured hop latency, decides whether M3 below is worth attempting at all. ⚠️ **Correctness rule you must not get wrong.** It is tempting to do the forward chaining locally on each segment and avoid the barrier. That is **only correct when the tables are co-distributed on the join key.** Otherwise segment *s* holds only part of table *j*'s key set, and filtering table *j+1* with it drops rows whose key lives on another segment — **wrong results, silently.** Where tables are not co-located, each hop must be global. Write this rule into the design note, with the co-location test, before any code. ## Milestones ### M0 — Read, map, measure the budget (blocking; no implementation) Design-note deliverable: 1. A corrected version of the mapping table above, after reading the paper and both repos. 2. **Measured single-hop Anser latency** (publish → union → deliver) at 3, 8 and 24 segments, using Issue 1's instrumentation. Then the projected cost of a 4-hop and 8-hop chain. State whether M3 is viable. 3. The co-location rule: how to detect from the plan that two tables are distributed on the same key, and how many JOB / TPC-DS joins qualify. 4. Where the join graph, equivalence classes and predicates come from. Anser's pass runs *post-plan* (`AnserApplyRuntimeFilters`), which sees a finished `PlannedStmt`, not the join graph — can equivalence classes be recovered there, or does this need a pre-plan hook? **This is the second make-or-break question.** ORCA and the Postgres planner will differ; report both. 5. Sizing: RPT creates one filter per (table, class), so a 10-table query needs many more channels than today's one-per-hash-join. Check `anser.max_channels` auto-sizing (`max_connections × gp_max_slices`) and `anser.max_consumers_per_channel` (default 64) against `segments × tables_in_class`. Propose new defaults if needed. Do not proceed past M0 without review. "The forward pass costs more than it saves in an MPP" is a legitimate and valuable outcome. ### M1 — Backward broadcast pass across an equivalence class (start here) Generalize today's one-producer-set/one-consumer-per-hash-join to: **all tables in an equivalence class publish, the union is built once, every table in that class receives it.** No ordering, no barrier beyond the one Anser already has, and it is exactly the pass RPT+ shows is best served by a broadcast tree. Expect the bulk of the achievable win here. ### M2 — Dynamic pipeline (Algorithm 2) Implement the abandon logic with the paper's constants, on the producer side. Note this is **the same mechanism as Issue 2 task 2.4** — coordinate, do not implement it twice; whoever gets there first owns it, the other reviews. Anser already has the cancel path (`ExecAnserBloomFilterProduceCancel()`), but mind the semantics: an Anser cancel kills the channel for *every* consumer, whereas RPT+ abandons one filter locally. Decide and document which you want. ### M3 — Asymmetric transfer plan (forward chaining) — only if M0 allows Chaining forward tree + broadcast backward tree, bridge tables between classes. Restrict chain depth by what M0's latency budget supports, and say what depth you allowed and why. ### M4 — Cascade filter min/max block skipping + a blocked/sectorized bloom layout. Cross-link to Issue 2: RPT+ affords **20 bits/key and 7 hashes** precisely because a blocked filter touches **one cache line per probe** (2.48 cycles/tuple), while Anser's classic bloom does k random accesses. That suggests the layout matters more than the bits/key knob — and a blocked filter is still unionable by bitwise OR when both sides share parameters, so Anser's cheap fold survives. Verify that claim before relying on it. ## Benchmarks - **JOB (Join Order Benchmark, IMDb) is mandatory** — it is the paper's headline workload and this technique's home turf. Loading IMDb into Cloudberry with sensible distribution keys is part of the task; commit the DDL under `gpcontrib/anser/doc/bench/`. - **TPC-DS SF100 minimum**, using the reporting protocol from Issue 2 (injected subset in detail, non-injected regression check, geomean). - Report **regressions first, wins second.** RPT's entire failure mode was regressions, and RPT+'s main claim is removing them; a report that shows the geomean win without a per-query regression table does not answer the question this work exists to answer. - Compare against the paper's numbers (1.47× JOB) as a sanity check, but expect to differ: we are distributed, our hops cost network round trips, and our filter implementation is not theirs. Explain your gap rather than matching it. ## Definition of done - [ ] M0 design note, reviewed: mapping, measured hop latency and chain budget, co-location correctness rule, join-graph feasibility for both optimizers, channel sizing. - [ ] M1 implemented behind a GUC, default off, fail-open on every failure path, visible in `EXPLAIN ANALYZE`. - [ ] Results identical to the unmodified plan: full result-set comparison (not row counts) on the regression suite, JOB and TPC-DS. - [ ] JOB + TPC-DS reports in `gpcontrib/anser/doc/performance.md`, regressions reported per query. - [ ] Licensing question answered explicitly in the PR. ## Out of scope - **Cyclic queries.** Yannakakis, RPT and RPT+ all target acyclic (α-acyclic) queries. Detect and skip the rest; do not try to generalize. - The full semi-join reduction *guarantee* — RPT+ itself trades it away ("restricts the forward transfer plan to tables with predicates, plus bridge tables … trades the theoretical guarantee for significantly reduced overhead"). We are following it there, not doing better. - Rewriting the join order or the optimizer. This technique exists to make a *bad* join order survivable, not to fix it. - Porting DuckDB code. See the licensing note. ### Use case/motivation _No response_ ### Related issues _No response_ ### Are you willing to submit a PR? - [ ] Yes I am willing to submit a PR! GitHub link: https://github.com/apache/cloudberry/discussions/1963 ---- 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]
