From 89512d0717a20325504638c06202875534c59a5d Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Thu, 10 Sep 2026 17:01:17 +0000 Subject: [PATCH 1/6] test: prove serial join runtime filter is absent (#752) RED evidence on current main f0f1f40: native_join_runtime_filter.sh fails six feature assertions with no crash; pytest twin fails at the coordinator assertion. No implementation is present in this commit. --- test/native_join_runtime_filter.sh | 39 ++++++++++++++++++++++ test/pytest/test_join_runtime_filter.py | 44 +++++++++++++++++++++++++ 2 files changed, 83 insertions(+) create mode 100755 test/native_join_runtime_filter.sh create mode 100644 test/pytest/test_join_runtime_filter.py diff --git a/test/native_join_runtime_filter.sh b/test/native_join_runtime_filter.sh new file mode 100755 index 00000000..2b9842b1 --- /dev/null +++ b/test/native_join_runtime_filter.sh @@ -0,0 +1,39 @@ +#!/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 "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('f',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;" >/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" +base="$(pc "SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQL")" +rf_set="" +if [[ "$(pc "SELECT current_setting('pgcolumnar.enable_join_runtime_filter', true) IS NOT NULL" | tail -1)" == t ]]; then + rf_set="SET pgcolumnar.enable_join_runtime_filter=on;" +fi +on="$(pc "${rf_set}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: yes' <<<"$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_set}$SQL"|tail -1)" "$(pc "$SQL"|tail -1)" +check "runtime answer equals heap" "$(pc "${rf_set}$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 JOIN (SELECT k::bigint k FROM d)x ON f.k=x.k"; do + n=${shape%%|*}; s=${shape#*|}; p="$(pc "${rf_set}EXPLAIN $s")"; check "$n refusal" "$(grep -c 'Columnar Runtime Filter Coordinator'<<<"$p")" 0 +done +pgc_summary diff --git a/test/pytest/test_join_runtime_filter.py b/test/pytest/test_join_runtime_filter.py new file mode 100644 index 00000000..a559149e --- /dev/null +++ b/test/pytest/test_join_runtime_filter.py @@ -0,0 +1,44 @@ +"""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('pgcolumnar.enable_join_runtime_filter', 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(f"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('f',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: yes"),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: + c.execute("SELECT current_setting(pgcolumnar.enable_join_runtime_filter,true)") + if c.fetchone()[0] is not None:c.execute("SET pgcolumnar.enable_join_runtime_filter=on") + c.execute(sql);a=c.fetchone() + if c.execute("SELECT current_setting(pgcolumnar.enable_join_runtime_filter,true)").fetchone()[0] is not None: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.ordered_rows([a],[o],"runtime answer equals off") + expect.ordered_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 JOIN(SELECT k::bigint k FROM d)x ON f.k=x.k")]: + expect.num(_plan(pgc_conn,s,True).count("Columnar Runtime Filter Coordinator"),0,name) From ae623cb95bf3297ba3402d974bef41c023139284 Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Thu, 10 Sep 2026 21:57:59 +0000 Subject: [PATCH 2/6] feat: add serial join runtime range filter (#752) Builds a private core HashPath, blocks to spool and replay the build side, and attaches its conservative interval to the direct columnar scan. Shell and pytest tests independently prove plan shape, exact answers, pruning, and removal causation. --- Makefile | 1 + src/columnar.h | 9 + src/columnar_customscan.c | 31 + src/columnar_reader.c | 87 +++ src/columnar_runtime_filter.c | 771 ++++++++++++++++++++++++ src/columnar_tableam.c | 10 + test/native_join_runtime_filter.sh | 18 +- test/pytest/test_join_runtime_filter.py | 11 +- test/run_all_versions.sh | 1 + 9 files changed, 925 insertions(+), 14 deletions(-) create mode 100644 src/columnar_runtime_filter.c 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/src/columnar.h b/src/columnar.h index a17d5f9b..abe152b8 100644 --- a/src/columnar.h +++ b/src/columnar.h @@ -667,6 +667,10 @@ 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 uint64 PgColumnarRuntimeGroupsRemoved(PgColumnarReadState *readState); extern void PgColumnarReadStats(PgColumnarReadState *readState, uint64 *groupsRead, uint64 *groupsSkipped, uint64 *groupsTotal); @@ -918,6 +922,11 @@ 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 bool PgColumnarAttachRuntimeRange(PlanState *scanState, + AttrNumber attno, Oid subtype, + Datum minimum, Datum maximum); extern Node *PgColumnarCreateAggScanState(CustomScan *cscan); extern Node *PgColumnarCreateGroupAggScanState(CustomScan *cscan); diff --git a/src/columnar_customscan.c b/src/columnar_customscan.c index 302a7885..34d22467 100644 --- a/src/columnar_customscan.c +++ b/src/columnar_customscan.c @@ -123,6 +123,7 @@ typedef struct PgColumnarCustomScanState Datum *projValues; /* scratch, length K+1 (index 0 = rownumber) */ bool *projNulls; PgColumnarLivenessCache *livenessCache; /* cached base liveness for the scan */ + bool runtimeRangeAttached; } PgColumnarCustomScanState; /* path -> plan */ @@ -3709,9 +3710,39 @@ 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); } } +/* + * 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; +} + /* ------------------------------------------------------------------------- * registration * ------------------------------------------------------------------------- */ diff --git a/src/columnar_reader.c b/src/columnar_reader.c index 30187ed2..00785ae0 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,85 @@ 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(&keys[0], 0, attno, + BTGreaterEqualStrategyNumber, InvalidOid, + InvalidOid, InvalidOid, minimum); + keys[0].sk_subtype = subtype; + ScanKeyEntryInitialize(&keys[1], 0, attno, + BTLessEqualStrategyNumber, InvalidOid, + InvalidOid, InvalidOid, maximum); + keys[1].sk_subtype = subtype; + + 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; +} + +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 +1582,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 +4581,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..00b87b73 --- /dev/null +++ b/src/columnar_runtime_filter.c @@ -0,0 +1,771 @@ +/*------------------------------------------------------------------------- + * + * 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 = true; + +typedef struct PgColumnarRuntimeFilterState +{ + CustomScanState customScanState; + PlanState *joinState; + CustomScanState *tapState; + AttrNumber factAttno; + bool rangeAttached; + bool prepared; +} PgColumnarRuntimeFilterState; + +typedef struct PgColumnarRuntimeTapState +{ + CustomScanState customScanState; + PlanState *sourceState; + Tuplestorestate *store; + TupleTableSlot *replaySlot; + AttrNumber keyResno; + Oid keyType; + Oid keyCollation; + FmgrInfo compareFn; + MemoryContext valueContext; + Datum minimum; + Datum maximum; + int16 typeLength; + bool typeByValue; + bool intervalAvailable; + 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) +{ + RestrictInfo *restrictInfo; + OpExpr *operatorExpr; + Node *left; + Node *right; + Var *factVar; + Var *buildVar; + Relids factRelids; + Relids buildRelids; + RangeTblEntry *rte; + + 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; + } + else if (bms_is_member(((Var *) right)->varno, factRelids) && + bms_is_member(((Var *) left)->varno, buildRelids)) + { + factVar = (Var *) right; + buildVar = (Var *) left; + } + 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; + + *factVarOut = factVar; + *buildVarOut = buildVar; + 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; + 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)) + 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; + customPath->path.disabled_nodes = + privateHashPath->jpath.path.disabled_nodes; + 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); + customPath->custom_private = + list_make4(makeInteger(factVar->varattno), + makeInteger(buildVar->varno), + makeInteger(buildVar->varattno), + makeInteger(((OpExpr *) linitial_node(RestrictInfo, + candidate->path_hashclauses)->clause)->opno)); +#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_make1(makeInteger(keyResno)); + 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 = linitial_node(Plan, 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 = TupleDescAttr(ExecGetResultType(state->sourceState), + state->keyResno - 1)->attcollation; + get_typlenbyval(state->keyType, + &state->typeLength, + &state->typeByValue); + { + TypeCacheEntry *typeCache = + lookup_type_cache(state->keyType, TYPECACHE_CMP_PROC_FINFO); + + if (OidIsValid(typeCache->cmp_proc_finfo.fn_oid)) + { + fmgr_info_copy(&state->compareFn, + &typeCache->cmp_proc_finfo, + estate->es_query_cxt); + state->intervalAvailable = 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 +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; + TupleTableSlot *slot; + + if (state->replay) + { + ExecClearTuple(state->replaySlot); + if (!tuplestore_gettupleslot(state->store, + true, + false, + state->replaySlot)) + return NULL; + return PgColumnarRuntimeTapOutput(node, state->replaySlot); + } + + slot = ExecProcNode(state->sourceState); + if (TupIsNull(slot)) + return NULL; + + tuplestore_puttupleslot(state->store, slot); + { + bool isNull; + Datum value = slot_getattr(slot, state->keyResno, &isNull); + + if (!isNull) + { + state->buildRows++; + PgColumnarRuntimeTapAddValue(state, value); + } + } + return PgColumnarRuntimeTapOutput(node, slot); +} + +static void +PgColumnarResetRuntimeTap(PgColumnarRuntimeTapState *state) +{ + tuplestore_clear(state->store); + MemoryContextReset(state->valueContext); + 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); +} + +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 = linitial_node(Plan, 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)); + node->custom_ps = list_make1(state->joinState); +} + +static void +PgColumnarPrepareRuntimeFilter(PgColumnarRuntimeFilterState *state) +{ + PgColumnarRuntimeTapState *tapState = + (PgColumnarRuntimeTapState *) state->tapState; + + while (!TupIsNull(ExecProcNode((PlanState *) state->tapState))) + CHECK_FOR_INTERRUPTS(); + + if (tapState->hasValues && tapState->intervalAvailable) + state->rangeAttached = + PgColumnarAttachRuntimeRange(outerPlanState(state->joinState), + state->factAttno, + tapState->keyType, + tapState->minimum, + tapState->maximum); + + tapState->replay = true; + tuplestore_rescan(tapState->store); + state->prepared = true; +} + +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); +} + +static void +PgColumnarReScanRuntimeFilter(CustomScanState *node) +{ + PgColumnarRuntimeFilterState *state = + (PgColumnarRuntimeFilterState *) node; + + ExecReScan(state->joinState); + PgColumnarResetRuntimeTap((PgColumnarRuntimeTapState *) state->tapState); + 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); +} + +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..77de1a3c 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, + true, + 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/native_join_runtime_filter.sh b/test/native_join_runtime_filter.sh index 2b9842b1..fe31f2cb 100755 --- a/test/native_join_runtime_filter.sh +++ b/test/native_join_runtime_filter.sh @@ -11,12 +11,14 @@ 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;" >/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" -base="$(pc "SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQL")" -rf_set="" +rf_on="" +rf_off="" if [[ "$(pc "SELECT current_setting('pgcolumnar.enable_join_runtime_filter', true) IS NOT NULL" | tail -1)" == t ]]; then - rf_set="SET pgcolumnar.enable_join_runtime_filter=on;" + rf_on="SET pgcolumnar.enable_join_runtime_filter=on;" + rf_off="SET pgcolumnar.enable_join_runtime_filter=off;" fi -on="$(pc "${rf_set}SET max_parallel_workers_per_gather=0;SET enable_nestloop=off;SET enable_mergejoin=off;EXPLAIN(ANALYZE,TIMING off,SUMMARY off)$SQL")" +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 @@ -24,16 +26,16 @@ check "plan has runtime coordinator" "$(grep -c 'Columnar Runtime Filter Coordin 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: yes' <<<"$on")" 1 +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_set}$SQL"|tail -1)" "$(pc "$SQL"|tail -1)" -check "runtime answer equals heap" "$(pc "${rf_set}$SQL"|tail -1)" "$(q 'SELECT count(*),sum(h.k),sum(length(h.p)) FROM h JOIN d ON h.k=d.k')" +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 JOIN (SELECT k::bigint k FROM d)x ON f.k=x.k"; do - n=${shape%%|*}; s=${shape#*|}; p="$(pc "${rf_set}EXPLAIN $s")"; check "$n refusal" "$(grep -c 'Columnar Runtime Filter Coordinator'<<<"$p")" 0 + n=${shape%%|*}; s=${shape#*|}; p="$(pc "${rf_on}EXPLAIN $s")"; check "$n refusal" "$(grep -c 'Columnar Runtime Filter Coordinator'<<<"$p")" 0 done pgc_summary diff --git a/test/pytest/test_join_runtime_filter.py b/test/pytest/test_join_runtime_filter.py index a559149e..68b1e5c2 100644 --- a/test/pytest/test_join_runtime_filter.py +++ b/test/pytest/test_join_runtime_filter.py @@ -28,17 +28,16 @@ def test_serial_join_runtime_filter(pgc_conn,expect): 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: yes"),1,"filter ready before scan") + 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: - c.execute("SELECT current_setting(pgcolumnar.enable_join_runtime_filter,true)") - if c.fetchone()[0] is not None:c.execute("SET pgcolumnar.enable_join_runtime_filter=on") + if _has_runtime_filter(pgc_conn):c.execute("SET pgcolumnar.enable_join_runtime_filter=on") c.execute(sql);a=c.fetchone() - if c.execute("SELECT current_setting(pgcolumnar.enable_join_runtime_filter,true)").fetchone()[0] is not None:c.execute("SET pgcolumnar.enable_join_runtime_filter=off") + 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.ordered_rows([a],[o],"runtime answer equals off") - expect.ordered_rows([a],[h],"runtime answer equals heap") + 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 JOIN(SELECT k::bigint k FROM d)x ON f.k=x.k")]: expect.num(_plan(pgc_conn,s,True).count("Columnar Runtime Filter Coordinator"),0,name) 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 From cbd0c2e349d2f8405239d20ed8ed0ec660c413a2 Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Fri, 11 Sep 2026 01:11:55 +0000 Subject: [PATCH 3/6] feat: add serial join runtime Bloom filter (#752) Scattered keys keep every group, so the interval hull cannot be the skip. Build-side hashes reuse the on-disk bloom saturation cap. Drain the tap from the source, not ExecProcNode, so early LIMIT cannot crash. --- CHANGELOG.md | 18 + docs/ARCHITECTURE.md | 7 + docs/configuration.md | 1 + docs/features.md | 5 + docs/how-to.md | 17 + docs/limitations.md | 12 + docs/user-guide.md | 1 + src/columnar.h | 3 + src/columnar_customscan.c | 58 +++ src/columnar_runtime_filter.c | 255 +++++++-- test/native_join_runtime_filter.sh | 249 ++++++++- test/pytest/TESTS.md | 56 +- test/pytest/test_join_runtime_filter.py | 658 ++++++++++++++++++++++-- 13 files changed, 1256 insertions(+), 84 deletions(-) mode change 100755 => 100644 test/native_join_runtime_filter.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index 64597979..fd04902a 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 on by default. + + `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/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..88af1b78 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 | `on` | Wrap a serial inner Hash Join so the build keys can skip fact-table groups and reject non-matching rows. Direct columnar scan only. | | `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..fda46778 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 on 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..7714ee95 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 on 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 abe152b8..f83e620f 100644 --- a/src/columnar.h +++ b/src/columnar.h @@ -924,6 +924,9 @@ extern void PgColumnarSerializeFlushRows(uint64 storageId, const uint64 *rows, 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); diff --git a/src/columnar_customscan.c b/src/columnar_customscan.c index 34d22467..54eec3ad 100644 --- a/src/columnar_customscan.c +++ b/src/columnar_customscan.c @@ -124,6 +124,9 @@ typedef struct PgColumnarCustomScanState bool *projNulls; PgColumnarLivenessCache *livenessCache; /* cached base liveness for the scan */ bool runtimeRangeAttached; + void *runtimeBloom; + AttrNumber runtimeBloomAttno; + uint64 runtimeRowsRejected; } PgColumnarCustomScanState; /* path -> plan */ @@ -3205,12 +3208,22 @@ 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 (cstate->runtimeBloom != NULL && + !PgColumnarRuntimeBloomMatch(cstate->runtimeBloom, + slot->tts_values[cstate->runtimeBloomAttno - 1], + slot->tts_isnull[cstate->runtimeBloomAttno - 1])) + { + cstate->runtimeRowsRejected++; + return false; + } + ResetExprContext(econtext); econtext->ecxt_scantuple = slot; @@ -3246,12 +3259,19 @@ 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 (cstate->runtimeBloom != NULL && + !PgColumnarRuntimeBloomMatch(cstate->runtimeBloom, + slot->tts_values[cstate->runtimeBloomAttno - 1], + slot->tts_isnull[cstate->runtimeBloomAttno - 1])) + return false; + ResetExprContext(econtext); econtext->ecxt_scantuple = slot; @@ -3435,6 +3455,7 @@ PgColumnarReScanCustomScan(CustomScanState *node) { PgColumnarCustomScanState *cstate = (PgColumnarCustomScanState *) node; + cstate->runtimeRowsRejected = 0; if (cstate->readState != NULL) { PgColumnarRescanRead(cstate->readState); @@ -3714,6 +3735,10 @@ PgColumnarExplainCustomScan(CustomScanState *node, List *ancestors, 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); } } @@ -3724,6 +3749,39 @@ PgColumnarExplainCustomScan(CustomScanState *node, List *ancestors, * 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; + if (state->qualCols == NULL) + state->qualCols = palloc0(sizeof(bool) * state->nTotalColumns); + state->qualCols[attno - 1] = true; + state->lateMat = true; +} + bool PgColumnarAttachRuntimeRange(PlanState *scanState, AttrNumber attno, Oid subtype, Datum minimum, Datum maximum) diff --git a/src/columnar_runtime_filter.c b/src/columnar_runtime_filter.c index 00b87b73..510849ce 100644 --- a/src/columnar_runtime_filter.c +++ b/src/columnar_runtime_filter.c @@ -44,6 +44,16 @@ typedef struct PgColumnarRuntimeFilterState 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; @@ -58,12 +68,17 @@ typedef struct PgColumnarRuntimeTapState 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; @@ -164,7 +179,10 @@ static bool PgColumnarRuntimeFilterVars(PlannerInfo *root, HashPath *hashPath, Var **factVarOut, - Var **buildVarOut) + Var **buildVarOut, + bool *factIsLeftOut, + Oid *hashCollationOut, + bool *intervalAllowedOut) { RestrictInfo *restrictInfo; OpExpr *operatorExpr; @@ -175,6 +193,9 @@ PgColumnarRuntimeFilterVars(PlannerInfo *root, Relids factRelids; Relids buildRelids; RangeTblEntry *rte; + Relation relation; + Oid factCollation; + bool factIsLeft; if (list_length(hashPath->path_hashclauses) != 1) return false; @@ -199,12 +220,14 @@ PgColumnarRuntimeFilterVars(PlannerInfo *root, { 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; @@ -218,8 +241,17 @@ PgColumnarRuntimeFilterVars(PlannerInfo *root, !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; } @@ -276,6 +308,13 @@ PgColumnarSetJoinPathlist(PlannerInfo *root, 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; @@ -308,7 +347,17 @@ PgColumnarSetJoinPathlist(PlannerInfo *root, !PgColumnarRuntimeFilterVars(root, hashPath, &factVar, - &buildVar)) + &buildVar, + &factIsLeft, + &hashCollation, + &intervalAllowed)) + continue; + + if (!get_op_hash_functions( + ((OpExpr *) linitial_node(RestrictInfo, + hashPath->path_hashclauses)->clause)->opno, + &leftHashOid, + &rightHashOid)) continue; candidate = hashPath; @@ -334,8 +383,10 @@ PgColumnarSetJoinPathlist(PlannerInfo *root, 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; @@ -358,12 +409,18 @@ PgColumnarSetJoinPathlist(PlannerInfo *root, 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_make4(makeInteger(factVar->varattno), + list_make5(makeInteger(factVar->varattno), makeInteger(buildVar->varno), makeInteger(buildVar->varattno), - makeInteger(((OpExpr *) linitial_node(RestrictInfo, - candidate->path_hashclauses)->clause)->opno)); + 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 @@ -430,7 +487,11 @@ PgColumnarPlanRuntimeFilterPath(PlannerInfo *root, tapScan->scan.scanrelid = 0; tapScan->flags = 0; tapScan->custom_plans = list_make1(sourcePlan); - tapScan->custom_private = list_make1(makeInteger(keyResno)); + 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; @@ -486,8 +547,7 @@ PgColumnarBeginRuntimeTap(CustomScanState *node, state->keyResno = intVal(linitial(customScan->custom_private)); state->keyType = TupleDescAttr(ExecGetResultType(state->sourceState), state->keyResno - 1)->atttypid; - state->keyCollation = TupleDescAttr(ExecGetResultType(state->sourceState), - state->keyResno - 1)->attcollation; + state->keyCollation = intVal(list_nth(customScan->custom_private, 2)); get_typlenbyval(state->keyType, &state->typeLength, &state->typeByValue); @@ -495,7 +555,8 @@ PgColumnarBeginRuntimeTap(CustomScanState *node, TypeCacheEntry *typeCache = lookup_type_cache(state->keyType, TYPECACHE_CMP_PROC_FINFO); - if (OidIsValid(typeCache->cmp_proc_finfo.fn_oid)) + 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, @@ -503,6 +564,17 @@ PgColumnarBeginRuntimeTap(CustomScanState *node, 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", @@ -513,6 +585,36 @@ PgColumnarBeginRuntimeTap(CustomScanState *node, &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) { @@ -580,35 +682,52 @@ static TupleTableSlot * PgColumnarExecRuntimeTap(CustomScanState *node) { PgColumnarRuntimeTapState *state = (PgColumnarRuntimeTapState *) node; - TupleTableSlot *slot; - if (state->replay) - { - ExecClearTuple(state->replaySlot); - if (!tuplestore_gettupleslot(state->store, - true, - false, - state->replaySlot)) - return NULL; - return PgColumnarRuntimeTapOutput(node, state->replaySlot); - } + if (!state->replay) + elog(ERROR, "pgcolumnar runtime filter tap read before drain"); - slot = ExecProcNode(state->sourceState); - if (TupIsNull(slot)) + ExecClearTuple(state->replaySlot); + if (!tuplestore_gettupleslot(state->store, + true, + false, + state->replaySlot)) return NULL; + return PgColumnarRuntimeTapOutput(node, state->replaySlot); +} - tuplestore_puttupleslot(state->store, slot); - { - bool isNull; - Datum value = slot_getattr(slot, state->keyResno, &isNull); +static void +PgColumnarDrainRuntimeTap(PgColumnarRuntimeTapState *state) +{ + TupleTableSlot *slot; - if (!isNull) + /* + * 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); { - state->buildRows++; - PgColumnarRuntimeTapAddValue(state, value); + bool isNull; + Datum value = slot_getattr(slot, state->keyResno, &isNull); + + if (!isNull) + { + state->buildRows++; + PgColumnarRuntimeTapAddValue(state, value); + PgColumnarRuntimeTapAddHash(state, value); + } } + CHECK_FOR_INTERRUPTS(); } - return PgColumnarRuntimeTapOutput(node, slot); + state->replay = true; + tuplestore_rescan(state->store); } static void @@ -616,6 +735,9 @@ 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; @@ -683,6 +805,19 @@ PgColumnarBeginRuntimeFilter(CustomScanState *node, 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); } @@ -692,8 +827,29 @@ PgColumnarPrepareRuntimeFilter(PgColumnarRuntimeFilterState *state) PgColumnarRuntimeTapState *tapState = (PgColumnarRuntimeTapState *) state->tapState; - while (!TupIsNull(ExecProcNode((PlanState *) state->tapState))) - CHECK_FOR_INTERRUPTS(); + 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 = @@ -703,9 +859,6 @@ PgColumnarPrepareRuntimeFilter(PgColumnarRuntimeFilterState *state) tapState->minimum, tapState->maximum); - tapState->replay = true; - tuplestore_rescan(tapState->store); - state->prepared = true; } static TupleTableSlot * @@ -727,6 +880,8 @@ PgColumnarEndRuntimeFilter(CustomScanState *node) (PgColumnarRuntimeFilterState *) node; ExecEndNode(state->joinState); + if (state->filterContext != NULL) + MemoryContextDelete(state->filterContext); } static void @@ -735,8 +890,17 @@ PgColumnarReScanRuntimeFilter(CustomScanState *node) PgColumnarRuntimeFilterState *state = (PgColumnarRuntimeFilterState *) node; + PgColumnarAttachRuntimeBloom(outerPlanState(state->joinState), + NULL, + InvalidAttrNumber); 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; } @@ -759,6 +923,27 @@ PgColumnarExplainRuntimeFilter(CustomScanState *node, (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 diff --git a/test/native_join_runtime_filter.sh b/test/native_join_runtime_filter.sh old mode 100755 new mode 100644 index fe31f2cb..b4d4b388 --- a/test/native_join_runtime_filter.sh +++ b/test/native_join_runtime_filter.sh @@ -3,17 +3,24 @@ set -uo pipefail . "$(dirname "${BASH_SOURCE[0]}")/lib.sh" pgc_setup "${1:-/usr/lib/postgresql/18/bin/pg_config}" -q "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); +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('f',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;" >/dev/null +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('pgcolumnar.enable_join_runtime_filter', true) IS NOT NULL" | tail -1)" == t ]]; then +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 @@ -30,12 +37,238 @@ check "filter ready before scan" "$(grep -c 'Runtime Filter Ready: true' <<<"$on 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')" +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 JOIN (SELECT k::bigint k FROM d)x ON f.k=x.k"; do + "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")" pgc_summary diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index d6b10b8f..98cc65a3 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,56 @@ 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`. + diff --git a/test/pytest/test_join_runtime_filter.py b/test/pytest/test_join_runtime_filter.py index 68b1e5c2..1fb263f8 100644 --- a/test/pytest/test_join_runtime_filter.py +++ b/test/pytest/test_join_runtime_filter.py @@ -1,43 +1,623 @@ """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('pgcolumnar.enable_join_runtime_filter', 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(f"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('f',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 JOIN(SELECT k::bigint k FROM d)x ON f.k=x.k")]: - expect.num(_plan(pgc_conn,s,True).count("Columnar Runtime Filter Coordinator"),0,name) + 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") From 2733b02bf87d17cb066622a27a6d09af5934672b Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Fri, 11 Sep 2026 01:12:04 +0000 Subject: [PATCH 4/6] test: restore executable bit on native_join_runtime_filter.sh The Bloom commit dropped the mode the RED commit set. A shebang without the bit fails harness_selftest. --- test/native_join_runtime_filter.sh | 0 1 file changed, 0 insertions(+), 0 deletions(-) mode change 100644 => 100755 test/native_join_runtime_filter.sh diff --git a/test/native_join_runtime_filter.sh b/test/native_join_runtime_filter.sh old mode 100644 new mode 100755 From 4ec98f6ae16f74c5bc9d82f4aec7e6fb772517c9 Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Fri, 11 Sep 2026 01:45:54 +0000 Subject: [PATCH 5/6] fix: decode every qual column before the runtime bloom probe (#752) A fact-table qual that named a non-key column returned no rows: attach forced late materialization with only the join key decoded. Derive the qual columns the same way Begin does, and probe Bloom after the full row when that path is refused. Also drop linitial_node(Plan), stop initializing a ScanKey with an invalid procedure, default the GUC off, and detach a stale hull on rescan. --- CHANGELOG.md | 2 +- docs/configuration.md | 2 +- docs/features.md | 2 +- docs/how-to.md | 2 +- src/columnar.h | 2 + src/columnar_customscan.c | 138 ++++++++++++++++++++---- src/columnar_reader.c | 28 +++-- src/columnar_runtime_filter.c | 10 +- src/columnar_tableam.c | 2 +- test/native_join_runtime_filter.sh | 20 ++++ test/pytest/TESTS.md | 5 + test/pytest/test_join_runtime_filter.py | 49 +++++++++ 12 files changed, 225 insertions(+), 37 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fd04902a..93635133 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,7 +30,7 @@ true until the next version shipped. The path is serial and INNER only. LEFT, SEMI, ANTI, CROSS, parallel, and projection-backed outers are refused. `pgcolumnar.enable_join_runtime_filter` - is on by default. + 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 diff --git a/docs/configuration.md b/docs/configuration.md index 88af1b78..ddab390c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -61,7 +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 | `on` | Wrap a serial inner Hash Join so the build keys can skip fact-table groups and reject non-matching rows. Direct columnar scan only. | +| `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 fda46778..7d3dad9b 100644 --- a/docs/features.md +++ b/docs/features.md @@ -73,7 +73,7 @@ settings see the [configuration reference](configuration.md); for constraints se - 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 on by default. + 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 diff --git a/docs/how-to.md b/docs/how-to.md index 7714ee95..46f112cd 100644 --- a/docs/how-to.md +++ b/docs/how-to.md @@ -143,7 +143,7 @@ It also rejects rows whose keys are absent from a Bloom filter of those keys. EXPLAIN (ANALYZE) SELECT sum(amount) FROM fact JOIN dim ON fact.k = dim.k; ``` -**Tuning.** It is on by default (`pgcolumnar.enable_join_runtime_filter`). +**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. diff --git a/src/columnar.h b/src/columnar.h index f83e620f..1b0b1fbd 100644 --- a/src/columnar.h +++ b/src/columnar.h @@ -670,6 +670,7 @@ extern void PgColumnarReadSetParallelCounter(PgColumnarReadState *readState, 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, @@ -930,6 +931,7 @@ 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 54eec3ad..d795950d 100644 --- a/src/columnar_customscan.c +++ b/src/columnar_customscan.c @@ -3204,6 +3204,20 @@ 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) { @@ -3215,14 +3229,8 @@ pgcolumnar_scan_row_filter(void *arg) ExecClearTuple(slot); ExecStoreVirtualTuple(slot); - if (cstate->runtimeBloom != NULL && - !PgColumnarRuntimeBloomMatch(cstate->runtimeBloom, - slot->tts_values[cstate->runtimeBloomAttno - 1], - slot->tts_isnull[cstate->runtimeBloomAttno - 1])) - { - cstate->runtimeRowsRejected++; + if (!pgcolumnar_runtime_bloom_keeps(cstate, slot)) return false; - } ResetExprContext(econtext); econtext->ecxt_scantuple = slot; @@ -3266,10 +3274,7 @@ pgcolumnar_scan_row_filter_nocount(void *arg) ExecClearTuple(slot); ExecStoreVirtualTuple(slot); - if (cstate->runtimeBloom != NULL && - !PgColumnarRuntimeBloomMatch(cstate->runtimeBloom, - slot->tts_values[cstate->runtimeBloomAttno - 1], - slot->tts_isnull[cstate->runtimeBloomAttno - 1])) + if (!pgcolumnar_runtime_bloom_keeps(cstate, slot)) return false; ResetExprContext(econtext); @@ -3425,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); @@ -3743,11 +3761,67 @@ PgColumnarExplainCustomScan(CustomScanState *node, List *ancestors, } /* - * 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. + * 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) @@ -3776,12 +3850,16 @@ PgColumnarAttachRuntimeBloom(PlanState *scanState, void *filter, AttrNumber attn state->runtimeBloom = filter; state->runtimeBloomAttno = attno; state->runtimeRowsRejected = 0; - if (state->qualCols == NULL) - state->qualCols = palloc0(sizeof(bool) * state->nTotalColumns); - state->qualCols[attno - 1] = true; - state->lateMat = true; + 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) @@ -3801,6 +3879,20 @@ PgColumnarAttachRuntimeRange(PlanState *scanState, AttrNumber attno, Oid subtype 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 00785ae0..b980d020 100644 --- a/src/columnar_reader.c +++ b/src/columnar_reader.c @@ -883,14 +883,20 @@ PgColumnarReadSetRuntimeRange(PgColumnarReadState *readState, MemSet(keys, 0, sizeof(keys)); MemSet(built, 0, sizeof(built)); - ScanKeyEntryInitialize(&keys[0], 0, attno, - BTGreaterEqualStrategyNumber, InvalidOid, - InvalidOid, InvalidOid, minimum); + /* + * 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; - ScanKeyEntryInitialize(&keys[1], 0, attno, - BTLessEqualStrategyNumber, InvalidOid, - InvalidOid, InvalidOid, maximum); + 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, @@ -932,6 +938,16 @@ PgColumnarReadSetRuntimeRange(PgColumnarReadState *readState, return true; } +void +PgColumnarReadClearRuntimeRange(PgColumnarReadState *readState) +{ + if (readState == NULL || readState->runtimePredicateCount == 0) + return; + readState->numPredicates = readState->runtimePredicateStart; + readState->runtimePredicateCount = 0; +} + + uint64 PgColumnarRuntimeGroupsRemoved(PgColumnarReadState *readState) { diff --git a/src/columnar_runtime_filter.c b/src/columnar_runtime_filter.c index 510849ce..cc8e6348 100644 --- a/src/columnar_runtime_filter.c +++ b/src/columnar_runtime_filter.c @@ -36,7 +36,7 @@ #include "utils/typcache.h" #include "utils/tuplestore.h" -bool pgcolumnar_enable_join_runtime_filter = true; +bool pgcolumnar_enable_join_runtime_filter = false; typedef struct PgColumnarRuntimeFilterState { @@ -541,7 +541,7 @@ PgColumnarBeginRuntimeTap(CustomScanState *node, if (list_length(customScan->custom_plans) != 1) elog(ERROR, "pgcolumnar runtime filter tap expected one source plan"); - sourcePlan = linitial_node(Plan, customScan->custom_plans); + 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)); @@ -773,6 +773,9 @@ PgColumnarExplainRuntimeTap(CustomScanState *node, NULL, (int64) state->buildRows, es); + ExplainPropertyText("Runtime Filter Drain", + "coordinator drained the source; Hash replays the spool", + es); } static void @@ -791,7 +794,7 @@ PgColumnarBeginRuntimeFilter(CustomScanState *node, elog(ERROR, "pgcolumnar runtime filter expected one core Hash Join plan"); - childPlan = linitial_node(Plan, customScan->custom_plans); + childPlan = (Plan *) linitial(customScan->custom_plans); state->joinState = ExecInitNode(childPlan, estate, eflags); if (!IsA(state->joinState, HashJoinState)) elog(ERROR, @@ -893,6 +896,7 @@ PgColumnarReScanRuntimeFilter(CustomScanState *node) PgColumnarAttachRuntimeBloom(outerPlanState(state->joinState), NULL, InvalidAttrNumber); + PgColumnarDetachRuntimeRange(outerPlanState(state->joinState)); ExecReScan(state->joinState); PgColumnarResetRuntimeTap((PgColumnarRuntimeTapState *) state->tapState); MemoryContextReset(state->filterContext); diff --git a/src/columnar_tableam.c b/src/columnar_tableam.c index 77de1a3c..80af1cbd 100644 --- a/src/columnar_tableam.c +++ b/src/columnar_tableam.c @@ -3456,7 +3456,7 @@ _PG_init(void) "Enable serial hash-join runtime filtering for direct columnar scans.", NULL, &pgcolumnar_enable_join_runtime_filter, - true, + false, PGC_USERSET, 0, NULL, NULL, NULL); diff --git a/test/native_join_runtime_filter.sh b/test/native_join_runtime_filter.sh index b4d4b388..e3ab2503 100755 --- a/test/native_join_runtime_filter.sh +++ b/test/native_join_runtime_filter.sh @@ -271,4 +271,24 @@ SQLL="SELECT fact_lim.k FROM fact_lim JOIN dim_lim ON fact_lim.k=dim_lim.k ORDER 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 98cc65a3..069bcbda 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -2254,3 +2254,8 @@ A covering projection scan stays an unwrapped Hash Join outer. `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 index 1fb263f8..27263884 100644 --- a/test/pytest/test_join_runtime_filter.py +++ b/test/pytest/test_join_runtime_filter.py @@ -621,3 +621,52 @@ def test_early_limit_matches_heap(pgc_conn, expect): 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") From 760fbd60a198c36c97f6f27f83813d2d2a737c4f Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Fri, 11 Sep 2026 01:48:28 +0000 Subject: [PATCH 6/6] test: seed the mutation ledger with native_join_runtime_filter (#752) Adding the suite raised suites_not_covered, and that ceiling may only fall. A green run of the 44 checks is merged as never-red so the suite is covered and the census matches. --- test/check_ledger.tsv | 44 ++++++++++++++++++++++++++++++++++++ test/check_ledger_budget.txt | 2 +- 2 files changed, 45 insertions(+), 1 deletion(-) 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