diff --git a/CHANGELOG.md b/CHANGELOG.md index 64597979..93635133 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,24 @@ true until the next version shipped. ### Added +- A serial inner Hash Join can push the build-side keys into a direct + columnar scan (#752). + + The coordinator wraps core Hash Join. It does not keep a HashPath from + `joinrel->pathlist`. It builds a private path, drains the build side into a + tuplestore, then lets Hash replay that spool. The scan skips chunk groups + outside the conservative key interval when types and collations match. It + also rejects non-matching rows with a Bloom filter of those keys, using the + same saturation cap as on-disk bloom filters. + + The path is serial and INNER only. LEFT, SEMI, ANTI, CROSS, parallel, and + projection-backed outers are refused. `pgcolumnar.enable_join_runtime_filter` + is off by default until the skip is measured on the join fixture. + + `EXPLAIN (ANALYZE)` reports `Runtime Filter Groups Removed` and + `Runtime Filter Rows Rejected`. Those counters are dedicated. They are not + `InstrCountFiltered1`. + - Every check result is machine-readable, and counting a check is the same operation as recording it (#917). diff --git a/Makefile b/Makefile index 40c8eb0d..7f3a76aa 100644 --- a/Makefile +++ b/Makefile @@ -15,6 +15,7 @@ OBJS = \ src/columnar_reader.o \ src/columnar_delete_vector.o \ src/columnar_customscan.o \ + src/columnar_runtime_filter.o \ src/columnar_vector.o \ src/columnar_vacuum.o \ src/columnar_curve.o \ diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d98d52e3..e0e4daa6 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -202,6 +202,13 @@ the module adds the vectorized aggregate path for a supported `SELECT agg(col) FROM t [WHERE ...]`. EXPLAIN reporting (projected columns, and under ANALYZE the row groups and vectors read versus skipped) lives here. + +### columnar_runtime_filter.c +Serial join runtime filter. A `set_join_pathlist_hook` wraps a serial inner +Hash Join whose outer path is a direct columnar scan. Core Hash Join keeps the +answers. The coordinator drains the build side first. It then attaches a +conservative key range and a Bloom filter to that scan. + ### columnar_vector.c The vectorized aggregate path and its shared filter. A column-at-a-time filter (`ColumnarVecSelect`) turns a plan's simple strict `column op const` clauses into diff --git a/docs/configuration.md b/docs/configuration.md index 17089eb7..ddab390c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -61,6 +61,7 @@ disk. It never changes the values that a table returns. | `pgcolumnar.enable_group_vectorization` | boolean | `off` | Use the vectorized aggregate path for `GROUP BY` queries on a columnar table. Off by default; see [why grouped vectorization is off by default](#why-grouped-vectorization-is-off-by-default). | | `pgcolumnar.groupagg_max_groups` | integer | `1000000` | Cap on the group count the grouped vectorized aggregate builds. Over the cap the query errors. Range 1 to INT_MAX. | | `pgcolumnar.enable_bloom_filter` | boolean | `on` | Skip chunk groups on equality filters using per-chunk bloom filters. | +| `pgcolumnar.enable_join_runtime_filter` | boolean | `off` | Wrap a serial inner Hash Join so the build keys can skip fact-table groups and reject non-matching rows. Direct columnar scan only. Off until the skip is measured. | | `pgcolumnar.enable_read_stream` | boolean | `on` | Prefetch block reads with the read stream API. Effective on PostgreSQL 17 and later. | | `pgcolumnar.enable_ungrouped_vector_agg` | boolean | `off` | Answer an ungrouped aggregate (`count`, `sum`, `avg`, `min`, `max` with no `GROUP BY`) with a batch fold over decoded vectors instead of row-at-a-time. Off by default. | | `pgcolumnar.enable_parallel_vector_agg` | boolean | `off` | Let the ungrouped batch fold run as a parallel partial aggregate under `Gather`, each worker folding its own row groups. Requires `pgcolumnar.enable_ungrouped_vector_agg`. Off by default. | diff --git a/docs/features.md b/docs/features.md index f7500f65..7d3dad9b 100644 --- a/docs/features.md +++ b/docs/features.md @@ -70,6 +70,11 @@ settings see the [configuration reference](configuration.md); for constraints se - A fetch by row number decodes only the columns that the executor asks for. It keeps the decoded row group for the rest of the statement. An index-driven read of a wide table therefore does not decode the columns that it will not return. +- Serial join runtime filter for a star-schema Hash Join. + A serial inner Hash Join can skip fact-table groups using the build-side key range. + It can also reject non-matching rows with a Bloom filter of those keys. + The GUC `pgcolumnar.enable_join_runtime_filter` is off by default. + It does not wrap LEFT, SEMI, ANTI, CROSS, parallel, or projection scans. - Parallel scan across a table's row groups. - Read stream prefetch of block reads on PostgreSQL 17 and later (`pgcolumnar.enable_read_stream`). diff --git a/docs/how-to.md b/docs/how-to.md index 6fb39640..46f112cd 100644 --- a/docs/how-to.md +++ b/docs/how-to.md @@ -132,6 +132,23 @@ then sit in few chunk groups. Read `Columnar Chunk Groups Removed by Filter` to confirm the skip. A scattered high-cardinality column gains little. Turn the feature off with `SET pgcolumnar.enable_bloom_filter = off` to compare. + +## Skip fact-table work under a star-schema join + +A serial inner Hash Join can push the dimension keys into the fact-table scan. +The scan drops chunk groups outside the build-side key range. +It also rejects rows whose keys are absent from a Bloom filter of those keys. + +```sql +EXPLAIN (ANALYZE) SELECT sum(amount) FROM fact JOIN dim ON fact.k = dim.k; +``` + +**Tuning.** It is off by default (`pgcolumnar.enable_join_runtime_filter`). +It applies only to a serial inner Hash Join whose outer path is a direct columnar scan. +A LEFT, SEMI, ANTI, or CROSS join is unchanged. +A covering projection is also unchanged. +Read `Runtime Filter Groups Removed` and `Runtime Filter Rows Rejected` to confirm the skip. + ## Add a projection for a second sort order A table has one physical sort order. A projection stores a column subset a second diff --git a/docs/limitations.md b/docs/limitations.md index 5f31c21e..d56e8a60 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -679,6 +679,18 @@ path rejects. `pgcolumnar.groupagg_max_groups` caps the group count, default that exceeds it errors rather than switching plans. Raise the cap or turn the path off. + +## Join runtime filter + +The runtime filter wraps a serial inner Hash Join only. +The outer path must be a direct columnar scan. +LEFT, SEMI, ANTI, and CROSS joins are unchanged. +A parallel Hash Join is unchanged. +A covering projection as the outer path is unchanged. +A mixed-type or mixed-collation join still hashes both sides. +It does not attach a key-range skip in those cases. +A build side past the on-disk bloom saturation cap disables Bloom rather than emitting a saturated filter. + ## Skipping and collation A pushed-down filter drives chunk-group skipping only when one condition is true. diff --git a/docs/user-guide.md b/docs/user-guide.md index a946398c..39dfb5c4 100644 --- a/docs/user-guide.md +++ b/docs/user-guide.md @@ -137,6 +137,7 @@ controlled by a setting in the [Configuration reference](configuration.md): - Vectorized aggregate. The zone-map metadata answers an ungrouped count, sum, avg, min, or max on a supported type. - `count(*)` answered from catalog metadata when there is no filter. +- Join runtime filter: a serial inner Hash Join can skip fact-table groups and reject non-matching rows using the build-side keys. #### Reading the filter counters diff --git a/src/columnar.h b/src/columnar.h index a17d5f9b..1b0b1fbd 100644 --- a/src/columnar.h +++ b/src/columnar.h @@ -667,6 +667,11 @@ extern void PgColumnarReadSetParallelCounter(PgColumnarReadState *readState, * scan's EXPLAIN output to show how many chunk groups the min/max skip lists * removed. total = read + skipped over the groups the scan has reached. */ +extern bool PgColumnarReadSetRuntimeRange(PgColumnarReadState *readState, + AttrNumber attno, Oid subtype, + Datum minimum, Datum maximum); +extern void PgColumnarReadClearRuntimeRange(PgColumnarReadState *readState); +extern uint64 PgColumnarRuntimeGroupsRemoved(PgColumnarReadState *readState); extern void PgColumnarReadStats(PgColumnarReadState *readState, uint64 *groupsRead, uint64 *groupsSkipped, uint64 *groupsTotal); @@ -918,6 +923,15 @@ extern void PgColumnarSerializeFlushRows(uint64 storageId, const uint64 *rows, * a scanrelid==0 upper node is the vectorized aggregate. */ extern const CustomScanMethods pgcolumnar_scan_methods; +extern bool pgcolumnar_enable_join_runtime_filter; +extern void PgColumnarRuntimeFilterInit(void); +extern void PgColumnarAttachRuntimeBloom(PlanState *scanState, + void *filter, AttrNumber attno); +extern bool PgColumnarRuntimeBloomMatch(void *filter, Datum value, bool isNull); +extern bool PgColumnarAttachRuntimeRange(PlanState *scanState, + AttrNumber attno, Oid subtype, + Datum minimum, Datum maximum); +extern void PgColumnarDetachRuntimeRange(PlanState *scanState); extern Node *PgColumnarCreateAggScanState(CustomScan *cscan); extern Node *PgColumnarCreateGroupAggScanState(CustomScan *cscan); diff --git a/src/columnar_customscan.c b/src/columnar_customscan.c index 302a7885..d795950d 100644 --- a/src/columnar_customscan.c +++ b/src/columnar_customscan.c @@ -123,6 +123,10 @@ typedef struct PgColumnarCustomScanState Datum *projValues; /* scratch, length K+1 (index 0 = rownumber) */ bool *projNulls; PgColumnarLivenessCache *livenessCache; /* cached base liveness for the scan */ + bool runtimeRangeAttached; + void *runtimeBloom; + AttrNumber runtimeBloomAttno; + uint64 runtimeRowsRejected; } PgColumnarCustomScanState; /* path -> plan */ @@ -3200,16 +3204,34 @@ pgcolumnar_setup_late_materialization(PgColumnarCustomScanState *cstate, * is reset per evaluation exactly as ExecScan resets it per fetched tuple, so a * scan that rejects millions of rows does not accumulate their qual allocations. */ +static bool +pgcolumnar_runtime_bloom_keeps(PgColumnarCustomScanState *cstate, + TupleTableSlot *slot) +{ + if (cstate->runtimeBloom == NULL) + return true; + if (PgColumnarRuntimeBloomMatch(cstate->runtimeBloom, + slot->tts_values[cstate->runtimeBloomAttno - 1], + slot->tts_isnull[cstate->runtimeBloomAttno - 1])) + return true; + cstate->runtimeRowsRejected++; + return false; +} + static bool pgcolumnar_scan_row_filter(void *arg) { ScanState *ss = (ScanState *) arg; + PgColumnarCustomScanState *cstate = (PgColumnarCustomScanState *) ss; ExprContext *econtext = ss->ps.ps_ExprContext; TupleTableSlot *slot = ss->ss_ScanTupleSlot; ExecClearTuple(slot); ExecStoreVirtualTuple(slot); + if (!pgcolumnar_runtime_bloom_keeps(cstate, slot)) + return false; + ResetExprContext(econtext); econtext->ecxt_scantuple = slot; @@ -3245,12 +3267,16 @@ static bool pgcolumnar_scan_row_filter_nocount(void *arg) { ScanState *ss = (ScanState *) arg; + PgColumnarCustomScanState *cstate = (PgColumnarCustomScanState *) ss; ExprContext *econtext = ss->ps.ps_ExprContext; TupleTableSlot *slot = ss->ss_ScanTupleSlot; ExecClearTuple(slot); ExecStoreVirtualTuple(slot); + if (!pgcolumnar_runtime_bloom_keeps(cstate, slot)) + return false; + ResetExprContext(econtext); econtext->ecxt_scantuple = slot; @@ -3404,9 +3430,22 @@ PgColumnarScanNext(ScanState *ss) ExecClearTuple(slot); } - else if (!PgColumnarReadNextRow(cstate->readState, slot->tts_values, - slot->tts_isnull, &rowNumber)) - return NULL; + else + { + /* + * When late materialization is off, the Bloom probe cannot run in the + * two-pass filter: that path is what would leave qual columns undecoded. + * Probe after the full row is built instead. + */ + for (;;) + { + if (!PgColumnarReadNextRow(cstate->readState, slot->tts_values, + slot->tts_isnull, &rowNumber)) + return NULL; + if (pgcolumnar_runtime_bloom_keeps(cstate, slot)) + break; + } + } ExecStoreVirtualTuple(slot); PgColumnarRowNumberToItemPointer(rowNumber, &slot->tts_tid); @@ -3434,6 +3473,7 @@ PgColumnarReScanCustomScan(CustomScanState *node) { PgColumnarCustomScanState *cstate = (PgColumnarCustomScanState *) node; + cstate->runtimeRowsRejected = 0; if (cstate->readState != NULL) { PgColumnarRescanRead(cstate->readState); @@ -3709,9 +3749,150 @@ PgColumnarExplainCustomScan(CustomScanState *node, List *ancestors, ExplainPropertyInteger("Columnar Rows Filtered Before Materialization", NULL, (int64) PgColumnarRowsFilteredEarly(cstate->readState), es); + if (cstate->runtimeRangeAttached) + ExplainPropertyInteger("Runtime Filter Groups Removed", NULL, + (int64) PgColumnarRuntimeGroupsRemoved(cstate->readState), + es); + if (cstate->runtimeBloom != NULL) + ExplainPropertyInteger("Runtime Filter Rows Rejected", NULL, + (int64) cstate->runtimeRowsRejected, + es); } } +/* + * pgcolumnar_ensure_runtime_bloom_columns + * The two-pass filter must decode every column ExecQual reads, plus the + * join key. Turning late materialization on with only the key marked + * drops rows whose qual names any other column. + * + * Do not force the path when the GUC is off or the qual is volatile. Bloom + * then probes in PgColumnarScanNext after the full row is built. + */ +static void +pgcolumnar_ensure_runtime_bloom_columns(PgColumnarCustomScanState *state, + AttrNumber attno) +{ + CustomScan *cscan = (CustomScan *) state->css.ss.ps.plan; + Bitmapset *qualAttrs = NULL; + int natts = state->nTotalColumns; + int x = -1; + + if (!pgcolumnar_enable_late_materialization) + return; + + if (state->qualCols != NULL) + { + state->qualCols[attno - 1] = true; + return; + } + + if (cscan->scan.plan.qual != NIL && + contain_volatile_functions((Node *) cscan->scan.plan.qual)) + return; + + if (cscan->scan.plan.qual != NIL) + { + pull_varattnos((Node *) cscan->scan.plan.qual, cscan->scan.scanrelid, + &qualAttrs); + while ((x = bms_next_member(qualAttrs, x)) >= 0) + { + AttrNumber qattno = x + FirstLowInvalidHeapAttributeNumber; + + if (qattno <= 0 || qattno > natts) + return; + } + } + + state->qualCols = palloc0(sizeof(bool) * natts); + x = -1; + while ((x = bms_next_member(qualAttrs, x)) >= 0) + { + AttrNumber qattno = x + FirstLowInvalidHeapAttributeNumber; + + state->qualCols[qattno - 1] = true; + } + state->qualCols[attno - 1] = true; + state->lateMat = true; +} + +/* + * PgColumnarAttachRuntimeBloom + * Publish a completed build-side Bloom filter to a direct base scan + * before its first tuple is requested. Projection scans are excluded + * by the planner; checking again here turns a planner mistake into an + * error, not a wrong answer. + */ +void +PgColumnarAttachRuntimeBloom(PlanState *scanState, void *filter, AttrNumber attno) +{ + PgColumnarCustomScanState *state; + + if (scanState == NULL) + return; + if (!IsA(scanState, CustomScanState)) + elog(ERROR, "pgcolumnar runtime bloom expected a custom scan"); + state = (PgColumnarCustomScanState *) scanState; + if (state->css.methods != &pgcolumnar_exec_methods) + elog(ERROR, "pgcolumnar runtime bloom expected a columnar scan"); + + if (filter == NULL) + { + state->runtimeBloom = NULL; + state->runtimeBloomAttno = InvalidAttrNumber; + return; + } + if (state->projScan || state->readState == NULL) + elog(ERROR, "pgcolumnar runtime bloom expected a direct base scan"); + if (attno <= 0 || attno > state->nTotalColumns) + elog(ERROR, "pgcolumnar runtime bloom key is out of range"); + + state->runtimeBloom = filter; + state->runtimeBloomAttno = attno; + state->runtimeRowsRejected = 0; + pgcolumnar_ensure_runtime_bloom_columns(state, attno); +} + +/* + * PgColumnarAttachRuntimeRange + * Publish a completed build-side hull to a direct base scan before its + * first tuple is requested. Projection scans are excluded by the planner; + * checking again here turns a planner mistake into an error, not a wrong + * answer. + */ +bool +PgColumnarAttachRuntimeRange(PlanState *scanState, AttrNumber attno, Oid subtype, + Datum minimum, Datum maximum) +{ + PgColumnarCustomScanState *state; + + if (!IsA(scanState, CustomScanState)) + elog(ERROR, "pgcolumnar runtime range expected a custom scan"); + state = (PgColumnarCustomScanState *) scanState; + if (state->css.methods != &pgcolumnar_exec_methods || state->projScan || + state->readState == NULL) + elog(ERROR, "pgcolumnar runtime range expected a direct base scan"); + + state->runtimeRangeAttached = + PgColumnarReadSetRuntimeRange(state->readState, attno, subtype, + minimum, maximum); + return state->runtimeRangeAttached; +} + +void +PgColumnarDetachRuntimeRange(PlanState *scanState) +{ + PgColumnarCustomScanState *state; + + if (scanState == NULL || !IsA(scanState, CustomScanState)) + return; + state = (PgColumnarCustomScanState *) scanState; + if (state->css.methods != &pgcolumnar_exec_methods || state->readState == NULL) + return; + PgColumnarReadClearRuntimeRange(state->readState); + state->runtimeRangeAttached = false; +} + /* ------------------------------------------------------------------------- * registration * ------------------------------------------------------------------------- */ diff --git a/src/columnar_reader.c b/src/columnar_reader.c index 30187ed2..b980d020 100644 --- a/src/columnar_reader.c +++ b/src/columnar_reader.c @@ -68,6 +68,7 @@ typedef struct SkipPredicate * would have made on a group that is thrown away anyway. */ uint64 excludes; + bool runtimeFilter; /* owned by the join runtime filter */ } SkipPredicate; struct PgColumnarReadState @@ -136,6 +137,9 @@ struct PgColumnarReadState /* chunk-group skip counters over the groups reached so far (spec 9) */ uint64 groupsRead; uint64 groupsSkipped; + uint64 runtimeGroupsRemoved; + int runtimePredicateStart; + int runtimePredicateCount; /* * Native format (PGCN v1) read state. The scan reads row groups and column @@ -592,6 +596,7 @@ PgColumnarBeginReadWithStorage(Relation rel, Snapshot snapshot, readState->started = false; readState->exhausted = false; + readState->runtimeGroupsRemoved = 0; readState->parallelScan = parallelScan; readState->readContext = readContext; readState->stripeContext = AllocSetContextCreate(readContext, @@ -854,6 +859,101 @@ pgcolumnar_build_predicates(PgColumnarReadState *readState, int nkeys, ScanKey k readState->predOrder[i] = i; } + +/* + * PgColumnarReadSetRuntimeRange + * Attach or replace the two conservative bounds derived from a fully + * spooled hash-join build side. The bounds use the same predicate builder + * as ordinary scan keys, so cross-type comparison and domain handling have + * exactly one implementation. + */ +bool +PgColumnarReadSetRuntimeRange(PgColumnarReadState *readState, + AttrNumber attno, Oid subtype, + Datum minimum, Datum maximum) +{ + ScanKeyData keys[2]; + SkipPredicate built[2]; + SkipPredicate *predicates; + int *order; + int oldCount = readState->numPredicates; + int start = readState->runtimePredicateStart; + int builtCount; + MemoryContext oldContext; + + MemSet(keys, 0, sizeof(keys)); + MemSet(built, 0, sizeof(built)); + /* + * ScanKeyEntryInitialize with InvalidOid procedure is legal only for a + * NULL-search key. The predicate builder reads these six fields only. + */ + keys[0].sk_flags = 0; + keys[0].sk_attno = attno; + keys[0].sk_strategy = BTGreaterEqualStrategyNumber; + keys[0].sk_subtype = subtype; + keys[0].sk_argument = minimum; + keys[1].sk_flags = 0; + keys[1].sk_attno = attno; + keys[1].sk_strategy = BTLessEqualStrategyNumber; + keys[1].sk_subtype = subtype; + keys[1].sk_argument = maximum; + + builtCount = pgcolumnar_make_predicates(built, 2, keys, + readState->tupdesc, + readState->natts, + readState->readContext); + if (builtCount != 2) + return false; + + if (readState->runtimePredicateCount == 2) + { + built[0].runtimeFilter = true; + built[1].runtimeFilter = true; + readState->predicates[start] = built[0]; + readState->predicates[start + 1] = built[1]; + return true; + } + + oldContext = MemoryContextSwitchTo(readState->readContext); + predicates = palloc0(sizeof(SkipPredicate) * (oldCount + 2)); + order = palloc0(sizeof(int) * (oldCount + 2)); + if (oldCount > 0) + { + memcpy(predicates, readState->predicates, + sizeof(SkipPredicate) * oldCount); + memcpy(order, readState->predOrder, sizeof(int) * oldCount); + } + built[0].runtimeFilter = true; + built[1].runtimeFilter = true; + predicates[oldCount] = built[0]; + predicates[oldCount + 1] = built[1]; + order[oldCount] = oldCount; + order[oldCount + 1] = oldCount + 1; + readState->predicates = predicates; + readState->predOrder = order; + readState->numPredicates = oldCount + 2; + readState->runtimePredicateStart = oldCount; + readState->runtimePredicateCount = 2; + MemoryContextSwitchTo(oldContext); + return true; +} + +void +PgColumnarReadClearRuntimeRange(PgColumnarReadState *readState) +{ + if (readState == NULL || readState->runtimePredicateCount == 0) + return; + readState->numPredicates = readState->runtimePredicateStart; + readState->runtimePredicateCount = 0; +} + + +uint64 +PgColumnarRuntimeGroupsRemoved(PgColumnarReadState *readState) +{ + return readState == NULL ? 0 : readState->runtimeGroupsRemoved; +} + /* * pgcolumnar_group_can_match * Decide whether a chunk group could contain a row satisfying every @@ -1498,6 +1598,8 @@ pgcolumnar_predicate_excluded(PgColumnarReadState *rs, int oi) int ahead; rs->predicates[here].excludes++; + if (rs->predicates[here].runtimeFilter) + rs->runtimeGroupsRemoved++; if (oi == 0) return; @@ -4495,6 +4597,7 @@ PgColumnarRescanRead(PgColumnarReadState *readState) MemoryContextReset(readState->stripeContext); readState->started = false; readState->exhausted = false; + readState->runtimeGroupsRemoved = 0; /* * Reclaim the previous start's row-group list (#734). Clearing the pointer diff --git a/src/columnar_runtime_filter.c b/src/columnar_runtime_filter.c new file mode 100644 index 00000000..cc8e6348 --- /dev/null +++ b/src/columnar_runtime_filter.c @@ -0,0 +1,960 @@ +/*------------------------------------------------------------------------- + * + * columnar_runtime_filter.c + * Serial hash-join runtime-filter coordinator. + * + * The coordinator owns execution order while PostgreSQL core retains exact + * Hash Join semantics. Its child HashPath is private: a HashPath already in + * joinrel->pathlist can be freed by add_path(), so retaining or copying that + * same-rel path is unsafe. + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "columnar.h" +#include "columnar_customscan.h" + +#include "access/table.h" +#include "commands/explain.h" +#if PG_VERSION_NUM >= 180000 +#include "commands/explain_format.h" +#endif +#include "executor/executor.h" +#include "miscadmin.h" +#include "nodes/extensible.h" +#include "nodes/makefuncs.h" +#include "nodes/nodeFuncs.h" +#include "optimizer/cost.h" +#include "optimizer/pathnode.h" +#include "optimizer/paths.h" +#include "optimizer/restrictinfo.h" +#include "parser/parsetree.h" +#include "utils/lsyscache.h" +#include "utils/datum.h" +#include "utils/memutils.h" +#include "utils/typcache.h" +#include "utils/tuplestore.h" + +bool pgcolumnar_enable_join_runtime_filter = false; + +typedef struct PgColumnarRuntimeFilterState +{ + CustomScanState customScanState; + PlanState *joinState; + CustomScanState *tapState; + AttrNumber factAttno; + Oid factHashOid; + Oid hashCollation; + FmgrInfo factHashFn; + MemoryContext filterContext; + char *bloom; + uint32 bloomLength; + bool hashAvailable; + bool bloomEnabled; + bool hasValues; + bool buildEmpty; + bool rangeAttached; + bool prepared; +} PgColumnarRuntimeFilterState; + +typedef struct PgColumnarRuntimeTapState +{ + CustomScanState customScanState; + PlanState *sourceState; + Tuplestorestate *store; + TupleTableSlot *replaySlot; + AttrNumber keyResno; + Oid keyType; + Oid keyCollation; + FmgrInfo compareFn; + FmgrInfo buildHashFn; + MemoryContext valueContext; + uint32 *hashes; + uint32 hashCount; + uint32 hashCapacity; + Datum minimum; + Datum maximum; + int16 typeLength; + bool typeByValue; + bool intervalAvailable; + bool hashAvailable; + bool hasValues; + uint64 buildRows; + bool replay; +} PgColumnarRuntimeTapState; + +static set_join_pathlist_hook_type previous_set_join_pathlist_hook = NULL; + +static Plan *PgColumnarPlanRuntimeFilterPath(PlannerInfo *root, + RelOptInfo *rel, + CustomPath *bestPath, + List *tlist, + List *clauses, + List *customPlans); +static Node *PgColumnarCreateRuntimeFilterState(CustomScan *customScan); +static Node *PgColumnarCreateRuntimeTapState(CustomScan *customScan); +static void PgColumnarBeginRuntimeTap(CustomScanState *node, + EState *estate, + int eflags); +static TupleTableSlot *PgColumnarExecRuntimeTap(CustomScanState *node); +static void PgColumnarEndRuntimeTap(CustomScanState *node); +static void PgColumnarReScanRuntimeTap(CustomScanState *node); +static void PgColumnarExplainRuntimeTap(CustomScanState *node, + List *ancestors, + ExplainState *es); +static void PgColumnarBeginRuntimeFilter(CustomScanState *node, + EState *estate, + int eflags); +static TupleTableSlot *PgColumnarExecRuntimeFilter(CustomScanState *node); +static void PgColumnarEndRuntimeFilter(CustomScanState *node); +static void PgColumnarReScanRuntimeFilter(CustomScanState *node); +static void PgColumnarShutdownRuntimeFilter(CustomScanState *node); +static void PgColumnarExplainRuntimeFilter(CustomScanState *node, + List *ancestors, + ExplainState *es); + +static const CustomPathMethods PgColumnarRuntimeFilterPathMethods = { + .CustomName = "Columnar Runtime Filter Coordinator", + .PlanCustomPath = PgColumnarPlanRuntimeFilterPath, + .ReparameterizeCustomPathByChild = NULL, +}; + +static const CustomScanMethods PgColumnarRuntimeFilterScanMethods = { + .CustomName = "Columnar Runtime Filter Coordinator", + .CreateCustomScanState = PgColumnarCreateRuntimeFilterState, +}; + +static const CustomExecMethods PgColumnarRuntimeFilterExecMethods = { + .CustomName = "Columnar Runtime Filter Coordinator", + .BeginCustomScan = PgColumnarBeginRuntimeFilter, + .ExecCustomScan = PgColumnarExecRuntimeFilter, + .EndCustomScan = PgColumnarEndRuntimeFilter, + .ReScanCustomScan = PgColumnarReScanRuntimeFilter, + .ShutdownCustomScan = PgColumnarShutdownRuntimeFilter, + .ExplainCustomScan = PgColumnarExplainRuntimeFilter, +}; + +static const CustomScanMethods PgColumnarRuntimeTapScanMethods = { + .CustomName = "Columnar Runtime Filter Build Tap", + .CreateCustomScanState = PgColumnarCreateRuntimeTapState, +}; + +static const CustomExecMethods PgColumnarRuntimeTapExecMethods = { + .CustomName = "Columnar Runtime Filter Build Tap", + .BeginCustomScan = PgColumnarBeginRuntimeTap, + .ExecCustomScan = PgColumnarExecRuntimeTap, + .EndCustomScan = PgColumnarEndRuntimeTap, + .ReScanCustomScan = PgColumnarReScanRuntimeTap, + .ExplainCustomScan = PgColumnarExplainRuntimeTap, +}; + +static Node * +PgColumnarStripRelabel(Node *node) +{ + while (node != NULL && IsA(node, RelabelType)) + node = (Node *) ((RelabelType *) node)->arg; + + return node; +} + +static bool +PgColumnarRuntimeBaseScanPath(Path *path) +{ + CustomPath *customPath; + + if (path == NULL || !IsA(path, CustomPath)) + return false; + + customPath = (CustomPath *) path; + return customPath->methods != NULL && + strcmp(customPath->methods->CustomName, "PgColumnarScan") == 0 && + customPath->custom_private == NIL && + path->param_info == NULL && + !path->parallel_aware && + path->parallel_workers == 0; +} + +static bool +PgColumnarRuntimeFilterVars(PlannerInfo *root, + HashPath *hashPath, + Var **factVarOut, + Var **buildVarOut, + bool *factIsLeftOut, + Oid *hashCollationOut, + bool *intervalAllowedOut) +{ + RestrictInfo *restrictInfo; + OpExpr *operatorExpr; + Node *left; + Node *right; + Var *factVar; + Var *buildVar; + Relids factRelids; + Relids buildRelids; + RangeTblEntry *rte; + Relation relation; + Oid factCollation; + bool factIsLeft; + + if (list_length(hashPath->path_hashclauses) != 1) + return false; + + restrictInfo = linitial_node(RestrictInfo, hashPath->path_hashclauses); + if (!IsA(restrictInfo->clause, OpExpr)) + return false; + + operatorExpr = (OpExpr *) restrictInfo->clause; + if (list_length(operatorExpr->args) != 2) + return false; + + left = PgColumnarStripRelabel(linitial(operatorExpr->args)); + right = PgColumnarStripRelabel(lsecond(operatorExpr->args)); + if (!IsA(left, Var) || !IsA(right, Var)) + return false; + + factRelids = hashPath->jpath.outerjoinpath->parent->relids; + buildRelids = hashPath->jpath.innerjoinpath->parent->relids; + if (bms_is_member(((Var *) left)->varno, factRelids) && + bms_is_member(((Var *) right)->varno, buildRelids)) + { + factVar = (Var *) left; + buildVar = (Var *) right; + factIsLeft = true; + } + else if (bms_is_member(((Var *) right)->varno, factRelids) && + bms_is_member(((Var *) left)->varno, buildRelids)) + { + factVar = (Var *) right; + buildVar = (Var *) left; + factIsLeft = false; + } + else + return false; + + if (factVar->varlevelsup != 0 || buildVar->varlevelsup != 0 || + factVar->varattno <= 0 || buildVar->varattno <= 0) + return false; + + rte = planner_rt_fetch(factVar->varno, root); + if (rte->rtekind != RTE_RELATION || + !PgColumnarIsColumnarRelation(rte->relid)) + return false; + + relation = table_open(rte->relid, NoLock); + factCollation = + TupleDescAttr(RelationGetDescr(relation), factVar->varattno - 1)->attcollation; + table_close(relation, NoLock); + *factVarOut = factVar; + *buildVarOut = buildVar; + *factIsLeftOut = factIsLeft; + *hashCollationOut = operatorExpr->inputcollid; + *intervalAllowedOut = + (operatorExpr->inputcollid == factCollation && + factVar->vartype == buildVar->vartype); + return true; +} + +static HashPath * +PgColumnarPrivateHashPath(PlannerInfo *root, + RelOptInfo *joinrel, + HashPath *candidate, + JoinPathExtraData *extra) +{ + JoinCostWorkspace workspace; + Path *outerPath = candidate->jpath.outerjoinpath; + Path *innerPath = candidate->jpath.innerjoinpath; + Relids requiredOuter; + + requiredOuter = calc_non_nestloop_required_outer(outerPath, innerPath); + if (requiredOuter != NULL) + { + bms_free(requiredOuter); + return NULL; + } + + initial_cost_hashjoin(root, + &workspace, + candidate->jpath.jointype, + candidate->path_hashclauses, + outerPath, + innerPath, + extra, + false); + + return create_hashjoin_path(root, + joinrel, + candidate->jpath.jointype, + &workspace, + extra, + outerPath, + innerPath, + false, + extra->restrictlist, + NULL, + candidate->path_hashclauses); +} + +static void +PgColumnarSetJoinPathlist(PlannerInfo *root, + RelOptInfo *joinrel, + RelOptInfo *outerrel, + RelOptInfo *innerrel, + JoinType jointype, + JoinPathExtraData *extra) +{ + HashPath *candidate = NULL; + HashPath *privateHashPath; + CustomPath *customPath; + Var *factVar = NULL; + Var *buildVar = NULL; + bool factIsLeft = false; + bool intervalAllowed = false; + Oid hashCollation = InvalidOid; + Oid leftHashOid = InvalidOid; + Oid rightHashOid = InvalidOid; + Oid factHashOid; + Oid buildHashOid; + ListCell *cell; + Cost availableOuterWork; + Cost cappedSaving; + + if (previous_set_join_pathlist_hook != NULL) + previous_set_join_pathlist_hook(root, + joinrel, + outerrel, + innerrel, + jointype, + extra); + + if (!pgcolumnar_enable_join_runtime_filter || jointype != JOIN_INNER) + return; + + foreach(cell, joinrel->pathlist) + { + Path *path = lfirst(cell); + HashPath *hashPath; + + if (!IsA(path, HashPath)) + continue; + + hashPath = (HashPath *) path; + if (path->param_info != NULL || + path->parallel_aware || + path->parallel_workers != 0 || + hashPath->jpath.innerjoinpath->param_info != NULL || + !PgColumnarRuntimeBaseScanPath(hashPath->jpath.outerjoinpath) || + !PgColumnarRuntimeFilterVars(root, + hashPath, + &factVar, + &buildVar, + &factIsLeft, + &hashCollation, + &intervalAllowed)) + continue; + + if (!get_op_hash_functions( + ((OpExpr *) linitial_node(RestrictInfo, + hashPath->path_hashclauses)->clause)->opno, + &leftHashOid, + &rightHashOid)) + continue; + + candidate = hashPath; + break; + } + + if (candidate == NULL) + return; + + privateHashPath = PgColumnarPrivateHashPath(root, + joinrel, + candidate, + extra); + if (privateHashPath == NULL) + return; + + customPath = makeNode(CustomPath); + customPath->path.pathtype = T_CustomScan; + customPath->path.parent = joinrel; + customPath->path.pathtarget = joinrel->reltarget; + customPath->path.param_info = NULL; + customPath->path.parallel_aware = false; + customPath->path.parallel_safe = false; + customPath->path.parallel_workers = 0; + customPath->path.rows = privateHashPath->jpath.path.rows; +#if PG_VERSION_NUM >= 180000 + customPath->path.disabled_nodes = + privateHashPath->jpath.path.disabled_nodes; +#endif + customPath->path.startup_cost = + privateHashPath->jpath.path.startup_cost; + + /* + * The eventual coordinator pays a spool cost and saves only fact-side work. + * Until those measured costs are installed, cap the competing-path credit at + * five percent of the core join and never below startup cost. This makes the + * choice deterministic without changing join order or inventing unbounded + * savings. + */ + availableOuterWork = + Max(0.0, + candidate->jpath.outerjoinpath->total_cost - + candidate->jpath.outerjoinpath->startup_cost); + cappedSaving = Min(availableOuterWork, + privateHashPath->jpath.path.total_cost * 0.05); + customPath->path.total_cost = + Max(customPath->path.startup_cost, + privateHashPath->jpath.path.total_cost - cappedSaving); + customPath->path.pathkeys = NIL; + customPath->flags = 0; + customPath->custom_paths = list_make1(privateHashPath); + factHashOid = factIsLeft ? leftHashOid : rightHashOid; + buildHashOid = factIsLeft ? rightHashOid : leftHashOid; + customPath->custom_private = + list_make5(makeInteger(factVar->varattno), + makeInteger(buildVar->varno), + makeInteger(buildVar->varattno), + makeInteger(factHashOid), + makeInteger(buildHashOid)); + customPath->custom_private = + lappend(customPath->custom_private, makeInteger(hashCollation)); + customPath->custom_private = + lappend(customPath->custom_private, makeInteger(intervalAllowed ? 1 : 0)); +#if PG_VERSION_NUM >= 170000 + customPath->custom_restrictinfo = extra->restrictlist; +#endif + customPath->methods = &PgColumnarRuntimeFilterPathMethods; + + /* add_path() may free candidate. Nothing below this line reads it. */ + add_path(joinrel, &customPath->path); +} + +static AttrNumber +PgColumnarFindKeyResno(Plan *plan, Index varno, AttrNumber attno) +{ + ListCell *cell; + + foreach(cell, plan->targetlist) + { + TargetEntry *entry = lfirst_node(TargetEntry, cell); + Node *expr = PgColumnarStripRelabel((Node *) entry->expr); + + if (IsA(expr, Var) && + ((Var *) expr)->varno == varno && + ((Var *) expr)->varattno == attno) + return entry->resno; + } + + return InvalidAttrNumber; +} + +static Plan * +PgColumnarPlanRuntimeFilterPath(PlannerInfo *root, + RelOptInfo *rel, + CustomPath *bestPath, + List *tlist, + List *clauses, + List *customPlans) +{ + CustomScan *customScan; + CustomScan *tapScan; + HashJoin *joinPlan; + Hash *hashPlan; + Plan *sourcePlan; + Index buildVarno; + AttrNumber buildAttno; + AttrNumber keyResno; + + if (list_length(customPlans) != 1) + elog(ERROR, "pgcolumnar runtime filter expected one core join plan"); + + joinPlan = linitial_node(HashJoin, customPlans); + if (!IsA(joinPlan, HashJoin) || !IsA(innerPlan(joinPlan), Hash)) + elog(ERROR, "pgcolumnar runtime filter expected a core Hash Join"); + + hashPlan = (Hash *) innerPlan(joinPlan); + sourcePlan = outerPlan(hashPlan); + buildVarno = intVal(list_nth(bestPath->custom_private, 1)); + buildAttno = intVal(list_nth(bestPath->custom_private, 2)); + keyResno = PgColumnarFindKeyResno(sourcePlan, buildVarno, buildAttno); + if (!AttributeNumberIsValid(keyResno)) + elog(ERROR, "pgcolumnar runtime filter could not locate the build key"); + + tapScan = makeNode(CustomScan); + tapScan->scan.plan.targetlist = copyObject(sourcePlan->targetlist); + tapScan->scan.plan.qual = NIL; + tapScan->scan.scanrelid = 0; + tapScan->flags = 0; + tapScan->custom_plans = list_make1(sourcePlan); + tapScan->custom_private = + list_make4(makeInteger(keyResno), + copyObject(list_nth(bestPath->custom_private, 4)), + copyObject(list_nth(bestPath->custom_private, 5)), + copyObject(list_nth(bestPath->custom_private, 6))); + tapScan->custom_scan_tlist = copyObject(sourcePlan->targetlist); + tapScan->methods = &PgColumnarRuntimeTapScanMethods; + outerPlan(hashPlan) = &tapScan->scan.plan; + + customScan = makeNode(CustomScan); + customScan->scan.plan.targetlist = tlist; + customScan->scan.plan.qual = NIL; + customScan->scan.scanrelid = 0; + customScan->flags = 0; + customScan->custom_plans = list_make1(joinPlan); + customScan->custom_private = copyObject(bestPath->custom_private); + customScan->custom_scan_tlist = copyObject(joinPlan->join.plan.targetlist); + customScan->methods = &PgColumnarRuntimeFilterScanMethods; + + return &customScan->scan.plan; +} + +static Node * +PgColumnarCreateRuntimeFilterState(CustomScan *customScan) +{ + PgColumnarRuntimeFilterState *state = palloc0(sizeof(*state)); + + state->customScanState.ss.ps.type = T_CustomScanState; + state->customScanState.methods = &PgColumnarRuntimeFilterExecMethods; + return (Node *) state; +} + +static Node * +PgColumnarCreateRuntimeTapState(CustomScan *customScan) +{ + PgColumnarRuntimeTapState *state = palloc0(sizeof(*state)); + + state->customScanState.ss.ps.type = T_CustomScanState; + state->customScanState.methods = &PgColumnarRuntimeTapExecMethods; + return (Node *) state; +} + +static void +PgColumnarBeginRuntimeTap(CustomScanState *node, + EState *estate, + int eflags) +{ + PgColumnarRuntimeTapState *state = (PgColumnarRuntimeTapState *) node; + CustomScan *customScan = (CustomScan *) node->ss.ps.plan; + Plan *sourcePlan; + + if (list_length(customScan->custom_plans) != 1) + elog(ERROR, "pgcolumnar runtime filter tap expected one source plan"); + + sourcePlan = (Plan *) linitial(customScan->custom_plans); + state->sourceState = ExecInitNode(sourcePlan, estate, eflags); + node->custom_ps = list_make1(state->sourceState); + state->keyResno = intVal(linitial(customScan->custom_private)); + state->keyType = TupleDescAttr(ExecGetResultType(state->sourceState), + state->keyResno - 1)->atttypid; + state->keyCollation = intVal(list_nth(customScan->custom_private, 2)); + get_typlenbyval(state->keyType, + &state->typeLength, + &state->typeByValue); + { + TypeCacheEntry *typeCache = + lookup_type_cache(state->keyType, TYPECACHE_CMP_PROC_FINFO); + + if (intVal(list_nth(customScan->custom_private, 3)) != 0 && + OidIsValid(typeCache->cmp_proc_finfo.fn_oid)) + { + fmgr_info_copy(&state->compareFn, + &typeCache->cmp_proc_finfo, + estate->es_query_cxt); + state->intervalAvailable = true; + } + } + { + Oid buildHashOid = intVal(list_nth(customScan->custom_private, 1)); + + if (OidIsValid(buildHashOid)) + { + fmgr_info_cxt(buildHashOid, + &state->buildHashFn, + estate->es_query_cxt); + state->hashAvailable = true; + } + } + state->valueContext = + AllocSetContextCreate(estate->es_query_cxt, + "pgcolumnar runtime filter values", + ALLOCSET_SMALL_SIZES); + state->store = tuplestore_begin_heap(true, false, work_mem); + state->replaySlot = ExecInitExtraTupleSlot(estate, + ExecGetResultType(state->sourceState), + &TTSOpsMinimalTuple); +} + +static void +PgColumnarRuntimeTapAddHash(PgColumnarRuntimeTapState *state, Datum value) +{ + uint32 hash; + MemoryContext oldContext; + + if (!state->hashAvailable) + return; + + if (state->hashCount == state->hashCapacity) + { + uint32 newCapacity = state->hashCapacity == 0 + ? 256 : state->hashCapacity * 2; + + oldContext = MemoryContextSwitchTo(state->valueContext); + if (state->hashes == NULL) + state->hashes = palloc(sizeof(uint32) * newCapacity); + else + state->hashes = repalloc(state->hashes, + sizeof(uint32) * newCapacity); + MemoryContextSwitchTo(oldContext); + state->hashCapacity = newCapacity; + } + + hash = DatumGetUInt32(FunctionCall1Coll(&state->buildHashFn, + state->keyCollation, + value)); + state->hashes[state->hashCount++] = hash; +} + +static void +PgColumnarRuntimeTapAddValue(PgColumnarRuntimeTapState *state, Datum value) +{ + MemoryContext oldContext; + int32 comparison; + + if (!state->intervalAvailable) + return; + + if (!state->hasValues) + { + oldContext = MemoryContextSwitchTo(state->valueContext); + state->minimum = datumCopy(value, + state->typeByValue, + state->typeLength); + state->maximum = datumCopy(value, + state->typeByValue, + state->typeLength); + MemoryContextSwitchTo(oldContext); + state->hasValues = true; + return; + } + + comparison = DatumGetInt32(FunctionCall2Coll(&state->compareFn, + state->keyCollation, + value, + state->minimum)); + if (comparison < 0) + { + if (!state->typeByValue) + pfree(DatumGetPointer(state->minimum)); + oldContext = MemoryContextSwitchTo(state->valueContext); + state->minimum = datumCopy(value, + state->typeByValue, + state->typeLength); + MemoryContextSwitchTo(oldContext); + } + + comparison = DatumGetInt32(FunctionCall2Coll(&state->compareFn, + state->keyCollation, + value, + state->maximum)); + if (comparison > 0) + { + if (!state->typeByValue) + pfree(DatumGetPointer(state->maximum)); + oldContext = MemoryContextSwitchTo(state->valueContext); + state->maximum = datumCopy(value, + state->typeByValue, + state->typeLength); + MemoryContextSwitchTo(oldContext); + } +} + +static TupleTableSlot * +PgColumnarRuntimeTapOutput(CustomScanState *node, TupleTableSlot *sourceSlot) +{ + ExecCopySlot(node->ss.ss_ScanTupleSlot, sourceSlot); + if (node->ss.ps.ps_ProjInfo != NULL) + return ExecProject(node->ss.ps.ps_ProjInfo); + return node->ss.ss_ScanTupleSlot; +} + +static TupleTableSlot * +PgColumnarExecRuntimeTap(CustomScanState *node) +{ + PgColumnarRuntimeTapState *state = (PgColumnarRuntimeTapState *) node; + + if (!state->replay) + elog(ERROR, "pgcolumnar runtime filter tap read before drain"); + + ExecClearTuple(state->replaySlot); + if (!tuplestore_gettupleslot(state->store, + true, + false, + state->replaySlot)) + return NULL; + return PgColumnarRuntimeTapOutput(node, state->replaySlot); +} + +static void +PgColumnarDrainRuntimeTap(PgColumnarRuntimeTapState *state) +{ + TupleTableSlot *slot; + + /* + * Consume the build child into the spool without producing CustomScan + * output slots. ExecProcNode(tap) would copy each heap tuple into the + * tap's virtual slot and hand that pointer to a caller that discards it; + * Hash later replays the tuplestore. The two consumers must not share + * that discarded-slot path. + */ + for (;;) + { + slot = ExecProcNode(state->sourceState); + if (TupIsNull(slot)) + break; + tuplestore_puttupleslot(state->store, slot); + { + bool isNull; + Datum value = slot_getattr(slot, state->keyResno, &isNull); + + if (!isNull) + { + state->buildRows++; + PgColumnarRuntimeTapAddValue(state, value); + PgColumnarRuntimeTapAddHash(state, value); + } + } + CHECK_FOR_INTERRUPTS(); + } + state->replay = true; + tuplestore_rescan(state->store); +} + +static void +PgColumnarResetRuntimeTap(PgColumnarRuntimeTapState *state) +{ + tuplestore_clear(state->store); + MemoryContextReset(state->valueContext); + state->hashes = NULL; + state->hashCount = 0; + state->hashCapacity = 0; + state->hasValues = false; + state->buildRows = 0; + state->replay = false; + ExecReScan(state->sourceState); +} + +static void +PgColumnarEndRuntimeTap(CustomScanState *node) +{ + PgColumnarRuntimeTapState *state = (PgColumnarRuntimeTapState *) node; + + if (state->store != NULL) + tuplestore_end(state->store); + if (state->valueContext != NULL) + MemoryContextDelete(state->valueContext); + ExecEndNode(state->sourceState); +} + +static void +PgColumnarReScanRuntimeTap(CustomScanState *node) +{ + PgColumnarResetRuntimeTap((PgColumnarRuntimeTapState *) node); +} + +static void +PgColumnarExplainRuntimeTap(CustomScanState *node, + List *ancestors, + ExplainState *es) +{ + PgColumnarRuntimeTapState *state = (PgColumnarRuntimeTapState *) node; + + ExplainPropertyInteger("Runtime Filter Build Rows", + NULL, + (int64) state->buildRows, + es); + ExplainPropertyText("Runtime Filter Drain", + "coordinator drained the source; Hash replays the spool", + es); +} + +static void +PgColumnarBeginRuntimeFilter(CustomScanState *node, + EState *estate, + int eflags) +{ + PgColumnarRuntimeFilterState *state = + (PgColumnarRuntimeFilterState *) node; + CustomScan *customScan = (CustomScan *) node->ss.ps.plan; + Plan *childPlan; + HashJoinState *joinState; + HashState *hashState; + + if (list_length(customScan->custom_plans) != 1) + elog(ERROR, + "pgcolumnar runtime filter expected one core Hash Join plan"); + + childPlan = (Plan *) linitial(customScan->custom_plans); + state->joinState = ExecInitNode(childPlan, estate, eflags); + if (!IsA(state->joinState, HashJoinState)) + elog(ERROR, + "pgcolumnar runtime filter expected one core Hash Join state"); + + joinState = (HashJoinState *) state->joinState; + hashState = (HashState *) innerPlanState(joinState); + if (!IsA(hashState, HashState) || + !IsA(outerPlanState(hashState), CustomScanState)) + elog(ERROR, "pgcolumnar runtime filter expected a build tap"); + + state->tapState = (CustomScanState *) outerPlanState(hashState); + state->factAttno = intVal(linitial(customScan->custom_private)); + state->factHashOid = intVal(list_nth(customScan->custom_private, 3)); + state->hashCollation = intVal(list_nth(customScan->custom_private, 5)); + if (OidIsValid(state->factHashOid)) + { + fmgr_info_cxt(state->factHashOid, + &state->factHashFn, + estate->es_query_cxt); + state->hashAvailable = true; + } + state->filterContext = + AllocSetContextCreate(estate->es_query_cxt, + "pgcolumnar runtime filter", + ALLOCSET_SMALL_SIZES); + node->custom_ps = list_make1(state->joinState); +} + +static void +PgColumnarPrepareRuntimeFilter(PgColumnarRuntimeFilterState *state) +{ + PgColumnarRuntimeTapState *tapState = + (PgColumnarRuntimeTapState *) state->tapState; + + PgColumnarDrainRuntimeTap(tapState); + + state->hasValues = tapState->hasValues; + state->buildEmpty = (tapState->buildRows == 0); + if (state->hashAvailable && tapState->hashAvailable && + tapState->hashCount > 0) + { + MemoryContext oldContext = + MemoryContextSwitchTo(state->filterContext); + + state->bloomEnabled = + PgColumnarBloomBuild(tapState->hashes, + tapState->hashCount, + &state->bloom, + &state->bloomLength); + MemoryContextSwitchTo(oldContext); + } + + /* Ready before attach so the first outer probe can hash-match. */ + state->prepared = true; + PgColumnarAttachRuntimeBloom(outerPlanState(state->joinState), + state, + state->factAttno); + + if (tapState->hasValues && tapState->intervalAvailable) + state->rangeAttached = + PgColumnarAttachRuntimeRange(outerPlanState(state->joinState), + state->factAttno, + tapState->keyType, + tapState->minimum, + tapState->maximum); + +} + +static TupleTableSlot * +PgColumnarExecRuntimeFilter(CustomScanState *node) +{ + PgColumnarRuntimeFilterState *state = + (PgColumnarRuntimeFilterState *) node; + + if (!state->prepared) + PgColumnarPrepareRuntimeFilter(state); + + return ExecProcNode(state->joinState); +} + +static void +PgColumnarEndRuntimeFilter(CustomScanState *node) +{ + PgColumnarRuntimeFilterState *state = + (PgColumnarRuntimeFilterState *) node; + + ExecEndNode(state->joinState); + if (state->filterContext != NULL) + MemoryContextDelete(state->filterContext); +} + +static void +PgColumnarReScanRuntimeFilter(CustomScanState *node) +{ + PgColumnarRuntimeFilterState *state = + (PgColumnarRuntimeFilterState *) node; + + PgColumnarAttachRuntimeBloom(outerPlanState(state->joinState), + NULL, + InvalidAttrNumber); + PgColumnarDetachRuntimeRange(outerPlanState(state->joinState)); + ExecReScan(state->joinState); + PgColumnarResetRuntimeTap((PgColumnarRuntimeTapState *) state->tapState); + MemoryContextReset(state->filterContext); + state->bloom = NULL; + state->bloomLength = 0; + state->bloomEnabled = false; + state->hasValues = false; + state->buildEmpty = false; + state->rangeAttached = false; + state->prepared = false; +} + +static void +PgColumnarShutdownRuntimeFilter(CustomScanState *node) +{ + /* + * ExecShutdownNode already walks custom_ps before invoking this callback. + * Explicitly shutting down the child here would do so twice. + */ +} + +static void +PgColumnarExplainRuntimeFilter(CustomScanState *node, + List *ancestors, + ExplainState *es) +{ + PgColumnarRuntimeFilterState *state = + (PgColumnarRuntimeFilterState *) node; + + ExplainPropertyBool("Runtime Filter Ready", state->prepared, es); + ExplainPropertyBool("Runtime Filter Bloom", state->bloomEnabled, es); +} + +bool +PgColumnarRuntimeBloomMatch(void *opaque, Datum value, bool isNull) +{ + PgColumnarRuntimeFilterState *state = + (PgColumnarRuntimeFilterState *) opaque; + uint32 hash; + + if (state == NULL || !state->prepared) + return true; + if (isNull || state->buildEmpty) + return false; + if (!state->bloomEnabled) + return true; + + hash = DatumGetUInt32(FunctionCall1Coll(&state->factHashFn, + state->hashCollation, + value)); + return PgColumnarBloomProbe(state->bloom, state->bloomLength, hash); +} + +void +PgColumnarRuntimeFilterInit(void) +{ + RegisterCustomScanMethods(&PgColumnarRuntimeFilterScanMethods); + RegisterCustomScanMethods(&PgColumnarRuntimeTapScanMethods); + previous_set_join_pathlist_hook = set_join_pathlist_hook; + set_join_pathlist_hook = PgColumnarSetJoinPathlist; +} diff --git a/src/columnar_tableam.c b/src/columnar_tableam.c index b754fcc2..80af1cbd 100644 --- a/src/columnar_tableam.c +++ b/src/columnar_tableam.c @@ -3452,6 +3452,15 @@ _PG_init(void) 0, NULL, NULL, NULL); + DefineCustomBoolVariable("pgcolumnar.enable_join_runtime_filter", + "Enable serial hash-join runtime filtering for direct columnar scans.", + NULL, + &pgcolumnar_enable_join_runtime_filter, + false, + PGC_USERSET, + 0, + NULL, NULL, NULL); + DefineCustomBoolVariable("pgcolumnar.enable_projection_scan", "Let the planner scan a covering projection instead of the " "base table when one serves the query better (gap 26).", @@ -3743,6 +3752,7 @@ _PG_init(void) /* register the custom scan provider and install the pathlist hook */ PgColumnarCustomScanInit(); + PgColumnarRuntimeFilterInit(); /* install the vectorized-aggregate upper-path hook (spec 9) */ PgColumnarVectorInit(); diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv index 422ae9f0..5b80283c 100644 --- a/test/check_ledger.tsv +++ b/test/check_ledger.tsv @@ -760,3 +760,47 @@ harness_selftest 440-a-count-grep-never-produced premise: nor is one inside a ge harness_selftest 440-a-count-grep-never-produced premise: the sweep has a corpus to read never - harness_selftest 440-a-count-grep-never-produced premise: while a numeric comparison is not an offence never - harness_selftest 440-a-count-grep-never-produced premise: while a valid pattern prints a number never - +native_join_runtime_filter native_join_runtime_filter 3-table answer equals filter-off never - +native_join_runtime_filter native_join_runtime_filter 3-table join order matches filter-off never - +native_join_runtime_filter native_join_runtime_filter 3-table plan has coordinator never - +native_join_runtime_filter native_join_runtime_filter ANTI refusal never - +native_join_runtime_filter native_join_runtime_filter CROSS refusal never - +native_join_runtime_filter native_join_runtime_filter LEFT refusal never - +native_join_runtime_filter native_join_runtime_filter SEMI refusal never - +native_join_runtime_filter native_join_runtime_filter baseline core Hash Join never - +native_join_runtime_filter native_join_runtime_filter baseline reads all groups never - +native_join_runtime_filter native_join_runtime_filter build rows omit NULL never - +native_join_runtime_filter native_join_runtime_filter clustered groups removed never - +native_join_runtime_filter native_join_runtime_filter clustered reads fewer groups never - +native_join_runtime_filter native_join_runtime_filter collation-mismatch answer equals heap never - +native_join_runtime_filter native_join_runtime_filter collation-mismatch bloom rejects most non-matches never - +native_join_runtime_filter native_join_runtime_filter collation-mismatch interval stays off never - +native_join_runtime_filter native_join_runtime_filter collation-mismatch plan has coordinator never - +native_join_runtime_filter native_join_runtime_filter early limit equals heap never - +native_join_runtime_filter native_join_runtime_filter early limit has coordinator never - +native_join_runtime_filter native_join_runtime_filter empty-build answer equals heap never - +native_join_runtime_filter native_join_runtime_filter empty-build plan has coordinator never - +native_join_runtime_filter native_join_runtime_filter fact-qual conjunction equals heap never - +native_join_runtime_filter native_join_runtime_filter fact-qual plan has coordinator never - +native_join_runtime_filter native_join_runtime_filter filter ready before scan never - +native_join_runtime_filter native_join_runtime_filter int4/int8 answer equals heap never - +native_join_runtime_filter native_join_runtime_filter int4/int8 bloom rejects most non-matches never - +native_join_runtime_filter native_join_runtime_filter int4/int8 interval stays off never - +native_join_runtime_filter native_join_runtime_filter int4/int8 plan has coordinator never - +native_join_runtime_filter native_join_runtime_filter plan has build tap never - +native_join_runtime_filter native_join_runtime_filter plan has runtime coordinator never - +native_join_runtime_filter native_join_runtime_filter plan retains core Hash Join never - +native_join_runtime_filter native_join_runtime_filter projection join equals heap never - +native_join_runtime_filter native_join_runtime_filter projection outer is not wrapped never - +native_join_runtime_filter native_join_runtime_filter projection scan is chosen never - +native_join_runtime_filter native_join_runtime_filter rescan answers equal heap never - +native_join_runtime_filter native_join_runtime_filter runtime answer equals heap never - +native_join_runtime_filter native_join_runtime_filter runtime answer equals off never - +native_join_runtime_filter native_join_runtime_filter saturated answer equals heap never - +native_join_runtime_filter native_join_runtime_filter saturated bloom is disabled never - +native_join_runtime_filter native_join_runtime_filter saturated plan has coordinator never - +native_join_runtime_filter native_join_runtime_filter scattered answer equals heap never - +native_join_runtime_filter native_join_runtime_filter scattered bloom rejects most non-matches never - +native_join_runtime_filter native_join_runtime_filter scattered interval cannot drop groups never - +native_join_runtime_filter native_join_runtime_filter scattered plan has coordinator never - +native_join_runtime_filter native_join_runtime_filter scattered still reads every group never - diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt index 08560d1b..d4e6df49 100644 --- a/test/check_ledger_budget.txt +++ b/test/check_ledger_budget.txt @@ -34,4 +34,4 @@ suites_not_covered 250 # Without that it is a hand-maintained count that drifts, which is the failure # this repository has spent a day proving. It is not a ceiling; it is a # measurement that must be true. -checks_never_observed_red 762 +checks_never_observed_red 806 diff --git a/test/native_join_runtime_filter.sh b/test/native_join_runtime_filter.sh new file mode 100755 index 00000000..e3ab2503 --- /dev/null +++ b/test/native_join_runtime_filter.sh @@ -0,0 +1,294 @@ +#!/usr/bin/env bash +# Serial join runtime filter public-seam regression (#752). +set -uo pipefail +. "$(dirname "${BASH_SOURCE[0]}")/lib.sh" +pgc_setup "${1:-/usr/lib/postgresql/18/bin/pg_config}" +q "$(cat <<'SQL' +CREATE EXTENSION IF NOT EXISTS pgcolumnar; +CREATE TABLE d(k int); +INSERT INTO d SELECT g FROM generate_series(8001,8200) g; +INSERT INTO d VALUES(8100),(NULL); +CREATE TABLE f(k int,p text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$f$t$, stripe_row_limit => 1000); +INSERT INTO f SELECT g, repeat(md5(g::text), 8) FROM generate_series(1,20000) g; +CREATE TABLE h AS SELECT * FROM f; +ANALYZE d; +ANALYZE f; +SQL +)" >/dev/null +pc(){ env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -At -v ON_ERROR_STOP=1 -c "$1" 2>&1; } +SQL="SELECT count(*),sum(f.k),sum(length(f.p)) FROM f JOIN d ON f.k=d.k" +rf_on="" +rf_off="" +if [[ "$(pc "SELECT current_setting(\$g\$pgcolumnar.enable_join_runtime_filter\$g\$, true) IS NOT NULL" | tail -1)" == t ]]; then + rf_on="SET pgcolumnar.enable_join_runtime_filter=on;" + rf_off="SET pgcolumnar.enable_join_runtime_filter=off;" +fi +base="$(pc "${rf_off}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQL")" +on="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQL")" +val(){ sed -n "s/.*$1: \([0-9]*\).*/\1/p" <<<"$2" | head -1; } +check "baseline core Hash Join" "$(grep -c 'Hash Join' <<<"$base")" 1 +check "baseline reads all groups" "$(val 'Columnar Chunk Groups Read' "$base")" 20 +check "plan has runtime coordinator" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$on")" 1 +check "plan has build tap" "$(grep -c 'Columnar Runtime Filter Build Tap' <<<"$on")" 1 +check "plan retains core Hash Join" "$(grep -c 'Hash Join' <<<"$on")" 1 +check "build rows omit NULL" "$(val 'Runtime Filter Build Rows' "$on")" 201 +check "filter ready before scan" "$(grep -c 'Runtime Filter Ready: true' <<<"$on")" 1 +check "clustered groups removed" "$(val 'Runtime Filter Groups Removed' "$on")" 19 +check "clustered reads fewer groups" "$(val 'Columnar Chunk Groups Read' "$on")" 1 +check "runtime answer equals off" "$(pc "${rf_on}$SQL"|tail -1)" "$(pc "${rf_off}$SQL"|tail -1)" +check "runtime answer equals heap" "$(pc "${rf_on}$SQL"|tail -1)" "$(q "SELECT count(*),sum(h.k),sum(length(h.p)) FROM h JOIN d ON h.k=d.k")" +for shape in \ + "LEFT|SELECT count(*) FROM f LEFT JOIN d ON f.k=d.k" \ + "SEMI|SELECT count(*) FROM f WHERE EXISTS(SELECT 1 FROM d WHERE d.k=f.k)" \ + "ANTI|SELECT count(*) FROM f WHERE NOT EXISTS(SELECT 1 FROM d WHERE d.k=f.k)" \ + "CROSS|SELECT count(*) FROM f CROSS JOIN d"; do + n=${shape%%|*}; s=${shape#*|}; p="$(pc "${rf_on}EXPLAIN $s")"; check "$n refusal" "$(grep -c 'Columnar Runtime Filter Coordinator'<<<"$p")" 0 +done + +# Scattered keys: the interval hull spans every 1000-row group, so group +# pruning cannot be the thing that avoids payload work. Bloom must reject +# non-matches on the public EXPLAIN counter. 200 keys, each present once in +# 20000 fact rows, leaves 19800 non-matches; 15000 is a conservative bound +# below that after allowing Bloom false positives. +q "$(cat <<'SQL' +CREATE TABLE dim_bloom(k int); +INSERT INTO dim_bloom SELECT 25 + 100 * g FROM generate_series(0,199) g; +CREATE TABLE fact_bloom(k int, payload text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$fact_bloom$t$, stripe_row_limit => 1000); +INSERT INTO fact_bloom SELECT g, repeat(md5(g::text), 8) FROM generate_series(1,20000) g; +CREATE TABLE heap_bloom AS SELECT * FROM fact_bloom; +ANALYZE dim_bloom; +ANALYZE fact_bloom; +SQL +)" >/dev/null +SQLB="SELECT count(*),sum(fact_bloom.k),sum(length(fact_bloom.payload)) FROM fact_bloom JOIN dim_bloom ON fact_bloom.k=dim_bloom.k" +onb="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQLB")" +check "scattered plan has coordinator" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$onb")" 1 +check_num "scattered interval cannot drop groups" "$(val 'Runtime Filter Groups Removed' "$onb")" 0 +check_num "scattered still reads every group" "$(val 'Columnar Chunk Groups Read' "$onb")" 20 +rejected="$(val 'Runtime Filter Rows Rejected' "$onb")" +bloom_hit="$(awk -v r="${rejected:-0}" 'BEGIN { print (r+0 >= 15000) ? 1 : 0 }')" +check_num "scattered bloom rejects most non-matches" "$bloom_hit" 1 +check "scattered answer equals heap" "$(pc "${rf_on}$SQLB"|tail -1)" "$(q "SELECT count(*),sum(heap_bloom.k),sum(length(heap_bloom.payload)) FROM heap_bloom JOIN dim_bloom ON heap_bloom.k=dim_bloom.k")" + +# Cross-type int4 fact vs int8 dimension: hash both sides, but do not make an +# interval claim with mixed representations. 200 matching keys, 20000 fact rows. +q "$(cat <<'SQL' +CREATE TABLE dim_i8(k bigint); +INSERT INTO dim_i8 SELECT g FROM generate_series(8001,8200) g; +CREATE TABLE fact_i4(k int, payload text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$fact_i4$t$, stripe_row_limit => 1000); +INSERT INTO fact_i4 SELECT g, repeat(md5(g::text), 8) FROM generate_series(1,20000) g; +CREATE TABLE heap_i4 AS SELECT * FROM fact_i4; +ANALYZE dim_i8; +ANALYZE fact_i4; +SQL +)" >/dev/null +SQLX="SELECT count(*),sum(fact_i4.k),sum(length(fact_i4.payload)) FROM fact_i4 JOIN dim_i8 ON fact_i4.k=dim_i8.k" +onx="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQLX")" +check "int4/int8 plan has coordinator" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$onx")" 1 +xgrp="$(val 'Runtime Filter Groups Removed' "$onx")" +check_num "int4/int8 interval stays off" "${xgrp:-0}" 0 +xrej="$(val 'Runtime Filter Rows Rejected' "$onx")" +xhit="$(awk -v r="${xrej:-0}" 'BEGIN { print (r+0 >= 15000) ? 1 : 0 }')" +check_num "int4/int8 bloom rejects most non-matches" "$xhit" 1 +check "int4/int8 answer equals heap" "$(pc "${rf_on}$SQLX"|tail -1)" "$(q "SELECT count(*),sum(heap_i4.k),sum(length(heap_i4.payload)) FROM heap_i4 JOIN dim_i8 ON heap_i4.k=dim_i8.k")" + +# Collation mismatch: operator collation is not the fact attribute's. Interval +# must not claim an ordering; Bloom may still reject. 200 text keys. +q "$(cat <<'SQL' +CREATE TABLE dim_txt(k text COLLATE "C"); +INSERT INTO dim_txt SELECT (40 + 100 * g)::text FROM generate_series(0,199) g; +CREATE TABLE fact_txt(k text COLLATE "C", payload text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$fact_txt$t$, stripe_row_limit => 1000); +INSERT INTO fact_txt SELECT g::text, repeat(md5(g::text), 8) FROM generate_series(1,20000) g; +CREATE TABLE heap_txt AS SELECT * FROM fact_txt; +ANALYZE dim_txt; +ANALYZE fact_txt; +SQL +)" >/dev/null +SQLC=$(cat <<'SQL' +SELECT count(*),sum(length(fact_txt.k)),sum(length(fact_txt.payload)) +FROM fact_txt JOIN dim_txt ON fact_txt.k = dim_txt.k COLLATE "POSIX" +SQL +) +onc="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQLC")" +check "collation-mismatch plan has coordinator" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$onc")" 1 +cgrp="$(val 'Runtime Filter Groups Removed' "$onc")" +check_num "collation-mismatch interval stays off" "${cgrp:-0}" 0 +crej="$(val 'Runtime Filter Rows Rejected' "$onc")" +chit="$(awk -v r="${crej:-0}" 'BEGIN { print (r+0 >= 15000) ? 1 : 0 }')" +check_num "collation-mismatch bloom rejects most non-matches" "$chit" 1 +check "collation-mismatch answer equals heap" "$(pc "${rf_on}$SQLC"|tail -1)" "$(q "$(cat <<'SQL' +SELECT count(*),sum(length(heap_txt.k)),sum(length(heap_txt.payload)) +FROM heap_txt JOIN dim_txt ON heap_txt.k = dim_txt.k COLLATE "POSIX" +SQL +)")" + +# Correlated LATERAL must rebuild the filter per outer parameter. 0 keeps all +# 200 keys; 8000 keeps g>=80, which is 120 keys in this fixture. +SQLR=$(cat <<'SQL' +SELECT v.x, s.c +FROM (VALUES (0),(8000)) v(x) +CROSS JOIN LATERAL ( + SELECT count(*) c + FROM fact_bloom JOIN dim_bloom ON fact_bloom.k = dim_bloom.k + WHERE dim_bloom.k > v.x +) s +ORDER BY 1 +SQL +) +SQLRH=$(cat <<'SQL' +SELECT v.x, s.c +FROM (VALUES (0),(8000)) v(x) +CROSS JOIN LATERAL ( + SELECT count(*) c + FROM heap_bloom JOIN dim_bloom ON heap_bloom.k = dim_bloom.k + WHERE dim_bloom.k > v.x +) s +ORDER BY 1 +SQL +) +check "rescan answers equal heap" "$(pc "${rf_on}SET max_parallel_workers_per_gather=0;$SQLR" | grep -v '^SET$')" "$(q "$SQLRH")" + +# 220000 distinct build keys exceed d*10 vs 2^21, so Bloom must refuse rather +# than emit a saturated filter. Fact is larger so it remains the hash outer. +q "$(cat <<'SQL' +CREATE TABLE dim_sat(k int); +INSERT INTO dim_sat SELECT g FROM generate_series(1,220000) g; +CREATE TABLE fact_sat(k int) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$fact_sat$t$, stripe_row_limit => 1000); +INSERT INTO fact_sat SELECT g FROM generate_series(1,300000) g; +CREATE TABLE heap_sat AS SELECT * FROM fact_sat; +ANALYZE dim_sat; +ANALYZE fact_sat; +SQL +)" >/dev/null +SQLS="SELECT count(*),sum(fact_sat.k) FROM fact_sat JOIN dim_sat ON fact_sat.k=dim_sat.k" +ons="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQLS")" +check "saturated plan has coordinator" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$ons")" 1 +check "saturated bloom is disabled" "$(grep -c 'Runtime Filter Bloom: false' <<<"$ons")" 1 +check "saturated answer equals heap" "$(pc "${rf_on}$SQLS"|tail -1)" "$(q "SELECT count(*),sum(heap_sat.k) FROM heap_sat JOIN dim_sat ON heap_sat.k=dim_sat.k")" + +# Three-table join: wrapping the columnar-outer hash join must not reorder the +# two heap dimensions relative to filter-off. +q "$(cat <<'SQL' +CREATE TABLE ja(k int); +INSERT INTO ja SELECT g FROM generate_series(1,200) g; +CREATE TABLE jb(k int, extra int); +INSERT INTO jb SELECT g, g+1 FROM generate_series(1,200) g; +CREATE TABLE jf(k int, payload text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$jf$t$, stripe_row_limit => 1000); +INSERT INTO jf SELECT g, repeat(md5(g::text), 4) FROM generate_series(1,5000) g; +ANALYZE ja; +ANALYZE jb; +ANALYZE jf; +SQL +)" >/dev/null +SQL3="SELECT count(*),sum(jf.k),sum(jb.extra) FROM jf JOIN ja ON jf.k=ja.k JOIN jb ON ja.k=jb.k" +rel_order() { + python3 -c ' +import json,sys +plan=json.loads(sys.stdin.read()) +def walk(n): + if isinstance(n, list): + for x in n: + yield from walk(x) + return + if not isinstance(n, dict): + return + name=n.get("Relation Name") + if name: + yield name + for child in n.get("Plans") or []: + yield from walk(child) + child=n.get("Plan") + if isinstance(child, dict): + yield from walk(child) +print(" ".join(walk(plan))) +' +} +on3="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(FORMAT JSON, COSTS OFF)$SQL3" | grep -v '^SET$')" +off3="$(pc "${rf_off}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(FORMAT JSON, COSTS OFF)$SQL3" | grep -v '^SET$')" +check "3-table plan has coordinator" "$(printf '%s\n' "$on3" | grep -c 'Columnar Runtime Filter Coordinator')" 1 +check "3-table join order matches filter-off" "$(printf '%s\n' "$on3" | rel_order)" "$(printf '%s\n' "$off3" | rel_order)" +check "3-table answer equals filter-off" "$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;$SQL3"|tail -1)" "$(pc "${rf_off}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;$SQL3"|tail -1)" + +# Empty build: Hash Join still exists, the coordinator still wraps it, and the +# answer is zero rather than a leftover hull from a previous execution. +q "$(cat <<'SQL' +CREATE TABLE dim_none(k int); +CREATE TABLE fact_none(k int, payload text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$fact_none$t$, stripe_row_limit => 1000); +INSERT INTO fact_none SELECT g, repeat(md5(g::text), 4) FROM generate_series(1,5000) g; +CREATE TABLE heap_none AS SELECT * FROM fact_none; +ANALYZE dim_none; +ANALYZE fact_none; +SQL +)" >/dev/null +SQLN="SELECT count(*),sum(fact_none.k) FROM fact_none JOIN dim_none ON fact_none.k=dim_none.k" +onn="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQLN")" +check "empty-build plan has coordinator" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$onn")" 1 +check "empty-build answer equals heap" "$(pc "${rf_on}$SQLN"|tail -1)" "$(q "SELECT count(*),sum(heap_none.k) FROM heap_none JOIN dim_none ON heap_none.k=dim_none.k")" + +# Projection-backed outer scans are excluded: a covering projection with a +# sort-key restriction is cheaper than the base scan, so Hash Join would wrap +# it if the coordinator did not refuse custom_private != NIL. +q "$(cat <<'SQL' +CREATE TABLE dim_pj(k int); +INSERT INTO dim_pj SELECT g FROM generate_series(1,200) g; +CREATE TABLE fact_pj(k int, payload text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$fact_pj$t$, stripe_row_limit => 1000); +INSERT INTO fact_pj SELECT g, repeat(md5(g::text), 4) FROM generate_series(1,4000) g; +SELECT pgcolumnar.add_projection($t$fact_pj$t$, $n$byk$n$, ARRAY['k','payload'], ARRAY['k']); +CREATE TABLE heap_pj AS SELECT * FROM fact_pj; +ANALYZE dim_pj; +ANALYZE fact_pj; +SQL +)" >/dev/null +SQLP="SELECT count(*),sum(fact_pj.k) FROM fact_pj JOIN dim_pj ON fact_pj.k=dim_pj.k WHERE fact_pj.k BETWEEN 1 AND 200" +onp="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQLP")" +check "projection scan is chosen" "$(grep -c 'Columnar Projection:' <<<"$onp")" 1 +check "projection outer is not wrapped" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$onp")" 0 +check "projection join equals heap" "$(pc "${rf_on}$SQLP"|tail -1)" "$(q "SELECT count(*),sum(h.k) FROM heap_pj h JOIN dim_pj ON h.k=dim_pj.k WHERE h.k BETWEEN 1 AND 200")" + +# Early LIMIT on a compact key range. The coordinator used to SIGSEGV while +# draining the tap through ExecProcNode on this shape; a heap control with the +# same ORDER BY is the public answer. +q "$(cat <<'SQL' +CREATE TABLE dim_lim(k int); +INSERT INTO dim_lim SELECT g FROM generate_series(300,420) g; +CREATE TABLE fact_lim(k int, payload text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$fact_lim$t$, stripe_row_limit => 1000); +INSERT INTO fact_lim SELECT g, repeat(md5(g::text), 4) FROM generate_series(1,2500) g; +CREATE TABLE heap_lim AS SELECT * FROM fact_lim; +ANALYZE dim_lim; +ANALYZE fact_lim; +SQL +)" >/dev/null +SQLL="SELECT fact_lim.k FROM fact_lim JOIN dim_lim ON fact_lim.k=dim_lim.k ORDER BY fact_lim.k LIMIT 3" +onl="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQLL")" +check "early limit has coordinator" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$onl")" 1 +check "early limit equals heap" "$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;$SQLL" | grep -v '^SET$')" "$(q "SELECT heap_lim.k FROM heap_lim JOIN dim_lim ON heap_lim.k=dim_lim.k ORDER BY heap_lim.k LIMIT 3")" +# Fact-table local qual must see every column it names. The coordinator +# used to mark only the join key, so a conjunction that also named a +# non-key column dropped every surviving row. Independent of the pytest +# twin: different names, key range, and row count. +q "$(cat <<'SQL' +CREATE TABLE dim_fq(k int); +INSERT INTO dim_fq SELECT g FROM generate_series(5500, 5699) g; +CREATE TABLE fact_fq(k int, extra int, payload text) USING pgcolumnar; +SELECT pgcolumnar.set_options($t$fact_fq$t$, stripe_row_limit => 1000); +INSERT INTO fact_fq SELECT g, g, repeat(md5(g::text), 4) +FROM generate_series(1, 12000) g; +CREATE TABLE heap_fq AS SELECT * FROM fact_fq; +ANALYZE dim_fq; +ANALYZE fact_fq; +SQL +)" >/dev/null +SQLFQ="SELECT count(*),sum(fact_fq.k),sum(fact_fq.extra) FROM fact_fq JOIN dim_fq ON fact_fq.k=dim_fq.k WHERE fact_fq.k > 100 AND fact_fq.extra > 40" +onfq="$(pc "${rf_on}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQLFQ")" +check "fact-qual plan has coordinator" "$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$onfq")" 1 +check "fact-qual conjunction equals heap" "$(pc "${rf_on}$SQLFQ"|tail -1)" "$(q "SELECT count(*),sum(heap_fq.k),sum(heap_fq.extra) FROM heap_fq JOIN dim_fq ON heap_fq.k=dim_fq.k WHERE heap_fq.k > 100 AND heap_fq.extra > 40")" +pgc_summary diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index d6b10b8f..069bcbda 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -70,6 +70,7 @@ behaviour, the source of that number is named. - [22. test_writes_wrote_rows.py: a write that wrote nothing](#22-test_writes_wrote_rowspy-a-write-that-wrote-nothing) - [23. test_mutation_ledger.py: which checks have ever been red](#23-test_mutation_ledgerpy-which-checks-have-ever-been-red) - [24. test_loop_coverage_premise.py: a loop that never ran asserted nothing](#24-test_loop_coverage_premisepy-a-loop-that-never-ran-asserted-nothing) +- [25. test_join_runtime_filter.py: serial join runtime filter](#25-test_join_runtime_filterpy-serial-join-runtime-filter) ## 1. How to read a test in here @@ -2149,8 +2150,6 @@ references in `.github/`, zero in the runner. If they disagree, one was edited by hand. `suites_not_covered` is 250 of 251, so the gate cannot refuse a new check in 250 suites — a real limit, counted rather than hidden, which falls as suites are seeded. - - ## 24. test_loop_coverage_premise.py: a loop that never ran asserted nothing **Why this file exists.** `assert-inside-a-loop-over-zero-rows` in VACUITY_MODES.md 3.5 @@ -2202,3 +2201,61 @@ The two differ, and the reason is dataflow. `test_harness_deps.py`'s loop iterat a preceding loop. A rule that demanded the names match would reject correct code, which is how a guard gets switched off. **So the residual is a loop whose premise bounds the wrong collection**, which a reviewer catches and a sweep does not. 3.5 names it. + + +## 25. test_join_runtime_filter.py: serial join runtime filter + +Pytest twin of `test/native_join_runtime_filter.sh`. The two files are independent: +each builds its own fixtures and expected values. They share only the public +EXPLAIN names and the SQL answers. + +### `test_serial_join_runtime_filter` + +Clustered integer keys. The coordinator wraps core Hash Join, the build tap +omits NULL, and nineteen of twenty groups are removed. LEFT, SEMI, ANTI, and +CROSS plans are refused. Answers match both filter-off and a heap twin. + +### `test_scattered_join_runtime_bloom` + +Scattered keys keep every group. Bloom must reject most non-matches on +`Runtime Filter Rows Rejected`. The hull cannot be the thing that avoids work. + +### `test_cross_type_int4_int8_bloom` + +int4 fact vs int8 dimension. Both sides hash. Interval stays off. + +### `test_collation_mismatch_bloom_only` + +Operator collation is not the fact attribute collation. Interval stays off. +Bloom may still reject. + +### `test_runtime_filter_rebuilds_on_lateral_rescan` + +A correlated LATERAL rebuilds the filter for each outer parameter. + +### `test_saturated_build_disables_bloom` + +A build side past the on-disk bloom cap disables Bloom rather than saturating. + +### `test_three_table_join_order_unchanged` + +Wrapping the columnar-outer hash join must not reorder the other inputs. + +### `test_empty_build_runtime_filter` + +An empty dimension still uses the coordinator and returns no join rows. + +### `test_projection_outer_is_not_wrapped` + +A covering projection scan stays an unwrapped Hash Join outer. + +### `test_early_limit_matches_heap` + +`ORDER BY ... LIMIT` still matches a heap control. The coordinator used to +crash on this shape when it drained the tap through `ExecProcNode`. + +### `test_fact_qual_with_late_mat_off_matches_heap` + +A non-key fact-table qual with late materialization off. The attach used to +force the two-pass path with only the join key decoded, so the qual dropped +every row. Heap is the oracle. Independent of the shell conjunction arm. diff --git a/test/pytest/test_join_runtime_filter.py b/test/pytest/test_join_runtime_filter.py new file mode 100644 index 00000000..27263884 --- /dev/null +++ b/test/pytest/test_join_runtime_filter.py @@ -0,0 +1,672 @@ +"""Pytest twin of native_join_runtime_filter.sh.""" +import re + + +def _has_runtime_filter(c): + with c.cursor() as x: + x.execute( + "SELECT current_setting($g$pgcolumnar.enable_join_runtime_filter$g$, true)" + ) + return x.fetchone()[0] is not None + + +def _plan(c, sql, on=None): + with c.cursor() as x: + if on is not None and _has_runtime_filter(c): + x.execute( + "SET pgcolumnar.enable_join_runtime_filter=" + + ("on" if on else "off") + ) + x.execute("SET max_parallel_workers_per_gather=0") + x.execute("SET enable_nestloop=off") + x.execute("SET enable_mergejoin=off") + x.execute("EXPLAIN(ANALYZE,TIMING off,SUMMARY off)" + sql) + return "\n".join(r[0] for r in x.fetchall()) + + +def _v(p, n): + m = re.search(re.escape(n) + r": ([0-9]+)", p) + return int(m.group(1)) if m else -1 + + +def test_serial_join_runtime_filter(pgc_conn, expect): + with pgc_conn.cursor() as c: + c.execute( + """ + CREATE TABLE d(k int); + INSERT INTO d SELECT g FROM generate_series(8001,8200) g; + INSERT INTO d VALUES(8100),(NULL); + CREATE TABLE f(k int,p text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$f$t$, stripe_row_limit => 1000); + INSERT INTO f SELECT g, repeat(md5(g::text), 8) + FROM generate_series(1,20000) g; + CREATE TABLE h AS SELECT * FROM f; + ANALYZE d; + ANALYZE f + """ + ) + sql = "SELECT count(*),sum(f.k),sum(length(f.p))FROM f JOIN d ON f.k=d.k" + b = _plan(pgc_conn, sql, False) + p = _plan(pgc_conn, sql, True) + expect.num(b.count("Hash Join"), 1, "baseline core Hash Join") + expect.num(_v(b, "Columnar Chunk Groups Read"), 20, "baseline reads all groups") + expect.num(p.count("Columnar Runtime Filter Coordinator"), 1, "plan has runtime coordinator") + expect.num(p.count("Columnar Runtime Filter Build Tap"), 1, "plan has build tap") + expect.num(p.count("Hash Join"), 1, "plan retains core Hash Join") + expect.num(_v(p, "Runtime Filter Build Rows"), 201, "build rows omit NULL") + expect.num(p.count("Runtime Filter Ready: true"), 1, "filter ready before scan") + expect.num(_v(p, "Runtime Filter Groups Removed"), 19, "clustered groups removed") + expect.num(_v(p, "Columnar Chunk Groups Read"), 1, "clustered reads fewer groups") + with pgc_conn.cursor() as c: + if _has_runtime_filter(pgc_conn): + c.execute("SET pgcolumnar.enable_join_runtime_filter=on") + c.execute(sql) + a = c.fetchone() + if _has_runtime_filter(pgc_conn): + c.execute("SET pgcolumnar.enable_join_runtime_filter=off") + c.execute(sql) + o = c.fetchone() + c.execute( + """ + SELECT count(*),sum(h.k),sum(length(h.p)) + FROM h JOIN d ON h.k=d.k + """ + ) + h = c.fetchone() + expect.rows([a], [o], "runtime answer equals off") + expect.rows([a], [h], "runtime answer equals heap") + for name, s in [ + ("LEFT refusal", "SELECT count(*)FROM f LEFT JOIN d ON f.k=d.k"), + ("SEMI refusal", "SELECT count(*)FROM f WHERE EXISTS(SELECT 1 FROM d WHERE d.k=f.k)"), + ("ANTI refusal", "SELECT count(*)FROM f WHERE NOT EXISTS(SELECT 1 FROM d WHERE d.k=f.k)"), + ("CROSS refusal", "SELECT count(*)FROM f CROSS JOIN d"), + ]: + expect.num( + _plan(pgc_conn, s, True).count("Columnar Runtime Filter Coordinator"), + 0, + name, + ) + + +def _json_nodes(plan): + if isinstance(plan, list): + for item in plan: + yield from _json_nodes(item) + return + if not isinstance(plan, dict): + return + yield plan + child = plan.get("Plan") + if isinstance(child, dict): + yield from _json_nodes(child) + for child in plan.get("Plans") or []: + yield from _json_nodes(child) + + +def _json_field(plan, key): + for node in _json_nodes(plan): + if key in node: + return node[key] + return None + + +def test_scattered_join_runtime_bloom(pgc_conn, expect): + """Scattered keys keep every group; Bloom must reject non-matching rows.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE dimb(k int); + INSERT INTO dimb SELECT 80 + 100 * g FROM generate_series(0, 199) g; + CREATE TABLE factb(k int, payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$factb$t$, stripe_row_limit => 1000); + INSERT INTO factb SELECT g, repeat(md5(g::text), 8) + FROM generate_series(1, 20000) g; + CREATE TABLE heapb AS SELECT * FROM factb; + ANALYZE dimb; + ANALYZE factb + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + sql = """ + SELECT count(*), sum(factb.k), sum(length(factb.payload)) + FROM factb JOIN dimb ON factb.k = dimb.k + """ + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + cur.execute(sql) + got = cur.fetchone() + cur.execute( + """ + SELECT count(*), sum(heapb.k), sum(length(heapb.payload)) + FROM heapb JOIN dimb ON heapb.k = dimb.k + """ + ) + heap = cur.fetchone() + + expect.plan_node( + plan, + provider="Columnar Runtime Filter Coordinator", + name="scattered plan uses the runtime coordinator", + ) + expect.num( + _json_field(plan, "Runtime Filter Groups Removed"), + 0, + "scattered hull cannot drop a group", + ) + expect.num( + _json_field(plan, "Columnar Chunk Groups Read"), + 20, + "scattered still reads every group", + ) + expect.plan_marker( + plan, + "Runtime Filter Rows Rejected", + name="scattered plan reports dedicated bloom rejects", + ) + expect.at_least( + _json_field(plan, "Runtime Filter Rows Rejected"), + 15000, + "scattered bloom rejects most non-matches", + ) + expect.rows([got], [heap], "scattered answer equals heap") + + +def _removed_or_zero(plan): + removed = _json_field(plan, "Runtime Filter Groups Removed") + return 0 if removed is None else removed + + +def test_cross_type_int4_int8_bloom(pgc_conn, expect): + """int4 fact vs int8 dim: hash both sides, do not claim an interval.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE dim8(k bigint); + INSERT INTO dim8 SELECT g FROM generate_series(7900, 8099) g; + CREATE TABLE fact4(k int, payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$fact4$t$, stripe_row_limit => 1000); + INSERT INTO fact4 SELECT g, repeat(md5(g::text), 8) + FROM generate_series(1, 20000) g; + CREATE TABLE heap4 AS SELECT * FROM fact4; + ANALYZE dim8; + ANALYZE fact4 + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + sql = """ + SELECT count(*), sum(fact4.k), sum(length(fact4.payload)) + FROM fact4 JOIN dim8 ON fact4.k = dim8.k + """ + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + cur.execute(sql) + got = cur.fetchone() + cur.execute( + """ + SELECT count(*), sum(heap4.k), sum(length(heap4.payload)) + FROM heap4 JOIN dim8 ON heap4.k = dim8.k + """ + ) + heap = cur.fetchone() + + expect.plan_node( + plan, + provider="Columnar Runtime Filter Coordinator", + name="int4/int8 plan uses the runtime coordinator", + ) + expect.num( + _removed_or_zero(plan), + 0, + "int4/int8 interval stays off", + ) + expect.at_least( + _json_field(plan, "Runtime Filter Rows Rejected") or 0, + 15000, + "int4/int8 bloom rejects most non-matches", + ) + expect.rows([got], [heap], "int4/int8 answer equals heap") + + +def test_collation_mismatch_bloom_only(pgc_conn, expect): + """Mismatched operator collation must not order; Bloom may still reject.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE dimc(k text COLLATE "C"); + INSERT INTO dimc SELECT (60 + 100 * g)::text FROM generate_series(0, 199) g; + CREATE TABLE factc(k text COLLATE "C", payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$factc$t$, stripe_row_limit => 1000); + INSERT INTO factc SELECT g::text, repeat(md5(g::text), 8) + FROM generate_series(1, 20000) g; + CREATE TABLE heapc AS SELECT * FROM factc; + ANALYZE dimc; + ANALYZE factc + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + sql = """ + SELECT count(*), sum(length(factc.k)), sum(length(factc.payload)) + FROM factc JOIN dimc ON factc.k = dimc.k COLLATE "POSIX" + """ + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + cur.execute(sql) + got = cur.fetchone() + cur.execute( + """ + SELECT count(*), sum(length(heapc.k)), sum(length(heapc.payload)) + FROM heapc JOIN dimc ON heapc.k = dimc.k COLLATE "POSIX" + """ + ) + heap = cur.fetchone() + + expect.plan_node( + plan, + provider="Columnar Runtime Filter Coordinator", + name="collation-mismatch plan uses the runtime coordinator", + ) + expect.num( + _removed_or_zero(plan), + 0, + "collation-mismatch interval stays off", + ) + expect.at_least( + _json_field(plan, "Runtime Filter Rows Rejected") or 0, + 15000, + "collation-mismatch bloom rejects most non-matches", + ) + expect.rows([got], [heap], "collation-mismatch answer equals heap") + + +def test_runtime_filter_rebuilds_on_lateral_rescan(pgc_conn, expect): + """A correlated LATERAL must rebuild the filter for each outer parameter.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE dimr(k int); + INSERT INTO dimr SELECT 90 + 100 * g FROM generate_series(0, 199) g; + CREATE TABLE factr(k int, payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$factr$t$, stripe_row_limit => 1000); + INSERT INTO factr SELECT g, repeat(md5(g::text), 8) + FROM generate_series(1, 20000) g; + CREATE TABLE heapr AS SELECT * FROM factr; + ANALYZE dimr; + ANALYZE factr + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET max_parallel_workers_per_gather=0") + sql = """ + SELECT v.x, s.c + FROM (VALUES (10),(7000)) v(x) + CROSS JOIN LATERAL ( + SELECT count(*) c + FROM factr JOIN dimr ON factr.k = dimr.k + WHERE dimr.k > v.x + ) s + ORDER BY 1 + """ + cur.execute(sql) + got = cur.fetchall() + cur.execute( + """ + SELECT v.x, s.c + FROM (VALUES (10),(7000)) v(x) + CROSS JOIN LATERAL ( + SELECT count(*) c + FROM heapr JOIN dimr ON heapr.k = dimr.k + WHERE dimr.k > v.x + ) s + ORDER BY 1 + """ + ) + heap = cur.fetchall() + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + + expect.plan_node( + plan, + provider="Columnar Runtime Filter Coordinator", + name="lateral rescan still uses the runtime coordinator", + ) + expect.rows(got, heap, "lateral rescan answers equal heap") + + +def test_saturated_build_disables_bloom(pgc_conn, expect): + """A build side past the #467 cap must disable Bloom rather than saturate.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE dims(k int); + INSERT INTO dims SELECT g FROM generate_series(1, 225000) g; + CREATE TABLE facts(k int) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$facts$t$, stripe_row_limit => 1000); + INSERT INTO facts SELECT g FROM generate_series(1, 310000) g; + CREATE TABLE heaps AS SELECT * FROM facts; + ANALYZE dims; + ANALYZE facts + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + sql = """ + SELECT count(*), sum(facts.k) + FROM facts JOIN dims ON facts.k = dims.k + """ + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + cur.execute(sql) + got = cur.fetchone() + cur.execute( + """ + SELECT count(*), sum(heaps.k) + FROM heaps JOIN dims ON heaps.k = dims.k + """ + ) + heap = cur.fetchone() + + expect.plan_node( + plan, + provider="Columnar Runtime Filter Coordinator", + name="saturated plan uses the runtime coordinator", + ) + bloom = _json_field(plan, "Runtime Filter Bloom") + expect.num( + 1 if bloom is False else 0, + 1, + "saturated bloom is disabled", + ) + expect.rows([got], [heap], "saturated answer equals heap") + + +def _relation_names(plan): + names = [] + for node in _json_nodes(plan): + name = node.get("Relation Name") + if name: + names.append(name) + return names + + +def test_three_table_join_order_unchanged(pgc_conn, expect): + """Wrapping the columnar-outer hash join must not reorder the other inputs.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE a3(k int); + INSERT INTO a3 SELECT g FROM generate_series(1, 150) g; + CREATE TABLE b3(k int, extra int); + INSERT INTO b3 SELECT g, g + 3 FROM generate_series(1, 150) g; + CREATE TABLE f3(k int, payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$f3$t$, stripe_row_limit => 1000); + INSERT INTO f3 SELECT g, repeat(md5(g::text), 4) + FROM generate_series(1, 4000) g; + ANALYZE a3; + ANALYZE b3; + ANALYZE f3 + """ + ) + sql = """ + SELECT count(*), sum(f3.k), sum(b3.extra) + FROM f3 JOIN a3 ON f3.k = a3.k JOIN b3 ON a3.k = b3.k + """ + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("EXPLAIN (FORMAT JSON, COSTS OFF) " + sql) + on_plan = cur.fetchone()[0] + cur.execute(sql) + on_rows = cur.fetchone() + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=off") + cur.execute("EXPLAIN (FORMAT JSON, COSTS OFF) " + sql) + off_plan = cur.fetchone()[0] + cur.execute(sql) + off_rows = cur.fetchone() + + expect.plan_node( + on_plan, + provider="Columnar Runtime Filter Coordinator", + name="3-table plan uses the runtime coordinator", + ) + expect.rows( + _relation_names(on_plan), + _relation_names(off_plan), + "3-table relation order matches filter-off", + ) + expect.rows([on_rows], [off_rows], "3-table answer matches filter-off") + + +def test_empty_build_runtime_filter(pgc_conn, expect): + """An empty dimension still uses the coordinator and returns no join rows.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE dim0(k int); + CREATE TABLE factz(k int, payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$factz$t$, stripe_row_limit => 1000); + INSERT INTO factz SELECT g, repeat(md5(g::text), 4) + FROM generate_series(1, 8000) g; + CREATE TABLE heap0 AS SELECT * FROM factz; + ANALYZE dim0; + ANALYZE factz + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + sql = """ + SELECT count(*), sum(factz.k) + FROM factz JOIN dim0 ON factz.k = dim0.k + """ + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + cur.execute(sql) + got = cur.fetchone() + cur.execute( + """ + SELECT count(*), sum(heap0.k) + FROM heap0 JOIN dim0 ON heap0.k = dim0.k + """ + ) + heap = cur.fetchone() + + expect.plan_node( + plan, + provider="Columnar Runtime Filter Coordinator", + name="empty-build plan uses the runtime coordinator", + ) + expect.rows([got], [heap], "empty-build answer equals heap") + + +def test_projection_outer_is_not_wrapped(pgc_conn, expect): + """A covering projection scan must stay an unwrapped Hash Join outer.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE dim_pr(k int); + INSERT INTO dim_pr SELECT g FROM generate_series(10, 180) g; + CREATE TABLE factp(k int, payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$factp$t$, stripe_row_limit => 1000); + INSERT INTO factp SELECT g, repeat(md5(g::text), 4) + FROM generate_series(1, 6000) g; + SELECT pgcolumnar.add_projection( + $t$factp$t$, $n$pk$n$, ARRAY['k','payload'], ARRAY['k']); + CREATE TABLE heapp AS SELECT * FROM factp; + ANALYZE dim_pr; + ANALYZE factp + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + sql = """ + SELECT count(*), sum(factp.k) + FROM factp JOIN dim_pr ON factp.k = dim_pr.k + WHERE factp.k BETWEEN 10 AND 180 + """ + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + cur.execute(sql) + got = cur.fetchone() + cur.execute( + """ + SELECT count(*), sum(heapp.k) + FROM heapp JOIN dim_pr ON heapp.k = dim_pr.k + WHERE heapp.k BETWEEN 10 AND 180 + """ + ) + heap = cur.fetchone() + + expect.plan_marker( + plan, + "Columnar Projection", + name="projection scan is chosen", + ) + expect.num( + sum( + 1 + for n in _json_nodes(plan) + if n.get("Custom Plan Provider") + == "Columnar Runtime Filter Coordinator" + ), + 0, + "projection outer is not wrapped", + ) + expect.rows([got], [heap], "projection join equals heap") + + +def test_early_limit_matches_heap(pgc_conn, expect): + """LIMIT must shut the coordinator down and still match a heap control.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE diml(k int); + INSERT INTO diml SELECT g FROM generate_series(400, 500) g; + CREATE TABLE factl(k int, payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$factl$t$, stripe_row_limit => 1000); + INSERT INTO factl SELECT g, repeat(md5(g::text), 4) + FROM generate_series(1, 3000) g; + CREATE TABLE heapl AS SELECT * FROM factl; + ANALYZE diml; + ANALYZE factl + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + sql = """ + SELECT factl.k + FROM factl JOIN diml ON factl.k = diml.k + ORDER BY factl.k + LIMIT 7 + """ + cur.execute(sql) + got = cur.fetchall() + cur.execute( + """ + SELECT heapl.k + FROM heapl JOIN diml ON heapl.k = diml.k + ORDER BY heapl.k + LIMIT 7 + """ + ) + heap = cur.fetchall() + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + + expect.plan_node( + plan, + provider="Columnar Runtime Filter Coordinator", + name="early limit still uses the runtime coordinator", + ) + expect.rows(got, heap, "early limit equals heap") + +def test_fact_qual_with_late_mat_off_matches_heap(pgc_conn, expect): + """A non-key fact qual with late materialization off must still match heap.""" + with pgc_conn.cursor() as cur: + cur.execute( + """ + CREATE TABLE dimq(k int); + INSERT INTO dimq SELECT g FROM generate_series(9100, 9249) g; + CREATE TABLE factq(k int, n int, payload text) USING pgcolumnar; + SELECT pgcolumnar.set_options($t$factq$t$, stripe_row_limit => 1000); + INSERT INTO factq SELECT g, g, repeat(md5(g::text), 3) + FROM generate_series(1, 16000) g; + CREATE TABLE heapq AS SELECT * FROM factq; + ANALYZE dimq; + ANALYZE factq + """ + ) + if _has_runtime_filter(pgc_conn): + cur.execute("SET pgcolumnar.enable_join_runtime_filter=on") + cur.execute("SET pgcolumnar.enable_late_materialization=off") + cur.execute("SET max_parallel_workers_per_gather=0") + cur.execute("SET enable_nestloop=off") + cur.execute("SET enable_mergejoin=off") + sql = """ + SELECT count(*), sum(factq.k), sum(factq.n) + FROM factq JOIN dimq ON factq.k = dimq.k + WHERE factq.n > 80 + """ + cur.execute( + "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql + ) + plan = cur.fetchone()[0] + cur.execute(sql) + got = cur.fetchone() + cur.execute( + """ + SELECT count(*), sum(heapq.k), sum(heapq.n) + FROM heapq JOIN dimq ON heapq.k = dimq.k + WHERE heapq.n > 80 + """ + ) + heap = cur.fetchone() + + expect.plan_node( + plan, + provider="Columnar Runtime Filter Coordinator", + name="late-mat-off fact qual still uses the runtime coordinator", + ) + expect.rows([got], [heap], "late-mat-off fact qual equals heap") diff --git a/test/run_all_versions.sh b/test/run_all_versions.sh index 8edb93e2..5d02cef9 100755 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -163,6 +163,7 @@ SUITES=( native_index_fetch_stripe_cost native_index_projection native_ios + native_join_runtime_filter native_late_materialization native_lazy_slot native_metadata_flush