feat: serial hash-join runtime filter (#752)

CHANGELOG.md

@@ -18,6 +18,24 @@ true until the next version shipped.
 
 ### Added
 
+- A serial inner Hash Join can push the build-side keys into a direct
+  columnar scan (#752).
+
+  The coordinator wraps core Hash Join. It does not keep a HashPath from
+  `joinrel->pathlist`. It builds a private path, drains the build side into a
+  tuplestore, then lets Hash replay that spool. The scan skips chunk groups
+  outside the conservative key interval when types and collations match. It
+  also rejects non-matching rows with a Bloom filter of those keys, using the
+  same saturation cap as on-disk bloom filters.
+
+  The path is serial and INNER only. LEFT, SEMI, ANTI, CROSS, parallel, and
+  projection-backed outers are refused. `pgcolumnar.enable_join_runtime_filter`
+  is off by default until the skip is measured on the join fixture.
+
+  `EXPLAIN (ANALYZE)` reports `Runtime Filter Groups Removed` and
+  `Runtime Filter Rows Rejected`. Those counters are dedicated. They are not
+  `InstrCountFiltered1`.
+
 - Every check result is machine-readable, and counting a check is the same
   operation as recording it (#917).
 

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 \

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

docs/configuration.md

@@ -61,6 +61,7 @@ disk. It never changes the values that a table returns.
 | `pgcolumnar.enable_group_vectorization` | boolean | `off` | Use the vectorized aggregate path for `GROUP BY` queries on a columnar table. Off by default; see [why grouped vectorization is off by default](#why-grouped-vectorization-is-off-by-default). |
 | `pgcolumnar.groupagg_max_groups` | integer | `1000000` | Cap on the group count the grouped vectorized aggregate builds. Over the cap the query errors. Range 1 to INT_MAX. |
 | `pgcolumnar.enable_bloom_filter` | boolean | `on` | Skip chunk groups on equality filters using per-chunk bloom filters. |
+| `pgcolumnar.enable_join_runtime_filter` | boolean | `off` | Wrap a serial inner Hash Join so the build keys can skip fact-table groups and reject non-matching rows. Direct columnar scan only. Off until the skip is measured. |
 | `pgcolumnar.enable_read_stream` | boolean | `on` | Prefetch block reads with the read stream API. Effective on PostgreSQL 17 and later. |
 | `pgcolumnar.enable_ungrouped_vector_agg` | boolean | `off` | Answer an ungrouped aggregate (`count`, `sum`, `avg`, `min`, `max` with no `GROUP BY`) with a batch fold over decoded vectors instead of row-at-a-time. Off by default. |
 | `pgcolumnar.enable_parallel_vector_agg` | boolean | `off` | Let the ungrouped batch fold run as a parallel partial aggregate under `Gather`, each worker folding its own row groups. Requires `pgcolumnar.enable_ungrouped_vector_agg`. Off by default. |

docs/features.md

@@ -70,6 +70,11 @@ settings see the [configuration reference](configuration.md); for constraints se
 - A fetch by row number decodes only the columns that the executor asks for. It
   keeps the decoded row group for the rest of the statement. An index-driven read
   of a wide table therefore does not decode the columns that it will not return.
+- Serial join runtime filter for a star-schema Hash Join.
+  A serial inner Hash Join can skip fact-table groups using the build-side key range.
+  It can also reject non-matching rows with a Bloom filter of those keys.
+  The GUC `pgcolumnar.enable_join_runtime_filter` is off by default.
+  It does not wrap LEFT, SEMI, ANTI, CROSS, parallel, or projection scans.
 - Parallel scan across a table's row groups.
 - Read stream prefetch of block reads on PostgreSQL 17 and later
   (`pgcolumnar.enable_read_stream`).

docs/how-to.md

@@ -132,6 +132,23 @@ then sit in few chunk groups. Read `Columnar Chunk Groups Removed by Filter` to
 confirm the skip. A scattered high-cardinality column gains little. Turn the
 feature off with `SET pgcolumnar.enable_bloom_filter = off` to compare.
 
+
+## Skip fact-table work under a star-schema join
+
+A serial inner Hash Join can push the dimension keys into the fact-table scan.
+The scan drops chunk groups outside the build-side key range.
+It also rejects rows whose keys are absent from a Bloom filter of those keys.
+
+```sql
+EXPLAIN (ANALYZE) SELECT sum(amount) FROM fact JOIN dim ON fact.k = dim.k;
+```
+
+**Tuning.** It is off by default (`pgcolumnar.enable_join_runtime_filter`).
+It applies only to a serial inner Hash Join whose outer path is a direct columnar scan.
+A LEFT, SEMI, ANTI, or CROSS join is unchanged.
+A covering projection is also unchanged.
+Read `Runtime Filter Groups Removed` and `Runtime Filter Rows Rejected` to confirm the skip.
+
 ## Add a projection for a second sort order
 
 A table has one physical sort order. A projection stores a column subset a second

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.

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
 

src/columnar.h

@@ -667,6 +667,11 @@ extern void PgColumnarReadSetParallelCounter(PgColumnarReadState *readState,
  * scan's EXPLAIN output to show how many chunk groups the min/max skip lists
  * removed. total = read + skipped over the groups the scan has reached.
  */
+extern bool PgColumnarReadSetRuntimeRange(PgColumnarReadState *readState,
+										AttrNumber attno, Oid subtype,
+										Datum minimum, Datum maximum);
+extern void PgColumnarReadClearRuntimeRange(PgColumnarReadState *readState);
+extern uint64 PgColumnarRuntimeGroupsRemoved(PgColumnarReadState *readState);
 extern void PgColumnarReadStats(PgColumnarReadState *readState,
 							  uint64 *groupsRead, uint64 *groupsSkipped,
 							  uint64 *groupsTotal);
@@ -918,6 +923,15 @@ extern void PgColumnarSerializeFlushRows(uint64 storageId, const uint64 *rows,
  * a scanrelid==0 upper node is the vectorized aggregate.
  */
 extern const CustomScanMethods pgcolumnar_scan_methods;
+extern bool pgcolumnar_enable_join_runtime_filter;
+extern void PgColumnarRuntimeFilterInit(void);
+extern void PgColumnarAttachRuntimeBloom(PlanState *scanState,
+										 void *filter, AttrNumber attno);
+extern bool PgColumnarRuntimeBloomMatch(void *filter, Datum value, bool isNull);
+extern bool PgColumnarAttachRuntimeRange(PlanState *scanState,
+										 AttrNumber attno, Oid subtype,
+										 Datum minimum, Datum maximum);
+extern void PgColumnarDetachRuntimeRange(PlanState *scanState);
 extern Node *PgColumnarCreateAggScanState(CustomScan *cscan);
 extern Node *PgColumnarCreateGroupAggScanState(CustomScan *cscan);
 

src/columnar_customscan.c

@@ -123,6 +123,10 @@ typedef struct PgColumnarCustomScanState
 	Datum	   *projValues;			/* scratch, length K+1 (index 0 = rownumber) */
 	bool	   *projNulls;
 	PgColumnarLivenessCache *livenessCache;	/* cached base liveness for the scan */
+	bool		runtimeRangeAttached;
+	void	   *runtimeBloom;
+	AttrNumber	runtimeBloomAttno;
+	uint64		runtimeRowsRejected;
 } PgColumnarCustomScanState;
 
 /* path -> plan */
@@ -3200,16 +3204,34 @@ pgcolumnar_setup_late_materialization(PgColumnarCustomScanState *cstate,
  * is reset per evaluation exactly as ExecScan resets it per fetched tuple, so a
  * scan that rejects millions of rows does not accumulate their qual allocations.
  */
+static bool
+pgcolumnar_runtime_bloom_keeps(PgColumnarCustomScanState *cstate,
+							   TupleTableSlot *slot)
+{
+	if (cstate->runtimeBloom == NULL)
+		return true;
+	if (PgColumnarRuntimeBloomMatch(cstate->runtimeBloom,
+								 slot->tts_values[cstate->runtimeBloomAttno - 1],
+								 slot->tts_isnull[cstate->runtimeBloomAttno - 1]))
+		return true;
+	cstate->runtimeRowsRejected++;
+	return false;
+}
+
 static bool
 pgcolumnar_scan_row_filter(void *arg)
 {
 	ScanState  *ss = (ScanState *) arg;
+	PgColumnarCustomScanState *cstate = (PgColumnarCustomScanState *) ss;
 	ExprContext *econtext = ss->ps.ps_ExprContext;
 	TupleTableSlot *slot = ss->ss_ScanTupleSlot;
 
 	ExecClearTuple(slot);
 	ExecStoreVirtualTuple(slot);
 
+	if (!pgcolumnar_runtime_bloom_keeps(cstate, slot))
+		return false;
+
 	ResetExprContext(econtext);
 	econtext->ecxt_scantuple = slot;
 
@@ -3245,12 +3267,16 @@ static bool
 pgcolumnar_scan_row_filter_nocount(void *arg)
 {
 	ScanState  *ss = (ScanState *) arg;
+	PgColumnarCustomScanState *cstate = (PgColumnarCustomScanState *) ss;
 	ExprContext *econtext = ss->ps.ps_ExprContext;
 	TupleTableSlot *slot = ss->ss_ScanTupleSlot;
 
 	ExecClearTuple(slot);
 	ExecStoreVirtualTuple(slot);
 
+	if (!pgcolumnar_runtime_bloom_keeps(cstate, slot))
+		return false;
+
 	ResetExprContext(econtext);
 	econtext->ecxt_scantuple = slot;
 
@@ -3404,9 +3430,22 @@ PgColumnarScanNext(ScanState *ss)
 
 		ExecClearTuple(slot);
 	}
-	else if (!PgColumnarReadNextRow(cstate->readState, slot->tts_values,
-									slot->tts_isnull, &rowNumber))
-		return NULL;
+	else
+	{
+		/*
+		 * When late materialization is off, the Bloom probe cannot run in the
+		 * two-pass filter: that path is what would leave qual columns undecoded.
+		 * Probe after the full row is built instead.
+		 */
+		for (;;)
+		{
+			if (!PgColumnarReadNextRow(cstate->readState, slot->tts_values,
+									   slot->tts_isnull, &rowNumber))
+				return NULL;
+			if (pgcolumnar_runtime_bloom_keeps(cstate, slot))
+				break;
+		}
+	}
 
 	ExecStoreVirtualTuple(slot);
 	PgColumnarRowNumberToItemPointer(rowNumber, &slot->tts_tid);
@@ -3434,6 +3473,7 @@ PgColumnarReScanCustomScan(CustomScanState *node)
 {
 	PgColumnarCustomScanState *cstate = (PgColumnarCustomScanState *) node;
 
+	cstate->runtimeRowsRejected = 0;
 	if (cstate->readState != NULL)
 	{
 		PgColumnarRescanRead(cstate->readState);
@@ -3709,9 +3749,150 @@ PgColumnarExplainCustomScan(CustomScanState *node, List *ancestors,
 		ExplainPropertyInteger("Columnar Rows Filtered Before Materialization", NULL,
 							   (int64) PgColumnarRowsFilteredEarly(cstate->readState),
 							   es);
+		if (cstate->runtimeRangeAttached)
+			ExplainPropertyInteger("Runtime Filter Groups Removed", NULL,
+								   (int64) PgColumnarRuntimeGroupsRemoved(cstate->readState),
+								   es);
+		if (cstate->runtimeBloom != NULL)
+			ExplainPropertyInteger("Runtime Filter Rows Rejected", NULL,
+								   (int64) cstate->runtimeRowsRejected,
+								   es);
 	}
 }
 
+/*
+ * pgcolumnar_ensure_runtime_bloom_columns
+ *		The two-pass filter must decode every column ExecQual reads, plus the
+ *		join key.  Turning late materialization on with only the key marked
+ *		drops rows whose qual names any other column.
+ *
+ * Do not force the path when the GUC is off or the qual is volatile.  Bloom
+ * then probes in PgColumnarScanNext after the full row is built.
+ */
+static void
+pgcolumnar_ensure_runtime_bloom_columns(PgColumnarCustomScanState *state,
+									   AttrNumber attno)
+{
+	CustomScan *cscan = (CustomScan *) state->css.ss.ps.plan;
+	Bitmapset  *qualAttrs = NULL;
+	int			natts = state->nTotalColumns;
+	int			x = -1;
+
+	if (!pgcolumnar_enable_late_materialization)
+		return;
+
+	if (state->qualCols != NULL)
+	{
+		state->qualCols[attno - 1] = true;
+		return;
+	}
+
+	if (cscan->scan.plan.qual != NIL &&
+		contain_volatile_functions((Node *) cscan->scan.plan.qual))
+		return;
+
+	if (cscan->scan.plan.qual != NIL)
+	{
+		pull_varattnos((Node *) cscan->scan.plan.qual, cscan->scan.scanrelid,
+					   &qualAttrs);
+		while ((x = bms_next_member(qualAttrs, x)) >= 0)
+		{
+			AttrNumber	qattno = x + FirstLowInvalidHeapAttributeNumber;
+
+			if (qattno <= 0 || qattno > natts)
+				return;
+		}
+	}
+
+	state->qualCols = palloc0(sizeof(bool) * natts);
+	x = -1;
+	while ((x = bms_next_member(qualAttrs, x)) >= 0)
+	{
+		AttrNumber	qattno = x + FirstLowInvalidHeapAttributeNumber;
+
+		state->qualCols[qattno - 1] = true;
+	}
+	state->qualCols[attno - 1] = true;
+	state->lateMat = true;
+}
+
+/*
+ * PgColumnarAttachRuntimeBloom
+ *		Publish a completed build-side Bloom filter to a direct base scan
+ *		before its first tuple is requested.  Projection scans are excluded
+ *		by the planner; checking again here turns a planner mistake into an
+ *		error, not a wrong answer.
+ */
+void
+PgColumnarAttachRuntimeBloom(PlanState *scanState, void *filter, AttrNumber attno)
+{
+	PgColumnarCustomScanState *state;
+
+	if (scanState == NULL)
+		return;
+	if (!IsA(scanState, CustomScanState))
+		elog(ERROR, "pgcolumnar runtime bloom expected a custom scan");
+	state = (PgColumnarCustomScanState *) scanState;
+	if (state->css.methods != &pgcolumnar_exec_methods)
+		elog(ERROR, "pgcolumnar runtime bloom expected a columnar scan");
+
+	if (filter == NULL)
+	{
+		state->runtimeBloom = NULL;
+		state->runtimeBloomAttno = InvalidAttrNumber;
+		return;
+	}
+	if (state->projScan || state->readState == NULL)
+		elog(ERROR, "pgcolumnar runtime bloom expected a direct base scan");
+	if (attno <= 0 || attno > state->nTotalColumns)
+		elog(ERROR, "pgcolumnar runtime bloom key is out of range");
+
+	state->runtimeBloom = filter;
+	state->runtimeBloomAttno = attno;
+	state->runtimeRowsRejected = 0;
+	pgcolumnar_ensure_runtime_bloom_columns(state, attno);
+}
+
+/*
+ * PgColumnarAttachRuntimeRange
+ *		Publish a completed build-side hull to a direct base scan before its
+ *		first tuple is requested.  Projection scans are excluded by the planner;
+ *		checking again here turns a planner mistake into an error, not a wrong
+ *		answer.
+ */
+bool
+PgColumnarAttachRuntimeRange(PlanState *scanState, AttrNumber attno, Oid subtype,
+							 Datum minimum, Datum maximum)
+{
+	PgColumnarCustomScanState *state;
+
+	if (!IsA(scanState, CustomScanState))
+		elog(ERROR, "pgcolumnar runtime range expected a custom scan");
+	state = (PgColumnarCustomScanState *) scanState;
+	if (state->css.methods != &pgcolumnar_exec_methods || state->projScan ||
+		state->readState == NULL)
+		elog(ERROR, "pgcolumnar runtime range expected a direct base scan");
+
+	state->runtimeRangeAttached =
+		PgColumnarReadSetRuntimeRange(state->readState, attno, subtype,
+									  minimum, maximum);
+	return state->runtimeRangeAttached;
+}
+
+void
+PgColumnarDetachRuntimeRange(PlanState *scanState)
+{
+	PgColumnarCustomScanState *state;
+
+	if (scanState == NULL || !IsA(scanState, CustomScanState))
+		return;
+	state = (PgColumnarCustomScanState *) scanState;
+	if (state->css.methods != &pgcolumnar_exec_methods || state->readState == NULL)
+		return;
+	PgColumnarReadClearRuntimeRange(state->readState);
+	state->runtimeRangeAttached = false;
+}
+
 /* -------------------------------------------------------------------------
  * registration
  * ------------------------------------------------------------------------- */