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]

Reply via email to