From 4700a99409674ef50f87f3e3e415c271cebdf537 Mon Sep 17 00:00:00 2001
Message-ID: <4700a99409674ef50f87f3e3e415c271cebdf537.1784099400.git.darthunix@gmail.com>
In-Reply-To: <4e0cffbfd728213677fa9f15800f4ac6a4607503.1784099400.git.darthunix@gmail.com>
References: <4e0cffbfd728213677fa9f15800f4ac6a4607503.1784099400.git.darthunix@gmail.com>
From: Denis Smirnov <darthunix@gmail.com>
Date: Tue, 14 Jul 2026 15:39:41 +0800
Subject: [PATCH v2 2/2] Evaluate SeqScan quals in heap page batches

Heap scans can now expose the visible tuples remaining on the current
page through HeapPageBatch.  Use this information in SeqScan to evaluate
suitable scan quals for several tuples at once.

For up to 64 tuples, first deform the referenced attribute into a Datum
array, then invoke the existing operator function in a tight loop.  Store
the results in a 64-bit mask.  With multiple quals, successively narrow
the mask and avoid deforming attributes or calling operators for tuples
already rejected by an earlier qual.  This reduces per-tuple executor
overhead without adding a new expression or function API.

Batch only an initial sequence of quals in the order chosen by the
planner.  Do not reorder them, both to preserve the required evaluation
order and to avoid losing any favorable tuple deformation order already
present in the plan.  Evaluate the first unsupported qual and everything
following it one tuple at a time.  When no HeapPageBatch is available,
evaluate the complete filter using the normal scalar path.

Recognize operators of the form Var op Const, Const op Var, Var op Param,
or Param op Var.  Require the operator function to return bool, not
return a set, and be marked IMMUTABLE, STRICT, and LEAKPROOF.  Batch
evaluation may call an operator for tuples that are never requested, for
example because of LIMIT or a later qual, so the operator must have no
observable side effects and its catalog declarations must be correct.
Preserve collation, NULL behavior, parameter evaluation, function usage
accounting, and both scan directions.

Apply ordinary projection with ExecProject() only after a tuple passes
all quals.  Projection itself remains row-at-a-time and does not yet
benefit from page batching.

Discussion:
https://postgr.es/m/3D8D1BA5-B8A6-400A-B605-42EEE37890B4%40gmail.com
---
 src/backend/executor/nodeSeqscan.c      | 562 +++++++++++++++++++++++-
 src/include/nodes/execnodes.h           |   1 +
 src/test/regress/expected/plancache.out | 129 ++++++
 src/test/regress/sql/plancache.sql      | 104 +++++
 4 files changed, 792 insertions(+), 4 deletions(-)

diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c
index b8c528ca089..0c0a67803ec 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -27,16 +27,417 @@
  */
 #include "postgres.h"
 
+#include "access/htup_details.h"
 #include "access/relscan.h"
 #include "access/tableam.h"
+#include "catalog/pg_proc.h"
+#include "catalog/pg_type.h"
 #include "executor/execParallel.h"
 #include "executor/execScan.h"
 #include "executor/executor.h"
 #include "executor/nodeSeqscan.h"
+#include "pgstat.h"
+#include "utils/lsyscache.h"
 #include "utils/rel.h"
 
 static TupleTableSlot *SeqNext(SeqScanState *node);
 
+/* Private state for one scan qual evaluated in page batches. */
+struct SeqScanBatchQual
+{
+	FmgrInfo	func;
+	FunctionCallInfo fcinfo;
+	ExprState  *param_expr;
+	AttrNumber	attnum;
+	uint8		var_argno;
+};
+
+/* Private state for page-batch qual evaluation. */
+struct SeqScanBatchState
+{
+	struct SeqScanBatchQual *quals;
+	ExprState  *scalar_qual;
+	uint64		mask;			/* bit i is for batch index first + i */
+	BlockNumber block;
+	int			first;
+	int			nrows;
+	int			nquals;
+};
+
+static inline void
+reset_batch_qual(struct SeqScanBatchState *state)
+{
+	state->nrows = 0;
+}
+
+static BufferHeapTupleTableSlot *
+get_valid_batch_slot(TupleTableSlot *slot, BlockNumber *block)
+{
+	BufferHeapTupleTableSlot *bslot;
+	HeapPageBatch *batch;
+
+	if (!TTS_IS_BUFFERTUPLE(slot) ||
+		!ItemPointerIsValid(&slot->tts_tid))
+		return NULL;
+
+	bslot = (BufferHeapTupleTableSlot *) slot;
+	batch = &bslot->batch;
+	if (batch->offsets == NULL ||
+		!BufferIsValid(bslot->buffer) ||
+		batch->current < 0 || batch->current >= batch->ntuples)
+		return NULL;
+
+	*block = BufferGetBlockNumber(bslot->buffer);
+	if (ItemPointerGetBlockNumber(&slot->tts_tid) != *block ||
+		ItemPointerGetOffsetNumber(&slot->tts_tid) !=
+		batch->offsets[batch->current])
+		return NULL;
+
+	return bslot;
+}
+
+static pg_always_inline bool
+reuse_batch_qual(struct SeqScanBatchState *state, TupleTableSlot *slot,
+				 bool *result)
+{
+	BufferHeapTupleTableSlot *bslot;
+	BlockNumber block;
+	int			position;
+
+	if (state->nrows <= 0 || state->nrows > 64)
+		goto invalid;
+
+	bslot = get_valid_batch_slot(slot, &block);
+	if (bslot == NULL)
+		goto invalid;
+
+	position = bslot->batch.current - state->first;
+	if (block != state->block ||
+		position < 0 || position >= state->nrows)
+		goto invalid;
+
+	*result = (state->mask & (UINT64CONST(1) << position)) != 0;
+	return true;
+
+invalid:
+	reset_batch_qual(state);
+	return false;
+}
+
+static pg_always_inline bool
+eval_scalar_qual(struct SeqScanBatchState *state, TupleTableSlot *slot,
+				 ExprContext *econtext)
+{
+	econtext->ecxt_scantuple = slot;
+	return ExecQual(state->scalar_qual, econtext);
+}
+
+static pg_always_inline Datum
+heap_page_batch_getattr(Page page, const HeapPageBatch *batch, int index,
+						TupleDesc tupledesc, AttrNumber attnum, bool *isnull)
+{
+	HeapTupleData tuple;
+	ItemId		lpp;
+
+	lpp = PageGetItemId(page, batch->offsets[index]);
+	Assert(ItemIdIsNormal(lpp));
+	tuple.t_data = (HeapTupleHeader) PageGetItem(page, lpp);
+	tuple.t_len = ItemIdGetLength(lpp);
+
+	return heap_getattr(&tuple, attnum, tupledesc, isnull);
+}
+
+static pg_always_inline bool
+eval_batch_qual(struct SeqScanBatchQual *qual, Datum value, bool isnull)
+{
+	FunctionCallInfo fcinfo = qual->fcinfo;
+	PgStat_FunctionCallUsage fcusage;
+	Datum		result;
+
+	fcinfo->args[qual->var_argno].value = value;
+	fcinfo->args[qual->var_argno].isnull = isnull;
+	if (isnull)
+		return false;
+	Assert(!fcinfo->args[1 - qual->var_argno].isnull);
+
+	pgstat_init_function_usage(fcinfo, &fcusage);
+	fcinfo->isnull = false;
+	result = FunctionCallInvoke(fcinfo);
+	pgstat_end_function_usage(&fcusage, true);
+
+	return !fcinfo->isnull && DatumGetBool(result);
+}
+
+static void
+prepare_batch_qual(struct SeqScanBatchState *state, TupleTableSlot *slot,
+				   ScanDirection direction, ExprContext *econtext)
+{
+	BufferHeapTupleTableSlot *bslot;
+	HeapPageBatch *batch;
+	Datum		values[64];
+	MemoryContext oldcontext;
+	BlockNumber block;
+	Page		page;
+	uint64		mask = 0;
+	int			first;
+	int			nrows;
+	int			position;
+
+	Assert(ScanDirectionIsForward(direction) ||
+		   ScanDirectionIsBackward(direction));
+	Assert(state->nquals > 0);
+
+	reset_batch_qual(state);
+
+	bslot = get_valid_batch_slot(slot, &block);
+	if (bslot == NULL)
+		return;
+
+	batch = &bslot->batch;
+	if (ScanDirectionIsForward(direction))
+	{
+		first = batch->current + 1;
+		nrows = Min(64, batch->ntuples - first);
+	}
+	else
+	{
+		nrows = Min(64, batch->current);
+		first = batch->current - nrows;
+	}
+	if (nrows == 0)
+		return;
+
+	oldcontext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
+	page = BufferGetPage(bslot->buffer);
+	/*
+	 * Every row is a candidate for the first qual. Using the masked loops
+	 * below would require an all-ones input mask, whose per-row tests add
+	 * measurable overhead to the common single-qual case. Track nulls
+	 * separately while gathering values, then build the result mask while
+	 * evaluating them.
+	 *
+	 * Evaluate in scan direction, but index values and mask in batch order.
+	 */
+	{
+		struct SeqScanBatchQual *qual = &state->quals[0];
+		FunctionCallInfo fcinfo = qual->fcinfo;
+		NullableDatum *arg = &fcinfo->args[1 - qual->var_argno];
+		uint64		nulls = 0;
+
+		if (qual->param_expr != NULL)
+			arg->value = ExecEvalExpr(qual->param_expr, econtext,
+								  &arg->isnull);
+		if (!arg->isnull)
+		{
+			position = ScanDirectionIsForward(direction) ? 0 : nrows - 1;
+			for (int i = 0; i < nrows; i++, position += direction)
+			{
+				bool		isnull;
+				uint64		bit = UINT64CONST(1) << position;
+
+				values[position] = heap_page_batch_getattr(page, batch,
+												first + position,
+												slot->tts_tupleDescriptor,
+												qual->attnum, &isnull);
+				if (isnull)
+					nulls |= bit;
+			}
+
+			position = ScanDirectionIsForward(direction) ? 0 : nrows - 1;
+			for (int i = 0; i < nrows; i++, position += direction)
+			{
+				uint64		bit = UINT64CONST(1) << position;
+
+				if (eval_batch_qual(qual, values[position],
+									(nulls & bit) != 0))
+					mask |= bit;
+			}
+		}
+
+		if (qual->param_expr != NULL)
+		{
+			arg->value = (Datum) 0;
+			arg->isnull = true;
+		}
+	}
+
+	/*
+	 * For later quals, mask contains only rows accepted by all preceding
+	 * quals. Testing it in both loops avoids fetching attributes and invoking
+	 * operators for rows already rejected.
+	 */
+	for (int q = 1; q < state->nquals && mask != 0; q++)
+	{
+		struct SeqScanBatchQual *qual = &state->quals[q];
+		FunctionCallInfo fcinfo = qual->fcinfo;
+		NullableDatum *arg = &fcinfo->args[1 - qual->var_argno];
+
+		if (qual->param_expr != NULL)
+			arg->value = ExecEvalExpr(qual->param_expr, econtext,
+								  &arg->isnull);
+		if (arg->isnull)
+		{
+			if (qual->param_expr != NULL)
+				arg->value = (Datum) 0;
+			mask = 0;
+			break;
+		}
+
+		position = ScanDirectionIsForward(direction) ? 0 : nrows - 1;
+		for (int i = 0; i < nrows; i++, position += direction)
+		{
+			bool		isnull;
+			uint64		bit = UINT64CONST(1) << position;
+
+			if ((mask & bit) == 0)
+				continue;
+			values[position] = heap_page_batch_getattr(page, batch,
+											first + position,
+											slot->tts_tupleDescriptor,
+											qual->attnum, &isnull);
+			if (isnull)
+				mask &= ~bit;
+		}
+
+		position = ScanDirectionIsForward(direction) ? 0 : nrows - 1;
+		for (int i = 0; i < nrows && mask != 0;
+			 i++, position += direction)
+		{
+			uint64		bit = UINT64CONST(1) << position;
+
+			if ((mask & bit) == 0)
+				continue;
+			if (!eval_batch_qual(qual, values[position], false))
+				mask &= ~bit;
+		}
+
+		if (qual->param_expr != NULL)
+		{
+			arg->value = (Datum) 0;
+			arg->isnull = true;
+		}
+	}
+	MemoryContextSwitchTo(oldcontext);
+
+	state->mask = mask;
+	state->block = block;
+	state->first = first;
+	state->nrows = nrows;
+}
+
+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 OpExpr *
+match_batch_qual(Node *clause, uint8 *var_argno)
+{
+	OpExpr	   *op;
+	Node	   *left;
+	Node	   *right;
+	Var		   *var;
+
+	if (!IsA(clause, OpExpr))
+		return NULL;
+	op = (OpExpr *) clause;
+	if (list_length(op->args) != 2)
+		return NULL;
+
+	left = linitial(op->args);
+	right = lsecond(op->args);
+	if (IsA(left, Var) && is_batch_qual_value(right))
+		*var_argno = 0;
+	else if (is_batch_qual_value(left) && IsA(right, Var))
+		*var_argno = 1;
+	else
+		return NULL;
+	var = list_nth_node(Var, op->args, *var_argno);
+
+	if (var->varattno <= 0 || var->varlevelsup != 0 ||
+		var->varreturningtype != VAR_RETURNING_DEFAULT)
+		return NULL;
+	if (op->opresulttype != BOOLOID ||
+		op->opretset ||
+		func_volatile(op->opfuncid) != PROVOLATILE_IMMUTABLE ||
+		!func_strict(op->opfuncid) ||
+		!get_func_leakproof(op->opfuncid))
+		return NULL;
+
+	return op;
+}
+
+static void
+init_batch_qual(struct SeqScanBatchQual *qual, OpExpr *op,
+				PlanState *parent)
+{
+	uint8		var_argno = IsA(linitial(op->args), Var) ? 0 : 1;
+	Expr	   *arg = (Expr *) list_nth(op->args, 1 - var_argno);
+	Var		   *var = list_nth_node(Var, 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->fcinfo->args[1 - var_argno].value =
+			castNode(Const, arg)->constvalue;
+		qual->fcinfo->args[1 - var_argno].isnull =
+			castNode(Const, arg)->constisnull;
+	}
+	else
+	{
+		Assert(IsA(arg, Param));
+		qual->param_expr = ExecInitExpr(arg, parent);
+	}
+	qual->attnum = var->varattno;
+	qual->var_argno = var_argno;
+}
+
+static struct SeqScanBatchState *
+init_batch_state(SeqScan *node, PlanState *parent)
+{
+	struct SeqScanBatchState *state;
+	List	   *clauses = node->scan.plan.qual;
+	List	   *batch_clauses;
+	List	   *remaining_clauses;
+	int			nclauses = list_length(clauses);
+	int			nquals = 0;
+	int			i = 0;
+
+	if (nclauses == 0)
+		return NULL;
+
+	foreach_ptr(Node, clause, clauses)
+	{
+		uint8		var_argno;
+
+		if (match_batch_qual(clause, &var_argno) == NULL)
+			break;
+		nquals++;
+	}
+	if (nquals == 0)
+		return NULL;
+
+	batch_clauses = list_copy_head(clauses, nquals);
+	remaining_clauses = list_copy_tail(clauses, nquals);
+	state = palloc0_object(struct SeqScanBatchState);
+	state->nquals = nquals;
+	state->scalar_qual = ExecInitQual(batch_clauses, parent);
+	parent->qual = ExecInitQual(remaining_clauses, parent);
+	state->quals = palloc0_array(struct SeqScanBatchQual, state->nquals);
+	foreach_ptr(OpExpr, op, batch_clauses)
+		init_batch_qual(&state->quals[i++], op, parent);
+
+	return state;
+}
+
 /* ----------------------------------------------------------------
  *						Scan Support
  * ----------------------------------------------------------------
@@ -156,6 +557,140 @@ ExecSeqScanWithQual(PlanState *pstate)
 							NULL);
 }
 
+static pg_always_inline TupleTableSlot *
+ExecSeqScanBatchLoop(PlanState *pstate, ExprState *rest_qual,
+					 ProjectionInfo *projInfo, bool use_batch)
+{
+	SeqScanState *node = castNode(SeqScanState, pstate);
+	ExprContext *econtext = pstate->ps_ExprContext;
+	TupleTableSlot *result_slot;
+	TupleTableSlot *slot;
+	ScanDirection direction;
+	bool		prepare_batch;
+	bool		qual_result;
+
+	Assert(pstate->state->es_epq_active == NULL);
+	Assert(projInfo == pstate->ps_ProjInfo);
+	Assert(node->batch_state != NULL);
+	pg_assume(node->batch_state->scalar_qual != NULL);
+
+	direction = pstate->state->es_direction;
+	if (!use_batch)
+		reset_batch_qual(node->batch_state);
+
+	/*
+	 * This is the no-EPQ path from ExecScanExtended(), kept here so batch qual
+	 * evaluation can be added below.
+	 */
+	ResetExprContext(econtext);
+	for (;;)
+	{
+		slot = ExecScanFetch(&node->ss, NULL,
+							 (ExecScanAccessMtd) SeqNext,
+							 (ExecScanRecheckMtd) SeqRecheck);
+		if (TupIsNull(slot))
+		{
+			reset_batch_qual(node->batch_state);
+			if (projInfo != NULL)
+				return ExecClearTuple(projInfo->pi_state.resultslot);
+			return slot;
+		}
+
+		prepare_batch = false;
+		if (!use_batch)
+			qual_result = eval_scalar_qual(node->batch_state, slot, econtext);
+		else if (!reuse_batch_qual(node->batch_state, slot, &qual_result))
+		{
+			qual_result = eval_scalar_qual(node->batch_state, slot, econtext);
+			prepare_batch = true;
+		}
+
+		if (qual_result && rest_qual != NULL)
+		{
+			econtext->ecxt_scantuple = slot;
+			qual_result = ExecQual(rest_qual, econtext);
+		}
+
+		result_slot = slot;
+		if (qual_result && projInfo != NULL)
+		{
+			econtext->ecxt_scantuple = slot;
+			result_slot = ExecProject(projInfo);
+		}
+
+		/* Finish the current tuple before evaluating quals for later tuples. */
+		if (prepare_batch)
+			prepare_batch_qual(node->batch_state, slot, direction, econtext);
+
+		if (qual_result)
+			return result_slot;
+
+		InstrCountFiltered1(node, 1);
+		ResetExprContext(econtext);
+	}
+}
+
+static pg_always_inline TupleTableSlot *
+ExecSeqScanBatchDispatch(PlanState *pstate, ExprState *rest_qual,
+						 ProjectionInfo *projInfo)
+{
+	ScanDirection direction = pstate->state->es_direction;
+
+	if (ScanDirectionIsForward(direction) ||
+		ScanDirectionIsBackward(direction))
+		return ExecSeqScanBatchLoop(pstate, rest_qual, projInfo, true);
+	return ExecSeqScanBatchLoop(pstate, rest_qual, projInfo, false);
+}
+
+/*
+ * Variant of ExecSeqScan() when all scan quals can be evaluated in batches.
+ */
+static TupleTableSlot *
+ExecSeqScanWithBatchQual(PlanState *pstate)
+{
+	Assert(pstate->qual == NULL);
+
+	return ExecSeqScanBatchDispatch(pstate, NULL, NULL);
+}
+
+/*
+ * Variant of ExecSeqScan() when batch quals are followed by remaining quals.
+ */
+static TupleTableSlot *
+ExecSeqScanWithBatchQualRest(PlanState *pstate)
+{
+	pg_assume(pstate->qual != NULL);
+
+	return ExecSeqScanBatchDispatch(pstate, pstate->qual, NULL);
+}
+
+/*
+ * Variant of ExecSeqScan() when all scan quals can be evaluated in batches
+ * and projection is required.
+ */
+static TupleTableSlot *
+ExecSeqScanWithBatchQualProject(PlanState *pstate)
+{
+	Assert(pstate->qual == NULL);
+	pg_assume(pstate->ps_ProjInfo != NULL);
+
+	return ExecSeqScanBatchDispatch(pstate, NULL, pstate->ps_ProjInfo);
+}
+
+/*
+ * Variant of ExecSeqScan() when batch quals are followed by remaining quals
+ * and projection is required.
+ */
+static TupleTableSlot *
+ExecSeqScanWithBatchQualRestProject(PlanState *pstate)
+{
+	pg_assume(pstate->qual != NULL);
+	pg_assume(pstate->ps_ProjInfo != NULL);
+
+	return ExecSeqScanBatchDispatch(pstate, pstate->qual,
+								pstate->ps_ProjInfo);
+}
+
 /*
  * Variant of ExecSeqScan() but when projection is required.
  */
@@ -265,16 +800,32 @@ 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 = init_batch_state(node, &scanstate->ss.ps);
+	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
-	 * based on the presence of qual and projection. Each ExecSeqScan*()
-	 * variant is optimized for the specific combination of these conditions.
+	 * based on the presence of batch state, qual and projection. Each
+	 * ExecSeqScan*() variant is optimized for the specific combination of
+	 * these conditions.
 	 */
 	if (scanstate->ss.ps.state->es_epq_active != NULL)
 		scanstate->ss.ps.ExecProcNode = ExecSeqScanEPQ;
+	else if (scanstate->batch_state != NULL)
+	{
+		if (scanstate->ss.ps.qual == NULL &&
+			scanstate->ss.ps.ps_ProjInfo == NULL)
+			scanstate->ss.ps.ExecProcNode = ExecSeqScanWithBatchQual;
+		else if (scanstate->ss.ps.qual == NULL)
+			scanstate->ss.ps.ExecProcNode = ExecSeqScanWithBatchQualProject;
+		else if (scanstate->ss.ps.ps_ProjInfo == NULL)
+			scanstate->ss.ps.ExecProcNode = ExecSeqScanWithBatchQualRest;
+		else
+			scanstate->ss.ps.ExecProcNode = ExecSeqScanWithBatchQualRestProject;
+	}
 	else if (scanstate->ss.ps.qual == NULL)
 	{
 		if (scanstate->ss.ps.ps_ProjInfo == NULL)
@@ -348,6 +899,9 @@ ExecReScanSeqScan(SeqScanState *node)
 {
 	TableScanDesc scan;
 
+	if (node->batch_state != NULL)
+		reset_batch_qual(node->batch_state);
+
 	scan = node->ss.ss_currentScanDesc;
 
 	if (scan != NULL)
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index e64fd8c7ea3..4c9b2e8986e 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1674,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;
+	struct SeqScanBatchState *batch_state;
 } SeqScanState;
 
 /* ----------------
diff --git a/src/test/regress/expected/plancache.out b/src/test/regress/expected/plancache.out
index d58534ca1cd..c8d471a2c2f 100644
--- a/src/test/regress/expected/plancache.out
+++ b/src/test/regress/expected/plancache.out
@@ -402,3 +402,132 @@ select name, generic_plans, custom_plans from pg_prepared_statements
 (1 row)
 
 drop table test_mode;
+-- Set up a table spanning multiple heap pages for batch qualification tests.
+create table batch_param_test (a int, t text, padding text)
+  with (autovacuum_enabled = false);
+insert into batch_param_test
+  select i, 'value-' || i, repeat('x', 500)
+  from generate_series(1, 200) i;
+-- Test multiple batch qualifications.
+select count(*) from batch_param_test
+  where a > 50 and a < 151 and t <> 'value-100' \gset batch_multi_const_
+\echo :batch_multi_const_count
+99
+-- Test batch qualifications followed by regular qualification.
+explain (analyze, costs off, timing off, summary off, buffers off)
+select * from batch_param_test where a > 50 and length(t) = 8;
+                        QUERY PLAN                        
+----------------------------------------------------------
+ Seq Scan on batch_param_test (actual rows=49.00 loops=1)
+   Filter: ((a > 50) AND (length(t) = 8))
+   Rows Removed by Filter: 151
+(3 rows)
+
+-- Test projection with batch qualification in both scan directions.
+begin;
+declare batch_multi_scroll scroll cursor for
+  select a + 1 as projected, upper(t) as projected_text
+  from batch_param_test
+  where a > 50 and a < 55 and length(t) = 8;
+fetch forward 1 from batch_multi_scroll \gset batch_scroll_first_
+fetch forward 1 from batch_multi_scroll \gset batch_scroll_second_
+fetch backward 1 from batch_multi_scroll \gset batch_scroll_backward_
+fetch forward 1 from batch_multi_scroll \gset batch_scroll_again_
+\echo :batch_scroll_first_projected :batch_scroll_first_projected_text
+52 VALUE-51
+\echo :batch_scroll_second_projected :batch_scroll_second_projected_text
+53 VALUE-52
+\echo :batch_scroll_backward_projected :batch_scroll_backward_projected_text
+52 VALUE-51
+\echo :batch_scroll_again_projected :batch_scroll_again_projected_text
+53 VALUE-52
+close batch_multi_scroll;
+commit;
+-- Check the projection result slot at end of scan.
+select a + 1 as projected
+from batch_param_test
+where a > 200 and length(t) = 8;
+ projected 
+-----------
+(0 rows)
+
+-- Test batch qualification with external parameters in generic plans.
+set plan_cache_mode = force_generic_plan;
+prepare batch_param_int_rev(int) as
+  select count(*) from batch_param_test where $1 > a;
+prepare batch_param_multi(int, int, text) as
+  select count(*) from batch_param_test
+  where a > $1 and a < $2 and t <> $3;
+prepare batch_param_rest(int, text) as
+  select count(*) from batch_param_test
+  where a > $1 and t <> $2 and length(t) = 8;
+explain (costs off) execute batch_param_rest(50, 'value-75');
+                          QUERY PLAN                          
+--------------------------------------------------------------
+ Aggregate
+   ->  Seq Scan on batch_param_test
+         Filter: ((a > $1) AND (t <> $2) AND (length(t) = 8))
+(3 rows)
+
+execute batch_param_int_rev(101) \gset batch_int_rev_
+execute batch_param_multi(50, 151, 'value-100') \gset batch_multi_first_
+execute batch_param_multi(100, 201, 'value-150') \gset batch_multi_second_
+execute batch_param_multi(null, 151, 'value-100') \gset batch_multi_null_first_
+execute batch_param_multi(50, 151, null) \gset batch_multi_null_last_
+\echo :batch_int_rev_count :batch_multi_first_count :batch_multi_second_count :batch_multi_null_first_count :batch_multi_null_last_count
+100 99 99 0 0
+execute batch_param_rest(50, 'value-75') \gset batch_rest_first_
+execute batch_param_rest(90, 'value-95') \gset batch_rest_second_
+\echo :batch_rest_first_count :batch_rest_second_count
+48 8
+-- Test PL/pgSQL parameter compilation with batch qualification.
+create function batch_param_plpgsql(p int, q text)
+returns table (int_result bigint, text_result bigint)
+language plpgsql as $$
+begin
+  return query
+    select (select count(*) from batch_param_test b where b.a < p),
+           (select count(*) from batch_param_test b where b.t = q);
+end
+$$;
+select * from batch_param_plpgsql(1, 'value-100') \gset batch_plpgsql_first_
+select * from batch_param_plpgsql(101, 'missing') \gset batch_plpgsql_second_
+select * from batch_param_plpgsql(null, null) \gset batch_plpgsql_null_
+\echo :batch_plpgsql_first_int_result :batch_plpgsql_first_text_result :batch_plpgsql_second_int_result :batch_plpgsql_second_text_result :batch_plpgsql_null_int_result :batch_plpgsql_null_text_result
+0 1 100 0 0 0
+drop function batch_param_plpgsql(int, text);
+-- Test batch qualification with internal executor parameters.
+create table batch_param_exec_test (id int, p int, q text);
+insert into batch_param_exec_test values
+  (1, 1, 'value-2'),
+  (2, 200, 'missing'),
+  (3, null, null),
+  (4, 199, 'value-200');
+-- OFFSET 0 prevents conversion of EXISTS, keeping the correlated condition
+-- as a scan qualification.
+explain (costs off)
+select exists (select from batch_param_test b
+               where o.p < b.a and b.t <> o.q and length(b.t) = 8
+               offset 0)
+from batch_param_exec_test o;
+                            QUERY PLAN                            
+------------------------------------------------------------------
+ Seq Scan on batch_param_exec_test o
+   SubPlan exists_1
+     ->  Seq Scan on batch_param_test b
+           Filter: ((o.p < a) AND (t <> o.q) AND (length(t) = 8))
+(4 rows)
+
+select array_agg(exists (select from batch_param_test b
+                         where o.p < b.a and b.t <> o.q and length(b.t) = 8
+                         offset 0)
+                 order by o.id) as combined_results
+from batch_param_exec_test o;
+ combined_results 
+------------------
+ {t,f,f,f}
+(1 row)
+
+drop table batch_param_exec_test;
+reset plan_cache_mode;
+drop table batch_param_test;
diff --git a/src/test/regress/sql/plancache.sql b/src/test/regress/sql/plancache.sql
index aed388d03a1..ecb35177ce9 100644
--- a/src/test/regress/sql/plancache.sql
+++ b/src/test/regress/sql/plancache.sql
@@ -228,3 +228,107 @@ select name, generic_plans, custom_plans from pg_prepared_statements
   where  name = 'test_mode_pp';
 
 drop table test_mode;
+
+-- Set up a table spanning multiple heap pages for batch qualification tests.
+create table batch_param_test (a int, t text, padding text)
+  with (autovacuum_enabled = false);
+insert into batch_param_test
+  select i, 'value-' || i, repeat('x', 500)
+  from generate_series(1, 200) i;
+
+-- Test multiple batch qualifications.
+select count(*) from batch_param_test
+  where a > 50 and a < 151 and t <> 'value-100' \gset batch_multi_const_
+\echo :batch_multi_const_count
+
+-- Test batch qualifications followed by regular qualification.
+explain (analyze, costs off, timing off, summary off, buffers off)
+select * from batch_param_test where a > 50 and length(t) = 8;
+
+-- Test projection with batch qualification in both scan directions.
+begin;
+declare batch_multi_scroll scroll cursor for
+  select a + 1 as projected, upper(t) as projected_text
+  from batch_param_test
+  where a > 50 and a < 55 and length(t) = 8;
+fetch forward 1 from batch_multi_scroll \gset batch_scroll_first_
+fetch forward 1 from batch_multi_scroll \gset batch_scroll_second_
+fetch backward 1 from batch_multi_scroll \gset batch_scroll_backward_
+fetch forward 1 from batch_multi_scroll \gset batch_scroll_again_
+\echo :batch_scroll_first_projected :batch_scroll_first_projected_text
+\echo :batch_scroll_second_projected :batch_scroll_second_projected_text
+\echo :batch_scroll_backward_projected :batch_scroll_backward_projected_text
+\echo :batch_scroll_again_projected :batch_scroll_again_projected_text
+close batch_multi_scroll;
+commit;
+
+-- Check the projection result slot at end of scan.
+select a + 1 as projected
+from batch_param_test
+where a > 200 and length(t) = 8;
+
+-- Test batch qualification with external parameters in generic plans.
+set plan_cache_mode = force_generic_plan;
+prepare batch_param_int_rev(int) as
+  select count(*) from batch_param_test where $1 > a;
+prepare batch_param_multi(int, int, text) as
+  select count(*) from batch_param_test
+  where a > $1 and a < $2 and t <> $3;
+prepare batch_param_rest(int, text) as
+  select count(*) from batch_param_test
+  where a > $1 and t <> $2 and length(t) = 8;
+
+explain (costs off) execute batch_param_rest(50, 'value-75');
+execute batch_param_int_rev(101) \gset batch_int_rev_
+execute batch_param_multi(50, 151, 'value-100') \gset batch_multi_first_
+execute batch_param_multi(100, 201, 'value-150') \gset batch_multi_second_
+execute batch_param_multi(null, 151, 'value-100') \gset batch_multi_null_first_
+execute batch_param_multi(50, 151, null) \gset batch_multi_null_last_
+\echo :batch_int_rev_count :batch_multi_first_count :batch_multi_second_count :batch_multi_null_first_count :batch_multi_null_last_count
+execute batch_param_rest(50, 'value-75') \gset batch_rest_first_
+execute batch_param_rest(90, 'value-95') \gset batch_rest_second_
+\echo :batch_rest_first_count :batch_rest_second_count
+
+-- Test PL/pgSQL parameter compilation with batch qualification.
+create function batch_param_plpgsql(p int, q text)
+returns table (int_result bigint, text_result bigint)
+language plpgsql as $$
+begin
+  return query
+    select (select count(*) from batch_param_test b where b.a < p),
+           (select count(*) from batch_param_test b where b.t = q);
+end
+$$;
+
+select * from batch_param_plpgsql(1, 'value-100') \gset batch_plpgsql_first_
+select * from batch_param_plpgsql(101, 'missing') \gset batch_plpgsql_second_
+select * from batch_param_plpgsql(null, null) \gset batch_plpgsql_null_
+\echo :batch_plpgsql_first_int_result :batch_plpgsql_first_text_result :batch_plpgsql_second_int_result :batch_plpgsql_second_text_result :batch_plpgsql_null_int_result :batch_plpgsql_null_text_result
+
+drop function batch_param_plpgsql(int, text);
+
+-- Test batch qualification with internal executor parameters.
+create table batch_param_exec_test (id int, p int, q text);
+insert into batch_param_exec_test values
+  (1, 1, 'value-2'),
+  (2, 200, 'missing'),
+  (3, null, null),
+  (4, 199, 'value-200');
+
+-- OFFSET 0 prevents conversion of EXISTS, keeping the correlated condition
+-- as a scan qualification.
+explain (costs off)
+select exists (select from batch_param_test b
+               where o.p < b.a and b.t <> o.q and length(b.t) = 8
+               offset 0)
+from batch_param_exec_test o;
+
+select array_agg(exists (select from batch_param_test b
+                         where o.p < b.a and b.t <> o.q and length(b.t) = 8
+                         offset 0)
+                 order by o.id) as combined_results
+from batch_param_exec_test o;
+
+drop table batch_param_exec_test;
+reset plan_cache_mode;
+drop table batch_param_test;
-- 
2.54.0

