From 472656c6eb73fe9478fadc145a7438a7735503e8 Mon Sep 17 00:00:00 2001
Message-ID: <472656c6eb73fe9478fadc145a7438a7735503e8.1785332874.git.darthunix@gmail.com>
In-Reply-To: <59baf8c1261da4518020e296480b4bc1dd71c31e.1785332874.git.darthunix@gmail.com>
References: <59baf8c1261da4518020e296480b4bc1dd71c31e.1785332874.git.darthunix@gmail.com>
From: Denis Smirnov <darthunix@gmail.com>
Date: Tue, 21 Jul 2026 17:14:58 +0700
Subject: [PATCH v4 2/3] Evaluate SeqScan quals in tuple batches

Tuple slots can expose several tuples that a scan has already made
available, even though executor nodes continue to exchange one tuple at
a time. Use this facility to evaluate eligible SeqScan quals in dense
loops. This improves locality without introducing a batch expression API
or changing the executor row-at-a-time interface.

Recognize a leading sequence of simple Var-op-Const and Var-op-Param
expressions, with either argument order. Only Vars read from the scan
tuple are eligible. Operators may use arbitrary and cross-type
arguments, but must return boolean, not return a set, and be marked
IMMUTABLE, STRICT, and LEAKPROOF. Invoke them through the existing fmgr
interface, including normal collation and function statistics handling.

Process up to 64 tuples at a time. Fetch each required attribute through
the tuple slot batch interface, then evaluate the operator in a dense
loop. Represent surviving rows with a TupleBatchMask and narrow that
mask for each subsequent qual. Reuse fetched values for adjacent quals
on the same attribute.

Run each dense qual loop in a dedicated memory context and reset it after
the loop. This amortizes context switching and reset overhead while
limiting temporary allocations left by an operator to one tuple batch.

Batch preparation is intentionally eager: before returning a qualifying
tuple, the executor may evaluate eligible quals for up to 64 later
tuples in the same window. A parent that stops early, such as Limit or
an EXISTS subplan, can therefore leave some of those evaluations unused.
Choosing between scalar and batch evaluation in such cases would require
combining the parent's tuple bound with qual selectivity and propagating
that decision to the scan. Leave that planner/executor plumbing for
future work rather than complicating this patch, and accept up to the
remainder of one window as speculative work each time a parent stops
early.

Preserve the qual order chosen by the planner. Only the supported
leading sequence is evaluated in batches; the first unsupported qual and
everything after it remain in the normal per-tuple ExecQual path. If no
active batch is available, execute the complete qual normally.
Projections are still performed one tuple at a time and only for rows
that pass all quals.

Put the mechanism in the common scan executor and expose initialization,
execution, and reset functions so extension scan nodes can reuse it.
ExecScanBatch() is their public execution entry point. SeqScan calls the
inline extended helper directly to preserve access-method inlining and
projection specialization. EPQ scans and slot types without batch
support retain the existing path, and parent nodes continue to receive
one ordinary TupleTableSlot per call.
---
 src/backend/executor/execScan.c          | 359 +++++++++++++++++++++++
 src/backend/executor/execScanBatch.h     | 140 +++++++++
 src/backend/executor/nodeSeqscan.c       |  49 +++-
 src/include/executor/executor.h          |  13 +
 src/include/nodes/execnodes.h            |   2 +
 src/test/regress/expected/scan_batch.out | 241 +++++++++++++++
 src/test/regress/parallel_schedule       |   2 +-
 src/test/regress/sql/scan_batch.sql      | 147 ++++++++++
 src/tools/pgindent/typedefs.list         |   1 +
 9 files changed, 951 insertions(+), 3 deletions(-)
 create mode 100644 src/backend/executor/execScanBatch.h
 create mode 100644 src/test/regress/expected/scan_batch.out
 create mode 100644 src/test/regress/sql/scan_batch.sql

diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c
index 9f68be17b99..4dc6f52996b 100644
--- a/src/backend/executor/execScan.c
+++ b/src/backend/executor/execScan.c
@@ -18,9 +18,336 @@
  */
 #include "postgres.h"
 
+#include "access/htup_details.h"
+#include "catalog/pg_proc.h"
+#include "catalog/pg_type.h"
+#include "execScanBatch.h"
 #include "executor/executor.h"
 #include "executor/execScan.h"
 #include "miscadmin.h"
+#include "pgstat.h"
+#include "utils/memutils.h"
+#include "utils/syscache.h"
+
+static Node *
+strip_batch_qual_relabel(Node *node)
+{
+	if (IsA(node, RelabelType))
+		node = (Node *) castNode(RelabelType, node)->arg;
+	return node;
+}
+
+static bool
+is_batch_qual_value(Node *node)
+{
+	return IsA(node, Const) ||
+		(IsA(node, Param) &&
+		 (castNode(Param, node)->paramkind == PARAM_EXTERN ||
+		  castNode(Param, node)->paramkind == PARAM_EXEC));
+}
+
+static bool
+is_batch_qual_function(Oid funcid)
+{
+	HeapTuple	proctup;
+	Form_pg_proc procform;
+	bool		result;
+
+	proctup = SearchSysCache1(PROCOID, ObjectIdGetDatum(funcid));
+	if (!HeapTupleIsValid(proctup))
+		elog(ERROR, "cache lookup failed for function %u", funcid);
+	procform = (Form_pg_proc) GETSTRUCT(proctup);
+	result = procform->prorettype == BOOLOID && !procform->proretset &&
+		procform->provolatile == PROVOLATILE_IMMUTABLE &&
+		procform->proisstrict && procform->proleakproof;
+	ReleaseSysCache(proctup);
+
+	return result;
+}
+
+static OpExpr *
+match_batch_qual(Node *clause, uint8 *var_argno)
+{
+	OpExpr	   *op;
+	Node	   *args[2];
+	Var		   *var;
+
+	if (!IsA(clause, OpExpr))
+		return NULL;
+	op = (OpExpr *) clause;
+	if (list_length(op->args) != 2)
+		return NULL;
+
+	args[0] = strip_batch_qual_relabel(linitial(op->args));
+	args[1] = strip_batch_qual_relabel(lsecond(op->args));
+	if (IsA(args[0], Var) && is_batch_qual_value(args[1]))
+		*var_argno = 0;
+	else if (is_batch_qual_value(args[0]) && IsA(args[1], Var))
+		*var_argno = 1;
+	else
+		return NULL;
+	var = castNode(Var, args[*var_argno]);
+
+	if (var->varno == INNER_VAR || var->varno == OUTER_VAR ||
+		var->varattno <= 0 || var->varlevelsup != 0 ||
+		var->varreturningtype != VAR_RETURNING_DEFAULT)
+		return NULL;
+
+	return op;
+}
+
+static void
+init_batch_qual(struct ExecScanBatchQual *qual, OpExpr *op,
+				uint8 var_argno, PlanState *parent)
+{
+	Node	   *arg;
+	Var		   *var;
+
+	arg = strip_batch_qual_relabel(list_nth(op->args, 1 - var_argno));
+	var = castNode(Var,
+				   strip_batch_qual_relabel(list_nth(op->args, var_argno)));
+	fmgr_info(op->opfuncid, &qual->func);
+	fmgr_info_set_expr((Node *) op, &qual->func);
+	qual->fcinfo = palloc0(SizeForFunctionCallInfo(2));
+	InitFunctionCallInfoData(*qual->fcinfo, &qual->func, 2,
+							 op->inputcollid, NULL, NULL);
+	if (IsA(arg, Const))
+	{
+		qual->constant.value = castNode(Const, arg)->constvalue;
+		qual->constant.isnull = castNode(Const, arg)->constisnull;
+	}
+	else
+	{
+		Assert(IsA(arg, Param));
+		qual->param_expr = ExecInitExpr((Expr *) arg, parent);
+	}
+	qual->attnum = var->varattno;
+	qual->var_argno = var_argno;
+}
+
+static Datum
+ExecScanBatchQual(ExprState *state, ExprContext *econtext, bool *isnull)
+{
+	ExecScanBatchState *batch_state = state->evalfunc_private;
+	bool		result;
+
+	result = ExecQual(batch_state->prefix_qual, econtext);
+	if (result)
+		result = ExecQual(batch_state->rest_qual, econtext);
+	*isnull = false;
+
+	return BoolGetDatum(result);
+}
+
+/*
+ * Initialize batch evaluation for the supported initial clauses. The scan
+ * expression context and slot must already be initialized. On success,
+ * node->ps.qual is initialized to evaluate the complete clause list.
+ */
+ExecScanBatchState *
+ExecInitScanBatch(ScanState *node, List *clauses)
+{
+	ExecScanBatchState *state;
+	ExprState  *qual;
+	int			nquals = 0;
+	int			i = 0;
+
+	Assert(node->ps.qual == NULL);
+	Assert(node->ps.ps_ExprContext != NULL);
+	if (!slot_supports_batch(node->ss_ScanTupleSlot))
+		return NULL;
+
+	/*
+	 * Count the supported prefix before allocating an exact-sized state
+	 * array.
+	 */
+	foreach_ptr(Node, clause, clauses)
+	{
+		OpExpr	   *op;
+		uint8		var_argno;
+
+		op = match_batch_qual(clause, &var_argno);
+		if (op == NULL || !is_batch_qual_function(op->opfuncid))
+			break;
+		nquals++;
+	}
+	if (nquals == 0)
+		return NULL;
+
+	state = palloc0_object(ExecScanBatchState);
+	state->quals = palloc0_array(struct ExecScanBatchQual, nquals);
+
+	/*
+	 * Do not make this a child of ecxt_per_tuple_memory: resetting the
+	 * expression context would delete the child and leave this pointer
+	 * dangling.
+	 */
+	state->operator_context =
+		AllocSetContextCreate(node->ps.ps_ExprContext->ecxt_per_query_memory,
+							  "ExecScanBatch operator",
+							  ALLOCSET_START_SMALL_SIZES);
+	state->prefix_qual =
+		ExecInitQual(list_copy_head(clauses, nquals), &node->ps);
+	state->rest_qual =
+		ExecInitQual(list_copy_tail(clauses, nquals), &node->ps);
+	state->nquals = nquals;
+	foreach_ptr(Node, clause, clauses)
+	{
+		OpExpr	   *op;
+		uint8		var_argno;
+
+		if (i >= nquals)
+			break;
+		op = match_batch_qual(clause, &var_argno);
+		Assert(op != NULL);
+		init_batch_qual(&state->quals[i++], op, var_argno, &node->ps);
+	}
+
+	/* Use the same scalar states when the complete qual must be evaluated. */
+	qual = makeNode(ExprState);
+	qual->flags = EEO_FLAG_IS_QUAL;
+	qual->expr = (Expr *) clauses;
+	qual->evalfunc = ExecScanBatchQual;
+	qual->evalfunc_private = state;
+	qual->parent = &node->ps;
+	node->ps.qual = qual;
+
+	return state;
+}
+
+void
+ExecResetScanBatch(ExecScanBatchState *state)
+{
+	if (state == NULL)
+		return;
+
+	state->batch = NULL;
+	state->nrows = 0;
+}
+
+static pg_always_inline bool
+eval_batch_qual(struct ExecScanBatchQual *qual, Datum value,
+				bool track_function)
+{
+	FunctionCallInfo fcinfo = qual->fcinfo;
+	PgStat_FunctionCallUsage fcusage;
+	Datum		result;
+
+	fcinfo->args[qual->var_argno].value = value;
+	fcinfo->args[qual->var_argno].isnull = false;
+	Assert(!fcinfo->args[1 - qual->var_argno].isnull);
+	if (track_function)
+		pgstat_init_function_usage(fcinfo, &fcusage);
+	fcinfo->isnull = false;
+	result = FunctionCallInvoke(fcinfo);
+	if (track_function)
+		pgstat_end_function_usage(&fcusage, true);
+
+	return !fcinfo->isnull && DatumGetBool(result);
+}
+
+void
+ExecScanBatchPrepare(ExecScanBatchState *state, TupleTableSlot *slot,
+					 ScanDirection direction, ExprContext *econtext)
+{
+	const TupleTableSlotBatch *batch = slot_getbatch(slot);
+	MemoryContext oldcontext;
+	Datum		values[TUPLE_BATCH_MASK_BITS];
+	TupleBatchMask mask;
+	int			first;
+	int			nrows;
+
+	Assert(ScanDirectionIsValid(direction));
+	ExecResetScanBatch(state);
+	if (batch == NULL || ScanDirectionIsNoMovement(direction))
+		return;
+	Assert(state->nquals > 0);
+
+	if (ScanDirectionIsForward(direction))
+	{
+		first = batch->current + 1;
+		nrows = Min((int) TUPLE_BATCH_MASK_BITS,
+					batch->ntuples - first);
+	}
+	else
+	{
+		nrows = Min((int) TUPLE_BATCH_MASK_BITS, batch->current);
+		first = batch->current - nrows;
+	}
+	if (nrows == 0)
+		return;
+	mask = PG_UINT64_MAX >> (TUPLE_BATCH_MASK_BITS - nrows);
+
+	oldcontext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
+	for (int q = 0; q < state->nquals && mask != 0; q++)
+	{
+		struct ExecScanBatchQual *qual = &state->quals[q];
+		FunctionCallInfo fcinfo = qual->fcinfo;
+		NullableDatum *other_arg = &fcinfo->args[1 - qual->var_argno];
+
+		if (qual->param_expr != NULL)
+			other_arg->value = ExecEvalExpr(qual->param_expr, econtext,
+											&other_arg->isnull);
+		else
+			*other_arg = qual->constant;
+
+		if (other_arg->isnull)
+			mask = 0;
+		else
+		{
+			MemoryContext oldoperatorcontext;
+			bool		track_function;
+			int			position;
+
+			/*
+			 * Fetch the first qual's attribute, and fetch again only when it
+			 * changes. Adjacent quals on the same attribute reuse its
+			 * values.
+			 */
+			if (q == 0 || qual->attnum != state->quals[q - 1].attnum)
+				mask &= ~batch->getattrs(slot, qual->attnum, first, nrows,
+										 mask, values);
+
+			track_function =
+				pgstat_track_functions > fcinfo->flinfo->fn_stats;
+			position = ScanDirectionIsForward(direction) ? 0 : nrows - 1;
+			oldoperatorcontext =
+				MemoryContextSwitchTo(state->operator_context);
+			for (int i = 0; i < nrows && mask != 0;
+				 i++, position += direction)
+			{
+				TupleBatchMask bit = UINT64CONST(1) << position;
+
+				if ((mask & bit) == 0)
+					continue;
+				if (!eval_batch_qual(qual, values[position], track_function))
+					mask &= ~bit;
+			}
+			MemoryContextSwitchTo(oldoperatorcontext);
+		}
+
+		if (qual->param_expr != NULL)
+		{
+			other_arg->value = (Datum) 0;
+			other_arg->isnull = true;
+		}
+
+		/*
+		 * Functions run in operator_context and may leave temporary
+		 * allocations there. Reset once after the dense loop to amortize the
+		 * reset overhead, while limiting retained memory to one batch for a
+		 * single qual.
+		 */
+		MemoryContextReset(state->operator_context);
+	}
+	MemoryContextSwitchTo(oldcontext);
+
+	state->batch = batch;
+	state->generation = batch->generation;
+	state->mask = mask;
+	state->first = first;
+	state->nrows = nrows;
+}
 
 /* ----------------------------------------------------------------
  *		ExecScan
@@ -64,6 +391,38 @@ ExecScan(ScanState *node,
 							projInfo);
 }
 
+/*
+ * ExecScanBatch
+ *		ExecScan variant that evaluates a supported initial run of scan
+ *		qualifications using tuple batches.
+ *
+ * A NULL state means that batch evaluation was unavailable, so use the
+ * ordinary path. Batch state cannot be used for EvalPlanQual tuples either.
+ * Callers must use ExecResetScanBatch() when rescanning the node.
+ *
+ * This entry point is intended for scan nodes implemented by extensions.
+ * Built-in nodes may call ExecScanBatchExtended() directly so the compiler
+ * can inline their access method and specialize projection handling.
+ */
+TupleTableSlot *
+ExecScanBatch(ScanState *node,
+			  ExecScanAccessMtd accessMtd,
+			  ExecScanRecheckMtd recheckMtd,
+			  ExecScanBatchState *state)
+{
+	if (state == NULL)
+		return ExecScan(node, accessMtd, recheckMtd);
+
+	if (node->ps.state->es_epq_active != NULL)
+	{
+		ExecResetScanBatch(state);
+		return ExecScan(node, accessMtd, recheckMtd);
+	}
+
+	return ExecScanBatchExtended(node, accessMtd, recheckMtd, state,
+								 node->ps.ps_ProjInfo);
+}
+
 /*
  * ExecAssignScanProjectionInfo
  *		Set up projection info for a scan node, if necessary.
diff --git a/src/backend/executor/execScanBatch.h b/src/backend/executor/execScanBatch.h
new file mode 100644
index 00000000000..8f85b6909aa
--- /dev/null
+++ b/src/backend/executor/execScanBatch.h
@@ -0,0 +1,140 @@
+/*-------------------------------------------------------------------------
+ *
+ * execScanBatch.h
+ *	  Internal support for evaluating scan qualifications in tuple batches.
+ *
+ * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/executor/execScanBatch.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef EXECSCANBATCH_H
+#define EXECSCANBATCH_H
+
+#include "executor/execScan.h"
+#include "executor/tuptable.h"
+
+struct ExecScanBatchQual
+{
+	FmgrInfo	func;
+	FunctionCallInfo fcinfo;
+	ExprState  *param_expr;
+	NullableDatum constant;
+	AttrNumber	attnum;
+	uint8		var_argno;
+};
+
+struct ExecScanBatchState
+{
+	struct ExecScanBatchQual *quals;
+	ExprState  *prefix_qual;
+	ExprState  *rest_qual;
+	MemoryContext operator_context;
+	const TupleTableSlotBatch *batch;
+	uint64		generation;
+	TupleBatchMask mask;		/* bit i is for batch index first + i */
+	int			first;
+	int			nrows;
+	int			nquals;
+};
+
+extern void ExecScanBatchPrepare(ExecScanBatchState *state,
+								 TupleTableSlot *slot,
+								 ScanDirection direction,
+								 ExprContext *econtext);
+
+static pg_always_inline bool
+ExecScanBatchReuse(ExecScanBatchState *state, TupleTableSlot *slot,
+				   ScanDirection direction, bool *qual_result)
+{
+	const TupleTableSlotBatch *batch;
+	int			position;
+
+	Assert(ScanDirectionIsValid(direction));
+	if (state->nrows == 0)
+		return false;
+
+	batch = slot_getbatch(slot);
+	if (ScanDirectionIsNoMovement(direction) ||
+		batch == NULL || state->batch != batch ||
+		state->generation != batch->generation)
+		goto invalid;
+
+	Assert(state->nrows <= (int) TUPLE_BATCH_MASK_BITS);
+	Assert(state->first >= 0);
+	Assert(state->nrows <= batch->ntuples);
+	Assert(state->first <= batch->ntuples - state->nrows);
+
+	position = batch->current - state->first;
+	if (position < 0 || position >= state->nrows)
+		goto invalid;
+
+	*qual_result = (state->mask & (UINT64CONST(1) << position)) != 0;
+	if (position == (ScanDirectionIsForward(direction) ?
+					 state->nrows - 1 : 0))
+		ExecResetScanBatch(state);
+	return true;
+
+invalid:
+	ExecResetScanBatch(state);
+	return false;
+}
+
+/*
+ * Keep the scan loop and expression-context cleanup below in sync with
+ * ExecScanExtended(). Differences should remain limited to batch
+ * qualification reuse and preparation.
+ */
+static pg_always_inline TupleTableSlot *
+ExecScanBatchExtended(ScanState *node,
+					  ExecScanAccessMtd accessMtd,
+					  ExecScanRecheckMtd recheckMtd,
+					  ExecScanBatchState *state,
+					  ProjectionInfo *projInfo)
+{
+	PlanState  *pstate = &node->ps;
+	ExprContext *econtext = pstate->ps_ExprContext;
+	ScanDirection direction = pstate->state->es_direction;
+	TupleTableSlot *result_slot;
+	TupleTableSlot *slot;
+	bool		qual_result;
+
+	Assert(pstate->state->es_epq_active == NULL);
+	Assert(projInfo == pstate->ps_ProjInfo);
+	Assert(state != NULL);
+	pg_assume(state->prefix_qual != NULL);
+
+	ResetExprContext(econtext);
+	for (;;)
+	{
+		slot = ExecScanFetch(node, NULL, accessMtd, recheckMtd);
+		if (TupIsNull(slot))
+		{
+			if (projInfo != NULL)
+				return ExecClearTuple(projInfo->pi_state.resultslot);
+			return slot;
+		}
+
+		econtext->ecxt_scantuple = slot;
+		if (!ExecScanBatchReuse(state, slot, direction, &qual_result))
+			qual_result = ExecQual(state->prefix_qual, econtext);
+		if (qual_result)
+			qual_result = ExecQual(state->rest_qual, econtext);
+
+		result_slot = slot;
+		if (qual_result && projInfo != NULL)
+			result_slot = ExecProject(projInfo);
+
+		if (state->nrows == 0)
+			ExecScanBatchPrepare(state, slot, direction, econtext);
+		if (qual_result)
+			return result_slot;
+
+		InstrCountFiltered1(node, 1);
+		ResetExprContext(econtext);
+	}
+}
+
+#endif							/* EXECSCANBATCH_H */
diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c
index a40e1b9e51d..a1d92246044 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -30,6 +30,7 @@
 #include "access/relscan.h"
 #include "access/tableam.h"
 #include "catalog/catalog.h"
+#include "execScanBatch.h"
 #include "executor/execParallel.h"
 #include "executor/execScan.h"
 #include "executor/executor.h"
@@ -158,6 +159,38 @@ ExecSeqScanWithQual(PlanState *pstate)
 							NULL);
 }
 
+/*
+ * Variant of ExecSeqScan() for initial quals that can use tuple batches.
+ */
+static TupleTableSlot *
+ExecSeqScanWithBatchQual(PlanState *pstate)
+{
+	SeqScanState *node = castNode(SeqScanState, pstate);
+
+	Assert(pstate->ps_ProjInfo == NULL);
+
+	return ExecScanBatchExtended(&node->ss,
+								 (ExecScanAccessMtd) SeqNext,
+								 (ExecScanRecheckMtd) SeqRecheck,
+								 node->batch_state, NULL);
+}
+
+/*
+ * Variant of ExecSeqScan() for initial batch quals and projection.
+ */
+static TupleTableSlot *
+ExecSeqScanWithBatchQualProject(PlanState *pstate)
+{
+	SeqScanState *node = castNode(SeqScanState, pstate);
+
+	pg_assume(pstate->ps_ProjInfo != NULL);
+
+	return ExecScanBatchExtended(&node->ss,
+								 (ExecScanAccessMtd) SeqNext,
+								 (ExecScanRecheckMtd) SeqRecheck,
+								 node->batch_state, pstate->ps_ProjInfo);
+}
+
 /*
  * Variant of ExecSeqScan() but when projection is required.
  */
@@ -279,8 +312,12 @@ ExecInitSeqScan(SeqScan *node, EState *estate, int eflags)
 	/*
 	 * initialize child expressions
 	 */
-	scanstate->ss.ps.qual =
-		ExecInitQual(node->scan.plan.qual, (PlanState *) scanstate);
+	if (estate->es_epq_active == NULL)
+		scanstate->batch_state =
+			ExecInitScanBatch(&scanstate->ss, node->scan.plan.qual);
+	if (scanstate->batch_state == NULL)
+		scanstate->ss.ps.qual =
+			ExecInitQual(node->scan.plan.qual, (PlanState *) scanstate);
 
 	/*
 	 * When EvalPlanQual() is not in use, assign ExecProcNode for this node
@@ -289,6 +326,13 @@ ExecInitSeqScan(SeqScan *node, EState *estate, int eflags)
 	 */
 	if (scanstate->ss.ps.state->es_epq_active != NULL)
 		scanstate->ss.ps.ExecProcNode = ExecSeqScanEPQ;
+	else if (scanstate->batch_state != NULL)
+	{
+		if (scanstate->ss.ps.ps_ProjInfo == NULL)
+			scanstate->ss.ps.ExecProcNode = ExecSeqScanWithBatchQual;
+		else
+			scanstate->ss.ps.ExecProcNode = ExecSeqScanWithBatchQualProject;
+	}
 	else if (scanstate->ss.ps.qual == NULL)
 	{
 		if (scanstate->ss.ps.ps_ProjInfo == NULL)
@@ -362,6 +406,7 @@ ExecReScanSeqScan(SeqScanState *node)
 {
 	TableScanDesc scan;
 
+	ExecResetScanBatch(node->batch_state);
 	scan = node->ss.ss_currentScanDesc;
 
 	if (scan != NULL)
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 1798e6027d4..3556feca7dd 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -589,6 +589,19 @@ typedef bool (*ExecScanRecheckMtd) (ScanState *node, TupleTableSlot *slot);
 
 extern TupleTableSlot *ExecScan(ScanState *node, ExecScanAccessMtd accessMtd,
 								ExecScanRecheckMtd recheckMtd);
+/*
+ * Batch scan support for built-in and extension scan nodes.
+ *
+ * Initialize after the expression context and scan slot. On success this
+ * also initializes ps.qual; otherwise the caller must initialize it normally.
+ * ExecScanBatch() is the public execution entry point. Reset before rescan.
+ */
+extern ExecScanBatchState *ExecInitScanBatch(ScanState *node, List *clauses);
+extern TupleTableSlot *ExecScanBatch(ScanState *node,
+									 ExecScanAccessMtd accessMtd,
+									 ExecScanRecheckMtd recheckMtd,
+									 ExecScanBatchState *state);
+extern void ExecResetScanBatch(ExecScanBatchState *state);
 extern void ExecAssignScanProjectionInfo(ScanState *node);
 extern void ExecAssignScanProjectionInfoWithVarno(ScanState *node, int varno);
 extern void ExecScanReScan(ScanState *node);
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index e95ac3eda35..5970c4ab18c 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -47,6 +47,7 @@
  */
 typedef struct BufferUsage BufferUsage;
 typedef struct ExecRowMark ExecRowMark;
+typedef struct ExecScanBatchState ExecScanBatchState;
 typedef struct ExprState ExprState;
 typedef struct ExprContext ExprContext;
 typedef struct HTAB HTAB;
@@ -1673,6 +1674,7 @@ typedef struct SeqScanState
 	ScanState	ss;				/* its first field is NodeTag */
 	Size		pscan_len;		/* size of parallel heap scan descriptor */
 	struct SharedSeqScanInstrumentation *sinstrument;
+	ExecScanBatchState *batch_state;
 } SeqScanState;
 
 /* ----------------
diff --git a/src/test/regress/expected/scan_batch.out b/src/test/regress/expected/scan_batch.out
new file mode 100644
index 00000000000..204dedb3485
--- /dev/null
+++ b/src/test/regress/expected/scan_batch.out
@@ -0,0 +1,241 @@
+--
+-- Tests for evaluating scan qualifications in tuple batches
+--
+CREATE TABLE scan_batch_test (id int, a int, t text)
+  WITH (autovacuum_enabled = false);
+INSERT INTO scan_batch_test
+SELECT g,
+	CASE WHEN g IN (1, 2) THEN NULL ELSE g END,
+	CASE WHEN g IN (3, 100) THEN NULL ELSE 'value-' || g END
+FROM generate_series(1, 200) AS g;
+-- Multiple batch qualifications, both argument orders, and NULL attributes.
+SELECT count(*) FROM scan_batch_test
+WHERE 0 < a AND a < 201 AND t <> 'missing';
+ count 
+-------
+   196
+(1 row)
+
+-- An unsupported first qualification must leave the whole expression scalar.
+SELECT count(*) FROM scan_batch_test
+WHERE length(t) > 0 AND a > 50;
+ count 
+-------
+   149
+(1 row)
+
+-- A supported prefix followed by a scalar remainder.
+EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF)
+SELECT count(*) FROM scan_batch_test
+WHERE a > 50 AND length(t) = 8;
+                          QUERY PLAN                           
+---------------------------------------------------------------
+ Aggregate (actual rows=1.00 loops=1)
+   ->  Seq Scan on scan_batch_test (actual rows=49.00 loops=1)
+         Filter: ((a > 50) AND (length(t) = 8))
+         Rows Removed by Filter: 151
+(4 rows)
+
+-- PARAM_EXTERN, including a pass-by-reference value and NULL parameters.
+SET plan_cache_mode = force_generic_plan;
+PREPARE scan_batch_ext(int, text) AS
+SELECT count(*) FROM scan_batch_test
+WHERE $1 < a AND t <> $2;
+EXECUTE scan_batch_ext(50, 'missing');
+ count 
+-------
+   149
+(1 row)
+
+EXECUTE scan_batch_ext(NULL, 'missing');
+ count 
+-------
+     0
+(1 row)
+
+EXECUTE scan_batch_ext(50, NULL);
+ count 
+-------
+     0
+(1 row)
+
+DEALLOCATE scan_batch_ext;
+RESET plan_cache_mode;
+-- PARAM_EXEC values and rescans of the inner sequential scan.
+SELECT p, q,
+	(SELECT count(*) FROM scan_batch_test AS b
+	 WHERE o.p < b.a AND b.t <> o.q) AS matches
+FROM (VALUES
+	(1, 0, 'missing'::text),
+	(2, 100, 'value-150'),
+	(3, NULL, 'missing'),
+	(4, 0, NULL)
+) AS o(ord, p, q)
+ORDER BY ord;
+  p  |     q     | matches 
+-----+-----------+---------
+   0 | missing   |     196
+ 100 | value-150 |      99
+     | missing   |       0
+   0 |           |       0
+(4 rows)
+
+-- A stateful subplan in the scalar remainder must follow parameter changes.
+CREATE TABLE scan_batch_lookup (k int, v int);
+INSERT INTO scan_batch_lookup
+SELECT -1, g FROM generate_series(1, 200) AS g
+UNION ALL
+SELECT -2, g FROM generate_series(1, 100) AS g;
+EXPLAIN (COSTS OFF)
+SELECT o.p,
+	(SELECT count(*) FROM scan_batch_test AS b
+	 WHERE b.id > o.p
+	   AND b.id NOT IN
+		   (SELECT l.v FROM scan_batch_lookup AS l WHERE l.k = o.p)) AS matches
+FROM (VALUES (1, -1), (2, -2)) AS o(ord, p)
+ORDER BY o.ord;
+                                                  QUERY PLAN                                                  
+--------------------------------------------------------------------------------------------------------------
+ Sort
+   Sort Key: "*VALUES*".column1
+   ->  Values Scan on "*VALUES*"
+         SubPlan expr_1
+           ->  Aggregate
+                 ->  Seq Scan on scan_batch_test b
+                       Filter: ((id > "*VALUES*".column2) AND (NOT (ANY (id = (hashed SubPlan any_1).col1))))
+                       SubPlan any_1
+                         ->  Seq Scan on scan_batch_lookup l
+                               Filter: (k = "*VALUES*".column2)
+(10 rows)
+
+SELECT o.p,
+	(SELECT count(*) FROM scan_batch_test AS b
+	 WHERE b.id > o.p
+	   AND b.id NOT IN
+		   (SELECT l.v FROM scan_batch_lookup AS l WHERE l.k = o.p)) AS matches
+FROM (VALUES (1, -1), (2, -2)) AS o(ord, p)
+ORDER BY o.ord;
+ p  | matches 
+----+---------
+ -1 |       0
+ -2 |     100
+(2 rows)
+
+DROP TABLE scan_batch_lookup;
+-- Cross batch windows and change direction while projecting values.
+BEGIN;
+DECLARE scan_batch_cursor SCROLL CURSOR FOR
+SELECT id + 1 AS projected, upper(t) AS projected_text
+FROM scan_batch_test
+WHERE id > 0 AND length(COALESCE(t, '')) >= 0;
+MOVE FORWARD 130 FROM scan_batch_cursor;
+FETCH BACKWARD 3 FROM scan_batch_cursor;
+ projected | projected_text 
+-----------+----------------
+       130 | VALUE-129
+       129 | VALUE-128
+       128 | VALUE-127
+(3 rows)
+
+FETCH FORWARD 3 FROM scan_batch_cursor;
+ projected | projected_text 
+-----------+----------------
+       129 | VALUE-128
+       130 | VALUE-129
+       131 | VALUE-130
+(3 rows)
+
+COMMIT;
+-- System columns must follow the current tuple in a batch.
+SELECT count(*)
+FROM scan_batch_test AS s
+WHERE s.id > 0
+	AND NOT EXISTS (SELECT 1 FROM scan_batch_test AS tid_check
+					WHERE tid_check.ctid = s.ctid AND tid_check.id = s.id);
+ count 
+-------
+     0
+(1 row)
+
+-- The batch scan must return the projection's empty result slot.
+SELECT id + 1 AS projected, upper(t) AS projected_text
+FROM scan_batch_test
+WHERE id > 200 AND length(COALESCE(t, '')) >= 0;
+ projected | projected_text 
+-----------+----------------
+(0 rows)
+
+-- Calls made while preparing a batch must honor track_functions.
+CREATE FUNCTION scan_batch_lt(int, int) RETURNS boolean
+LANGUAGE plpgsql IMMUTABLE STRICT LEAKPROOF
+AS $$BEGIN RETURN $1 < $2; END$$;
+CREATE OPERATOR #<# (
+	FUNCTION = scan_batch_lt,
+	LEFTARG = int,
+	RIGHTARG = int
+);
+SELECT 'scan_batch_lt(integer,integer)'::regprocedure::oid AS batch_func_oid \gset
+SET track_functions = 'all';
+SELECT count(*) FROM scan_batch_test WHERE a #<# 201;
+ count 
+-------
+   198
+(1 row)
+
+SELECT pg_stat_force_next_flush() \gset
+SELECT calls FROM pg_stat_user_functions WHERE funcid = :batch_func_oid;
+ calls 
+-------
+   198
+(1 row)
+
+-- Updating returned rows must not cause a batch to be evaluated repeatedly.
+SELECT pg_stat_reset_single_function_counters(:batch_func_oid) AS reset \gset
+UPDATE scan_batch_test SET t = t WHERE a #<# 201;
+SELECT pg_stat_force_next_flush() \gset
+SELECT calls FROM pg_stat_user_functions WHERE funcid = :batch_func_oid;
+ calls 
+-------
+   198
+(1 row)
+
+-- Relations that can receive in-place updates must use scalar evaluation.
+SELECT pg_stat_reset_single_function_counters(:batch_func_oid) AS reset \gset
+SELECT count(*) FROM
+	(SELECT FROM pg_class
+	 WHERE relpages #<# 2147483647
+	 LIMIT 1) AS core_catalog;
+ count 
+-------
+     1
+(1 row)
+
+SELECT pg_stat_force_next_flush() \gset
+SELECT calls FROM pg_stat_user_functions WHERE funcid = :batch_func_oid;
+ calls 
+-------
+     1
+(1 row)
+
+ALTER TABLE scan_batch_test SET (user_catalog_table = true);
+SELECT pg_stat_reset_single_function_counters(:batch_func_oid) AS reset \gset
+SELECT count(*) FROM
+	(SELECT FROM scan_batch_test
+	 WHERE id #<# 201
+	 LIMIT 1) AS user_catalog;
+ count 
+-------
+     1
+(1 row)
+
+SELECT pg_stat_force_next_flush() \gset
+SELECT calls FROM pg_stat_user_functions WHERE funcid = :batch_func_oid;
+ calls 
+-------
+     1
+(1 row)
+
+RESET track_functions;
+DROP OPERATOR #<# (int, int);
+DROP FUNCTION scan_batch_lt(int, int);
+DROP TABLE scan_batch_test;
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 8fa0a6c47fb..fed9a941ae9 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -115,7 +115,7 @@ test: json jsonb json_encoding jsonpath jsonpath_encoding jsonb_jsonpath sqljson
 # NB: temp.sql does reconnects which transiently uses 2 connections,
 # so keep this parallel group to at most 19 tests
 # ----------
-test: plancache limit plpgsql copy2 temp domain rangefuncs prepare conversion truncate alter_table sequence polymorphism rowtypes returning largeobject with xml
+test: plancache scan_batch limit plpgsql copy2 temp domain rangefuncs prepare conversion truncate alter_table sequence polymorphism rowtypes returning largeobject with xml
 
 # ----------
 # Another group of parallel tests
diff --git a/src/test/regress/sql/scan_batch.sql b/src/test/regress/sql/scan_batch.sql
new file mode 100644
index 00000000000..44a2de64659
--- /dev/null
+++ b/src/test/regress/sql/scan_batch.sql
@@ -0,0 +1,147 @@
+--
+-- Tests for evaluating scan qualifications in tuple batches
+--
+
+CREATE TABLE scan_batch_test (id int, a int, t text)
+  WITH (autovacuum_enabled = false);
+
+INSERT INTO scan_batch_test
+SELECT g,
+	CASE WHEN g IN (1, 2) THEN NULL ELSE g END,
+	CASE WHEN g IN (3, 100) THEN NULL ELSE 'value-' || g END
+FROM generate_series(1, 200) AS g;
+
+-- Multiple batch qualifications, both argument orders, and NULL attributes.
+SELECT count(*) FROM scan_batch_test
+WHERE 0 < a AND a < 201 AND t <> 'missing';
+
+-- An unsupported first qualification must leave the whole expression scalar.
+SELECT count(*) FROM scan_batch_test
+WHERE length(t) > 0 AND a > 50;
+
+-- A supported prefix followed by a scalar remainder.
+EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF)
+SELECT count(*) FROM scan_batch_test
+WHERE a > 50 AND length(t) = 8;
+
+-- PARAM_EXTERN, including a pass-by-reference value and NULL parameters.
+SET plan_cache_mode = force_generic_plan;
+PREPARE scan_batch_ext(int, text) AS
+SELECT count(*) FROM scan_batch_test
+WHERE $1 < a AND t <> $2;
+
+EXECUTE scan_batch_ext(50, 'missing');
+EXECUTE scan_batch_ext(NULL, 'missing');
+EXECUTE scan_batch_ext(50, NULL);
+
+DEALLOCATE scan_batch_ext;
+RESET plan_cache_mode;
+
+-- PARAM_EXEC values and rescans of the inner sequential scan.
+SELECT p, q,
+	(SELECT count(*) FROM scan_batch_test AS b
+	 WHERE o.p < b.a AND b.t <> o.q) AS matches
+FROM (VALUES
+	(1, 0, 'missing'::text),
+	(2, 100, 'value-150'),
+	(3, NULL, 'missing'),
+	(4, 0, NULL)
+) AS o(ord, p, q)
+ORDER BY ord;
+
+-- A stateful subplan in the scalar remainder must follow parameter changes.
+CREATE TABLE scan_batch_lookup (k int, v int);
+INSERT INTO scan_batch_lookup
+SELECT -1, g FROM generate_series(1, 200) AS g
+UNION ALL
+SELECT -2, g FROM generate_series(1, 100) AS g;
+
+EXPLAIN (COSTS OFF)
+SELECT o.p,
+	(SELECT count(*) FROM scan_batch_test AS b
+	 WHERE b.id > o.p
+	   AND b.id NOT IN
+		   (SELECT l.v FROM scan_batch_lookup AS l WHERE l.k = o.p)) AS matches
+FROM (VALUES (1, -1), (2, -2)) AS o(ord, p)
+ORDER BY o.ord;
+
+SELECT o.p,
+	(SELECT count(*) FROM scan_batch_test AS b
+	 WHERE b.id > o.p
+	   AND b.id NOT IN
+		   (SELECT l.v FROM scan_batch_lookup AS l WHERE l.k = o.p)) AS matches
+FROM (VALUES (1, -1), (2, -2)) AS o(ord, p)
+ORDER BY o.ord;
+
+DROP TABLE scan_batch_lookup;
+
+-- Cross batch windows and change direction while projecting values.
+BEGIN;
+DECLARE scan_batch_cursor SCROLL CURSOR FOR
+SELECT id + 1 AS projected, upper(t) AS projected_text
+FROM scan_batch_test
+WHERE id > 0 AND length(COALESCE(t, '')) >= 0;
+
+MOVE FORWARD 130 FROM scan_batch_cursor;
+FETCH BACKWARD 3 FROM scan_batch_cursor;
+FETCH FORWARD 3 FROM scan_batch_cursor;
+COMMIT;
+
+-- System columns must follow the current tuple in a batch.
+SELECT count(*)
+FROM scan_batch_test AS s
+WHERE s.id > 0
+	AND NOT EXISTS (SELECT 1 FROM scan_batch_test AS tid_check
+					WHERE tid_check.ctid = s.ctid AND tid_check.id = s.id);
+
+-- The batch scan must return the projection's empty result slot.
+SELECT id + 1 AS projected, upper(t) AS projected_text
+FROM scan_batch_test
+WHERE id > 200 AND length(COALESCE(t, '')) >= 0;
+
+-- Calls made while preparing a batch must honor track_functions.
+CREATE FUNCTION scan_batch_lt(int, int) RETURNS boolean
+LANGUAGE plpgsql IMMUTABLE STRICT LEAKPROOF
+AS $$BEGIN RETURN $1 < $2; END$$;
+
+CREATE OPERATOR #<# (
+	FUNCTION = scan_batch_lt,
+	LEFTARG = int,
+	RIGHTARG = int
+);
+
+SELECT 'scan_batch_lt(integer,integer)'::regprocedure::oid AS batch_func_oid \gset
+SET track_functions = 'all';
+
+SELECT count(*) FROM scan_batch_test WHERE a #<# 201;
+SELECT pg_stat_force_next_flush() \gset
+SELECT calls FROM pg_stat_user_functions WHERE funcid = :batch_func_oid;
+
+-- Updating returned rows must not cause a batch to be evaluated repeatedly.
+SELECT pg_stat_reset_single_function_counters(:batch_func_oid) AS reset \gset
+UPDATE scan_batch_test SET t = t WHERE a #<# 201;
+SELECT pg_stat_force_next_flush() \gset
+SELECT calls FROM pg_stat_user_functions WHERE funcid = :batch_func_oid;
+
+-- Relations that can receive in-place updates must use scalar evaluation.
+SELECT pg_stat_reset_single_function_counters(:batch_func_oid) AS reset \gset
+SELECT count(*) FROM
+	(SELECT FROM pg_class
+	 WHERE relpages #<# 2147483647
+	 LIMIT 1) AS core_catalog;
+SELECT pg_stat_force_next_flush() \gset
+SELECT calls FROM pg_stat_user_functions WHERE funcid = :batch_func_oid;
+
+ALTER TABLE scan_batch_test SET (user_catalog_table = true);
+SELECT pg_stat_reset_single_function_counters(:batch_func_oid) AS reset \gset
+SELECT count(*) FROM
+	(SELECT FROM scan_batch_test
+	 WHERE id #<# 201
+	 LIMIT 1) AS user_catalog;
+SELECT pg_stat_force_next_flush() \gset
+SELECT calls FROM pg_stat_user_functions WHERE funcid = :batch_func_oid;
+
+RESET track_functions;
+DROP OPERATOR #<# (int, int);
+DROP FUNCTION scan_batch_lt(int, int);
+DROP TABLE scan_batch_test;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index c42f4a3453c..0436eda73df 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -787,6 +787,7 @@ ExecPhraseData
 ExecProcNodeMtd
 ExecRowMark
 ExecScanAccessMtd
+ExecScanBatchState
 ExecScanRecheckMtd
 ExecStatus
 ExecStatusType
-- 
2.54.0

