v38-0005-Row-pattern-recognition-patch-executor-and-comma.patch
application/octet-stream
Filename: v38-0005-Row-pattern-recognition-patch-executor-and-comma.patch
Type: application/octet-stream
Part: 4
Message:
Re: Row pattern recognition
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: format-patch
Series: patch v38-0005
Subject: Row pattern recognition patch (executor and commands).
| File | + | − |
|---|---|---|
| src/backend/commands/explain.c | 143 | 0 |
| src/backend/executor/nodeWindowAgg.c | 1770 | 9 |
| src/backend/utils/adt/windowfuncs.c | 32 | 2 |
| src/include/catalog/pg_proc.dat | 6 | 0 |
| src/include/nodes/execnodes.h | 64 | 0 |
From 19dfddeb03bf6d824ecbc94c87462990581af00d Mon Sep 17 00:00:00 2001
From: Tatsuo Ishii <ishii@postgresql.org>
Date: Thu, 15 Jan 2026 13:26:44 +0900
Subject: [PATCH v38 5/8] Row pattern recognition patch (executor and
commands).
---
src/backend/commands/explain.c | 143 +++
src/backend/executor/nodeWindowAgg.c | 1779 +++++++++++++++++++++++++-
src/backend/utils/adt/windowfuncs.c | 34 +-
src/include/catalog/pg_proc.dat | 6 +
src/include/nodes/execnodes.h | 64 +
5 files changed, 2015 insertions(+), 11 deletions(-)
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index b7bb111688c..969c9195864 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -29,6 +29,7 @@
#include "nodes/extensible.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
+#include "optimizer/rpr.h"
#include "parser/analyze.h"
#include "parser/parsetree.h"
#include "rewrite/rewriteHandler.h"
@@ -117,6 +118,8 @@ static void show_window_def(WindowAggState *planstate,
static void show_window_keys(StringInfo buf, PlanState *planstate,
int nkeys, AttrNumber *keycols,
List *ancestors, ExplainState *es);
+static void append_rpr_quantifier(StringInfo buf, RPRPatternElement *elem);
+static char *deparse_rpr_pattern(RPRPattern *pattern);
static void show_storage_info(char *maxStorageType, int64 maxSpaceUsed,
ExplainState *es);
static void show_tablesample(TableSampleClause *tsc, PlanState *planstate,
@@ -2889,6 +2892,134 @@ show_sortorder_options(StringInfo buf, Node *sortexpr,
}
}
+/*
+ * Append quantifier suffix for a pattern element.
+ */
+static void
+append_rpr_quantifier(StringInfo buf, RPRPatternElement *elem)
+{
+ if (elem->min == 1 && elem->max == 1)
+ return; /* no quantifier */
+ else if (elem->min == 0 && elem->max == RPR_QUANTITY_INF)
+ appendStringInfoChar(buf, '*');
+ else if (elem->min == 1 && elem->max == RPR_QUANTITY_INF)
+ appendStringInfoChar(buf, '+');
+ else if (elem->min == 0 && elem->max == 1)
+ appendStringInfoChar(buf, '?');
+ else if (elem->max == RPR_QUANTITY_INF)
+ appendStringInfo(buf, "{%d,}", elem->min);
+ else if (elem->min == elem->max)
+ appendStringInfo(buf, "{%d}", elem->min);
+ else
+ appendStringInfo(buf, "{%d,%d}", elem->min, elem->max);
+
+ if (RPRElemIsReluctant(elem))
+ appendStringInfoChar(buf, '?');
+}
+
+/*
+ * Deparse a compiled RPRPattern (bytecode) back to pattern string.
+ * Simple approach: output parens for each depth level as-is.
+ */
+static char *
+deparse_rpr_pattern(RPRPattern *pattern)
+{
+ StringInfoData buf;
+ int i;
+ RPRDepth prevDepth = 0;
+ bool needSpace = false;
+ RPRElemIdx altSepAt = RPR_ELEMIDX_INVALID;
+
+ if (pattern == NULL || pattern->numElements == 0)
+ return NULL;
+
+ initStringInfo(&buf);
+
+ for (i = 0; i < pattern->numElements; i++)
+ {
+ RPRPatternElement *elem = &pattern->elements[i];
+
+ if (RPRElemIsFin(elem))
+ break;
+
+ /* Alternation separator */
+ if (altSepAt == i)
+ {
+ appendStringInfoString(&buf, " | ");
+ needSpace = false;
+ altSepAt = RPR_ELEMIDX_INVALID;
+ }
+
+ if (RPRElemIsAlt(elem))
+ {
+ /* Open parens up to ALT's content depth */
+ while (prevDepth <= elem->depth)
+ {
+ if (needSpace)
+ appendStringInfoChar(&buf, ' ');
+ appendStringInfoChar(&buf, '(');
+ prevDepth++;
+ needSpace = false;
+ }
+ continue;
+ }
+
+ if (RPRElemIsEnd(elem))
+ {
+ /* Close down to END's depth, output quantifier */
+ while (prevDepth > elem->depth + 1)
+ {
+ appendStringInfoChar(&buf, ')');
+ prevDepth--;
+ }
+ appendStringInfoChar(&buf, ')');
+ append_rpr_quantifier(&buf, elem);
+ prevDepth = elem->depth;
+ needSpace = true;
+ continue;
+ }
+
+ if (RPRElemIsVar(elem))
+ {
+ /* Open parens for depth increase */
+ while (prevDepth < elem->depth)
+ {
+ if (needSpace)
+ appendStringInfoChar(&buf, ' ');
+ appendStringInfoChar(&buf, '(');
+ prevDepth++;
+ needSpace = false;
+ }
+
+ /* Close parens for depth decrease */
+ while (prevDepth > elem->depth)
+ {
+ appendStringInfoChar(&buf, ')');
+ prevDepth--;
+ }
+
+ if (needSpace)
+ appendStringInfoChar(&buf, ' ');
+
+ appendStringInfoString(&buf, pattern->varNames[elem->varId]);
+ append_rpr_quantifier(&buf, elem);
+ needSpace = true;
+
+ if (elem->jump != RPR_ELEMIDX_INVALID)
+ altSepAt = elem->jump;
+ }
+ }
+
+ /* Close remaining */
+ while (prevDepth > 0)
+ {
+ appendStringInfoChar(&buf, ')');
+ prevDepth--;
+ }
+
+ return buf.data;
+}
+
/*
* Show the window definition for a WindowAgg node.
*/
@@ -2947,6 +3078,18 @@ show_window_def(WindowAggState *planstate, List *ancestors, ExplainState *es)
appendStringInfoChar(&wbuf, ')');
ExplainPropertyText("Window", wbuf.data, es);
pfree(wbuf.data);
+
+ /* Show Row Pattern Recognition pattern if present */
+ if (wagg->rpPattern != NULL)
+ {
+ char *patternStr = deparse_rpr_pattern(wagg->rpPattern);
+
+ if (patternStr != NULL)
+ {
+ ExplainPropertyText("Pattern", patternStr, es);
+ pfree(patternStr);
+ }
+ }
}
/*
diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c
index d9b64b0f465..cf43c1f6127 100644
--- a/src/backend/executor/nodeWindowAgg.c
+++ b/src/backend/executor/nodeWindowAgg.c
@@ -36,18 +36,23 @@
#include "access/htup_details.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_aggregate.h"
+#include "catalog/pg_collation_d.h"
#include "catalog/pg_proc.h"
#include "executor/executor.h"
#include "executor/nodeWindowAgg.h"
#include "miscadmin.h"
#include "nodes/nodeFuncs.h"
+#include "nodes/plannodes.h"
#include "optimizer/clauses.h"
#include "optimizer/optimizer.h"
+#include "optimizer/rpr.h"
#include "parser/parse_agg.h"
#include "parser/parse_coerce.h"
+#include "regex/regex.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/datum.h"
+#include "utils/fmgroids.h"
#include "utils/expandeddatum.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
@@ -170,6 +175,15 @@ typedef struct WindowStatePerAggData
bool restart; /* need to restart this agg in this cycle? */
} WindowStatePerAggData;
+/*
+ * Structure used by check_rpr_navigation() and rpr_navigation_walker().
+ */
+typedef struct NavigationInfo
+{
+ bool is_prev; /* true if PREV */
+ int num_vars; /* number of var nodes */
+} NavigationInfo;
+
static void initialize_windowaggregate(WindowAggState *winstate,
WindowStatePerFunc perfuncstate,
WindowStatePerAgg peraggstate);
@@ -206,6 +220,9 @@ static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
static bool are_peers(WindowAggState *winstate, TupleTableSlot *slot1,
TupleTableSlot *slot2);
+static int WinGetSlotInFrame(WindowObject winobj, TupleTableSlot *slot,
+ int relpos, int seektype, bool set_mark,
+ bool *isnull, bool *isout);
static bool window_gettupleslot(WindowObject winobj, int64 pos,
TupleTableSlot *slot);
@@ -224,6 +241,40 @@ static uint8 get_notnull_info(WindowObject winobj,
int64 pos, int argno);
static void put_notnull_info(WindowObject winobj,
int64 pos, int argno, bool isnull);
+static void attno_map(Node *node);
+static bool attno_map_walker(Node *node, void *context);
+static int row_is_in_reduced_frame(WindowObject winobj, int64 pos);
+static bool rpr_is_defined(WindowAggState *winstate);
+
+static void create_reduced_frame_map(WindowAggState *winstate);
+static int get_reduced_frame_map(WindowAggState *winstate, int64 pos);
+static void register_reduced_frame_map(WindowAggState *winstate, int64 pos,
+ int val);
+static void clear_reduced_frame_map(WindowAggState *winstate);
+static void update_reduced_frame(WindowObject winobj, int64 pos);
+
+static void check_rpr_navigation(Node *node, bool is_prev);
+static bool rpr_navigation_walker(Node *node, void *context);
+
+/* NFA-based pattern matching functions */
+static RPRNFAState *nfa_state_alloc(WindowAggState *winstate);
+static void nfa_state_free(WindowAggState *winstate, RPRNFAState *state);
+static void nfa_state_free_list(WindowAggState *winstate, RPRNFAState *list);
+static RPRNFAState *nfa_state_clone(WindowAggState *winstate, int16 elemIdx,
+ int16 altPriority, int16 *counts,
+ RPRNFAState *list);
+static bool nfa_evaluate_row(WindowObject winobj, int64 pos, bool *varMatched, bool *anyMatch);
+static RPRNFAContext *nfa_context_alloc(WindowAggState *winstate);
+static void nfa_unlink_context(WindowAggState *winstate, RPRNFAContext *ctx);
+static void nfa_context_free(WindowAggState *winstate, RPRNFAContext *ctx);
+static void nfa_start_context(WindowAggState *winstate, int64 startPos);
+static void nfa_step_single(WindowAggState *winstate, RPRNFAContext *ctx,
+ RPRNFAState *state, bool *varMatched, int64 currentPos);
+static void nfa_finalize_boundary(WindowAggState *winstate, RPRNFAContext *ctx,
+ int64 matchEndPos);
+static RPRNFAContext *nfa_find_context_for_pos(WindowAggState *winstate, int64 pos);
+static void nfa_remove_contexts_up_to(WindowAggState *winstate, int64 endPos);
+static void nfa_absorb_contexts(WindowAggState *winstate, RPRNFAContext *excludeCtx, int64 currentPos);
/*
* Not null info bit array consists of 2-bit items
@@ -817,6 +868,7 @@ eval_windowaggregates(WindowAggState *winstate)
* transition function, or
* - we have an EXCLUSION clause, or
* - if the new frame doesn't overlap the old one
+ * - if RPR is enabled
*
* Note that we don't strictly need to restart in the last case, but if
* we're going to remove all rows from the aggregation anyway, a restart
@@ -831,7 +883,8 @@ eval_windowaggregates(WindowAggState *winstate)
(winstate->aggregatedbase != winstate->frameheadpos &&
!OidIsValid(peraggstate->invtransfn_oid)) ||
(winstate->frameOptions & FRAMEOPTION_EXCLUSION) ||
- winstate->aggregatedupto <= winstate->frameheadpos)
+ winstate->aggregatedupto <= winstate->frameheadpos ||
+ rpr_is_defined(winstate))
{
peraggstate->restart = true;
numaggs_restart++;
@@ -905,7 +958,22 @@ eval_windowaggregates(WindowAggState *winstate)
* head, so that tuplestore can discard unnecessary rows.
*/
if (agg_winobj->markptr >= 0)
- WinSetMarkPosition(agg_winobj, winstate->frameheadpos);
+ {
+ int64 markpos = winstate->frameheadpos;
+
+ if (rpr_is_defined(winstate))
+ {
+ /*
+ * If RPR is used, it is possible PREV wants to look at the
+ * previous row. So the mark pos should be frameheadpos - 1
+ * unless it is below 0.
+ */
+ markpos -= 1;
+ if (markpos < 0)
+ markpos = 0;
+ }
+ WinSetMarkPosition(agg_winobj, markpos);
+ }
/*
* Now restart the aggregates that require it.
@@ -960,6 +1028,14 @@ eval_windowaggregates(WindowAggState *winstate)
{
winstate->aggregatedupto = winstate->frameheadpos;
ExecClearTuple(agg_row_slot);
+
+ /*
+ * If RPR is defined, we do not use aggregatedupto_nonrestarted. To
+ * avoid assertion failure below, we reset aggregatedupto_nonrestarted
+ * to frameheadpos.
+ */
+ if (rpr_is_defined(winstate))
+ aggregatedupto_nonrestarted = winstate->frameheadpos;
}
/*
@@ -973,6 +1049,12 @@ eval_windowaggregates(WindowAggState *winstate)
{
int ret;
+#ifdef RPR_DEBUG
+ printf("===== loop in frame starts: aggregatedupto: " INT64_FORMAT " aggregatedbase: " INT64_FORMAT "\n",
+ winstate->aggregatedupto,
+ winstate->aggregatedbase);
+#endif
+
/* Fetch next row if we didn't already */
if (TupIsNull(agg_row_slot))
{
@@ -989,9 +1071,53 @@ eval_windowaggregates(WindowAggState *winstate)
agg_row_slot, false);
if (ret < 0)
break;
+
if (ret == 0)
goto next_tuple;
+ if (rpr_is_defined(winstate))
+ {
+#ifdef RPR_DEBUG
+ printf("reduced_frame_map: %d aggregatedupto: " INT64_FORMAT " aggregatedbase: " INT64_FORMAT "\n",
+ get_reduced_frame_map(winstate,
+ winstate->aggregatedupto),
+ winstate->aggregatedupto,
+ winstate->aggregatedbase);
+#endif
+
+ /*
+ * If the row status at currentpos is already decided and current
+ * row status is not decided yet, it means we passed the last
+ * reduced frame. Time to break the loop.
+ */
+ if (get_reduced_frame_map(winstate, winstate->currentpos)
+ != RF_NOT_DETERMINED &&
+ get_reduced_frame_map(winstate, winstate->aggregatedupto)
+ == RF_NOT_DETERMINED)
+ break;
+
+ /*
+ * Otherwise we need to calculate the reduced frame.
+ */
+ ret = row_is_in_reduced_frame(winstate->agg_winobj,
+ winstate->aggregatedupto);
+ if (ret == -1) /* unmatched row */
+ break;
+
+ /*
+ * Check if current row needs to be skipped due to no match.
+ */
+ if (get_reduced_frame_map(winstate,
+ winstate->aggregatedupto) == RF_SKIPPED &&
+ winstate->aggregatedupto == winstate->aggregatedbase)
+ {
+#ifdef RPR_DEBUG
+ printf("skip current row for aggregation\n");
+#endif
+ break;
+ }
+ }
+
/* Set tuple context for evaluation of aggregate arguments */
winstate->tmpcontext->ecxt_outertuple = agg_row_slot;
@@ -1020,6 +1146,7 @@ next_tuple:
ExecClearTuple(agg_row_slot);
}
+
/* The frame's end is not supposed to move backwards, ever */
Assert(aggregatedupto_nonrestarted <= winstate->aggregatedupto);
@@ -1243,6 +1370,7 @@ begin_partition(WindowAggState *winstate)
winstate->framehead_valid = false;
winstate->frametail_valid = false;
winstate->grouptail_valid = false;
+ create_reduced_frame_map(winstate);
winstate->spooled_rows = 0;
winstate->currentpos = 0;
winstate->frameheadpos = 0;
@@ -1464,6 +1592,13 @@ release_partition(WindowAggState *winstate)
tuplestore_clear(winstate->buffer);
winstate->partition_spooled = false;
winstate->next_partition = true;
+
+ /* Reset NFA state for new partition */
+ winstate->nfaContext = NULL;
+ winstate->nfaContextTail = NULL;
+ winstate->nfaContextFree = NULL;
+ winstate->nfaStateFree = NULL;
+ winstate->nfaLastProcessedRow = -1;
}
/*
@@ -2237,6 +2372,11 @@ ExecWindowAgg(PlanState *pstate)
CHECK_FOR_INTERRUPTS();
+#ifdef RPR_DEBUG
+ printf("ExecWindowAgg called. pos: " INT64_FORMAT "\n",
+ winstate->currentpos);
+#endif
+
if (winstate->status == WINDOWAGG_DONE)
return NULL;
@@ -2345,6 +2485,17 @@ ExecWindowAgg(PlanState *pstate)
/* don't evaluate the window functions when we're in pass-through mode */
if (winstate->status == WINDOWAGG_RUN)
{
+ /*
+ * If RPR is defined and skip mode is next row, we need to clear
+ * existing reduced frame info so that we newly calculate the info
+ * starting from current row.
+ */
+ if (rpr_is_defined(winstate))
+ {
+ if (winstate->rpSkipTo == ST_NEXT_ROW)
+ clear_reduced_frame_map(winstate);
+ }
+
/*
* Evaluate true window functions
*/
@@ -2511,6 +2662,9 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags)
TupleDesc scanDesc;
ListCell *l;
+ TargetEntry *te;
+ Expr *expr;
+
/* check for unsupported flags */
Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
@@ -2609,6 +2763,16 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags)
winstate->temp_slot_2 = ExecInitExtraTupleSlot(estate, scanDesc,
&TTSOpsMinimalTuple);
+ winstate->prev_slot = ExecInitExtraTupleSlot(estate, scanDesc,
+ &TTSOpsMinimalTuple);
+
+ winstate->next_slot = ExecInitExtraTupleSlot(estate, scanDesc,
+ &TTSOpsMinimalTuple);
+
+ winstate->null_slot = ExecInitExtraTupleSlot(estate, scanDesc,
+ &TTSOpsMinimalTuple);
+ winstate->null_slot = ExecStoreAllNullTuple(winstate->null_slot);
+
/*
* create frame head and tail slots only if needed (must create slots in
* exactly the same cases that update_frameheadpos and update_frametailpos
@@ -2795,6 +2959,66 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags)
winstate->inRangeAsc = node->inRangeAsc;
winstate->inRangeNullsFirst = node->inRangeNullsFirst;
+ /* Set up SKIP TO type */
+ winstate->rpSkipTo = node->rpSkipTo;
+ /* Set up row pattern recognition PATTERN clause (compiled NFA) */
+ winstate->rpPattern = node->rpPattern;
+
+ /* Calculate NFA state size for allocation */
+ if (node->rpPattern != NULL)
+ {
+ winstate->nfaStateSize = offsetof(RPRNFAState, counts) +
+ sizeof(int16) * node->rpPattern->maxDepth;
+ }
+
+ /* Set up row pattern recognition DEFINE clause */
+ winstate->defineInitial = node->defineInitial;
+ winstate->defineVariableList = NIL;
+ winstate->defineClauseList = NIL;
+ if (node->defineClause != NIL)
+ {
+ /*
+ * Tweak arg var of PREV/NEXT so that it refers to scan/inner slot.
+ */
+ foreach(l, node->defineClause)
+ {
+ char *name;
+ ExprState *exps;
+
+ te = lfirst(l);
+ name = te->resname;
+ expr = te->expr;
+
+#ifdef RPR_DEBUG
+ printf("defineVariable name: %s\n", name);
+#endif
+ winstate->defineVariableList =
+ lappend(winstate->defineVariableList,
+ makeString(pstrdup(name)));
+ attno_map((Node *) expr);
+ exps = ExecInitExpr(expr, (PlanState *) winstate);
+ winstate->defineClauseList =
+ lappend(winstate->defineClauseList, exps);
+ }
+ }
+
+ /* Initialize NFA free lists for row pattern matching */
+ winstate->nfaContext = NULL;
+ winstate->nfaContextTail = NULL;
+ winstate->nfaContextFree = NULL;
+ winstate->nfaStateFree = NULL;
+ winstate->nfaLastProcessedRow = -1;
+
+ /*
+ * Allocate varMatched array for NFA evaluation.
+ * With the new varNames ordering (DEFINE order first), varId == defineIdx
+ * for all defined variables, so no mapping is needed.
+ */
+ if (list_length(winstate->defineVariableList) > 0)
+ winstate->nfaVarMatched = palloc0(sizeof(bool) *
+ list_length(winstate->defineVariableList));
+ else
+ winstate->nfaVarMatched = NULL;
winstate->all_first = true;
winstate->partition_spooled = false;
winstate->more_partitions = false;
@@ -2803,6 +3027,111 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags)
return winstate;
}
+/*
+ * Rewrite varno of Var nodes that are the argument of PREV/NET so that they
+ * see scan tuple (PREV) or inner tuple (NEXT). Also we check the arguments
+ * of PREV/NEXT include at least 1 column reference. This is required by the
+ * SQL standard.
+ */
+static void
+attno_map(Node *node)
+{
+ (void) expression_tree_walker(node, attno_map_walker, NULL);
+}
+
+static bool
+attno_map_walker(Node *node, void *context)
+{
+ FuncExpr *func;
+ int nargs;
+ bool is_prev;
+
+ if (node == NULL)
+ return false;
+
+ if (IsA(node, FuncExpr))
+ {
+ func = (FuncExpr *) node;
+
+ if (func->funcid == F_PREV || func->funcid == F_NEXT)
+ {
+ /*
+ * The SQL standard allows to have two more arguments form of
+ * PREV/NEXT. But currently we allow only 1 argument form.
+ */
+ nargs = list_length(func->args);
+ if (list_length(func->args) != 1)
+ elog(ERROR, "PREV/NEXT must have 1 argument but function %d has %d args",
+ func->funcid, nargs);
+
+ /*
+ * Check expr of PREV/NEXT aruguments and replace varno.
+ */
+ is_prev = (func->funcid == F_PREV) ? true : false;
+ check_rpr_navigation(node, is_prev);
+ }
+ }
+ return expression_tree_walker(node, attno_map_walker, NULL);
+}
+
+/*
+ * Rewrite varno of Var of RPR navigation operations (PREV/NEXT).
+ * If is_prev is true, we take care PREV, otherwise NEXT.
+ */
+static void
+check_rpr_navigation(Node *node, bool is_prev)
+{
+ NavigationInfo context;
+
+ context.is_prev = is_prev;
+ context.num_vars = 0;
+ (void) expression_tree_walker(node, rpr_navigation_walker, &context);
+ if (context.num_vars < 1)
+ ereport(ERROR,
+ errmsg("row pattern navigation operation's argument must include at least one column reference"));
+}
+
+static bool
+rpr_navigation_walker(Node *node, void *context)
+{
+ NavigationInfo *nav = (NavigationInfo *) context;
+
+ if (node == NULL)
+ return false;
+
+ switch (nodeTag(node))
+ {
+ case T_Var:
+ {
+ Var *var = (Var *) node;
+
+ nav->num_vars++;
+
+ if (nav->is_prev)
+ {
+ /*
+ * Rewrite varno from OUTER_VAR to regular var no so that
+ * the var references scan tuple.
+ */
+ var->varno = var->varnosyn;
+ }
+ else
+ var->varno = INNER_VAR;
+ }
+ break;
+ case T_Const:
+ case T_FuncExpr:
+ case T_OpExpr:
+ break;
+
+ default:
+ ereport(ERROR,
+ errmsg("row pattern navigation operation's argument includes unsupported expression"));
+ }
+ return expression_tree_walker(node, rpr_navigation_walker, context);
+}
+
+
/* -----------------
* ExecEndWindowAgg
* -----------------
@@ -2860,6 +3189,8 @@ ExecReScanWindowAgg(WindowAggState *node)
ExecClearTuple(node->agg_row_slot);
ExecClearTuple(node->temp_slot_1);
ExecClearTuple(node->temp_slot_2);
+ ExecClearTuple(node->prev_slot);
+ ExecClearTuple(node->next_slot);
if (node->framehead_slot)
ExecClearTuple(node->framehead_slot);
if (node->frametail_slot)
@@ -3220,7 +3551,8 @@ window_gettupleslot(WindowObject winobj, int64 pos, TupleTableSlot *slot)
return false;
if (pos < winobj->markpos)
- elog(ERROR, "cannot fetch row before WindowObject's mark position");
+ elog(ERROR, "cannot fetch row: " INT64_FORMAT " before WindowObject's mark position: " INT64_FORMAT,
+ pos, winobj->markpos);
oldcontext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_query_memory);
@@ -3922,8 +4254,6 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno,
WindowAggState *winstate;
ExprContext *econtext;
TupleTableSlot *slot;
- int64 abs_pos;
- int64 mark_pos;
Assert(WindowObjectIsValid(winobj));
winstate = winobj->winstate;
@@ -3934,6 +4264,48 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno,
return ignorenulls_getfuncarginframe(winobj, argno, relpos, seektype,
set_mark, isnull, isout);
+ if (WinGetSlotInFrame(winobj, slot,
+ relpos, seektype, set_mark,
+ isnull, isout) == 0)
+ {
+ econtext->ecxt_outertuple = slot;
+ return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno),
+ econtext, isnull);
+ }
+
+ if (isout)
+ *isout = true;
+ *isnull = true;
+ return (Datum) 0;
+}
+
+/*
+ * WinGetSlotInFrame
+ * slot: TupleTableSlot to store the result
+ * relpos: signed rowcount offset from the seek position
+ * seektype: WINDOW_SEEK_HEAD or WINDOW_SEEK_TAIL
+ * set_mark: If the row is found/in frame and set_mark is true, the mark is
+ * moved to the row as a side-effect.
+ * isnull: output argument, receives isnull status of result
+ * isout: output argument, set to indicate whether target row position
+ * is out of frame (can pass NULL if caller doesn't care about this)
+ *
+ * Returns 0 if we successfullt got the slot. false if out of frame.
+ * (also isout is set)
+ */
+static int
+WinGetSlotInFrame(WindowObject winobj, TupleTableSlot *slot,
+ int relpos, int seektype, bool set_mark,
+ bool *isnull, bool *isout)
+{
+ WindowAggState *winstate;
+ int64 abs_pos;
+ int64 mark_pos;
+ int num_reduced_frame;
+
+ Assert(WindowObjectIsValid(winobj));
+ winstate = winobj->winstate;
+
switch (seektype)
{
case WINDOW_SEEK_CURRENT:
@@ -4000,11 +4372,25 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno,
winstate->frameOptions);
break;
}
+ num_reduced_frame = row_is_in_reduced_frame(winobj,
+ winstate->frameheadpos);
+ if (num_reduced_frame < 0)
+ goto out_of_frame;
+ else if (num_reduced_frame > 0)
+ if (relpos >= num_reduced_frame)
+ goto out_of_frame;
break;
case WINDOW_SEEK_TAIL:
/* rejecting relpos > 0 is easy and simplifies code below */
if (relpos > 0)
goto out_of_frame;
+
+ /*
+ * RPR cares about frame head pos. Need to call
+ * update_frameheadpos
+ */
+ update_frameheadpos(winstate);
+
update_frametailpos(winstate);
abs_pos = winstate->frametailpos - 1 + relpos;
@@ -4071,6 +4457,14 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno,
mark_pos = 0; /* keep compiler quiet */
break;
}
+
+ num_reduced_frame = row_is_in_reduced_frame(winobj,
+ winstate->frameheadpos + relpos);
+ if (num_reduced_frame < 0)
+ goto out_of_frame;
+ else if (num_reduced_frame > 0)
+ abs_pos = winstate->frameheadpos + relpos +
+ num_reduced_frame - 1;
break;
default:
elog(ERROR, "unrecognized window seek type: %d", seektype);
@@ -4089,15 +4483,13 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno,
*isout = false;
if (set_mark)
WinSetMarkPosition(winobj, mark_pos);
- econtext->ecxt_outertuple = slot;
- return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno),
- econtext, isnull);
+ return 0;
out_of_frame:
if (isout)
*isout = true;
*isnull = true;
- return (Datum) 0;
+ return -1;
}
/*
@@ -4128,3 +4520,1372 @@ WinGetFuncArgCurrent(WindowObject winobj, int argno, bool *isnull)
return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno),
econtext, isnull);
}
+
+/*
+ * rpr_is_defined
+ * return true if Row pattern recognition is defined.
+ */
+static bool
+rpr_is_defined(WindowAggState *winstate)
+{
+ return winstate->rpPattern != NULL;
+}
+
+/*
+ * -----------------
+ * row_is_in_reduced_frame
+ * Determine whether a row is in the current row's reduced window frame
+ * according to row pattern matching
+ *
+ * The row must has been already determined that it is in a full window frame
+ * and fetched it into slot.
+ *
+ * Returns:
+ * = 0, RPR is not defined.
+ * >0, if the row is the first in the reduced frame. Return the number of rows
+ * in the reduced frame.
+ * -1, if the row is unmatched row
+ * -2, if the row is in the reduced frame but needed to be skipped because of
+ * AFTER MATCH SKIP PAST LAST ROW
+ * -----------------
+ */
+static int
+row_is_in_reduced_frame(WindowObject winobj, int64 pos)
+{
+ WindowAggState *winstate = winobj->winstate;
+ int state;
+ int rtn;
+
+ if (!rpr_is_defined(winstate))
+ {
+ /*
+ * RPR is not defined. Assume that we are always in the the reduced
+ * window frame.
+ */
+ rtn = 0;
+#ifdef RPR_DEBUG
+ printf("row_is_in_reduced_frame returns %d: pos: " INT64_FORMAT "\n",
+ rtn, pos);
+#endif
+ return rtn;
+ }
+
+ state = get_reduced_frame_map(winstate, pos);
+
+ if (state == RF_NOT_DETERMINED)
+ {
+ update_frameheadpos(winstate);
+ update_reduced_frame(winobj, pos);
+ }
+
+ state = get_reduced_frame_map(winstate, pos);
+
+ switch (state)
+ {
+ int64 i;
+ int num_reduced_rows;
+
+ case RF_FRAME_HEAD:
+ num_reduced_rows = 1;
+ for (i = pos + 1;
+ get_reduced_frame_map(winstate, i) == RF_SKIPPED; i++)
+ num_reduced_rows++;
+ rtn = num_reduced_rows;
+ break;
+
+ case RF_SKIPPED:
+ rtn = -2;
+ break;
+
+ case RF_UNMATCHED:
+ rtn = -1;
+ break;
+
+ default:
+ elog(ERROR, "Unrecognized state: %d at: " INT64_FORMAT,
+ state, pos);
+ break;
+ }
+
+#ifdef RPR_DEBUG
+ printf("row_is_in_reduced_frame returns %d: pos: " INT64_FORMAT "\n",
+ rtn, pos);
+#endif
+ return rtn;
+}
+
+#define REDUCED_FRAME_MAP_INIT_SIZE 1024L
+
+/*
+ * create_reduced_frame_map
+ * Create reduced frame map
+ */
+static void
+create_reduced_frame_map(WindowAggState *winstate)
+{
+ winstate->reduced_frame_map =
+ MemoryContextAlloc(winstate->partcontext,
+ REDUCED_FRAME_MAP_INIT_SIZE);
+ winstate->alloc_sz = REDUCED_FRAME_MAP_INIT_SIZE;
+ clear_reduced_frame_map(winstate);
+}
+
+/*
+ * clear_reduced_frame_map
+ * Clear reduced frame map
+ */
+static void
+clear_reduced_frame_map(WindowAggState *winstate)
+{
+ Assert(winstate->reduced_frame_map != NULL);
+ MemSet(winstate->reduced_frame_map, RF_NOT_DETERMINED,
+ winstate->alloc_sz);
+}
+
+/*
+ * get_reduced_frame_map
+ * Get reduced frame map specified by pos
+ */
+static int
+get_reduced_frame_map(WindowAggState *winstate, int64 pos)
+{
+ Assert(winstate->reduced_frame_map != NULL);
+ Assert(pos >= 0);
+
+ /*
+ * If pos is not in the reduced frame map, it means that any info
+ * regarding the pos has not been registered yet. So we return
+ * RF_NOT_DETERMINED.
+ */
+ if (pos >= winstate->alloc_sz)
+ return RF_NOT_DETERMINED;
+
+ return winstate->reduced_frame_map[pos];
+}
+
+/*
+ * register_reduced_frame_map
+ * Add/replace reduced frame map member at pos.
+ * If there's no enough space, expand the map.
+ */
+static void
+register_reduced_frame_map(WindowAggState *winstate, int64 pos, int val)
+{
+ int64 realloc_sz;
+
+ Assert(winstate->reduced_frame_map != NULL);
+
+ if (pos < 0)
+ elog(ERROR, "wrong pos: " INT64_FORMAT, pos);
+
+ if (pos > winstate->alloc_sz - 1)
+ {
+ realloc_sz = winstate->alloc_sz * 2;
+
+ winstate->reduced_frame_map =
+ repalloc(winstate->reduced_frame_map, realloc_sz);
+
+ MemSet(winstate->reduced_frame_map + winstate->alloc_sz,
+ RF_NOT_DETERMINED, realloc_sz - winstate->alloc_sz);
+
+ winstate->alloc_sz = realloc_sz;
+ }
+
+ winstate->reduced_frame_map[pos] = val;
+}
+
+/*
+ * update_reduced_frame
+ * Update reduced frame info using multi-context NFA pattern matching.
+ *
+ * Maintains multiple NFA contexts simultaneously, one for each potential
+ * match start position. This allows sharing row evaluations across contexts,
+ * avoiding redundant DEFINE clause evaluations when rewinding for SKIP TO
+ * NEXT ROW mode.
+ *
+ * Key optimizations:
+ * - Row evaluations (expensive DEFINE clauses) happen only once per row
+ * - All active contexts share the same evaluation results
+ * - Contexts persist across calls, enabling O(n) DEFINE evaluations
+ */
+static void
+update_reduced_frame(WindowObject winobj, int64 pos)
+{
+ WindowAggState *winstate = winobj->winstate;
+ RPRNFAContext *targetCtx;
+ RPRNFAContext *firstCtx;
+ int64 matchLength = 0;
+ int64 currentPos;
+ int64 startPos;
+ int frameOptions = winstate->frameOptions;
+ bool hasLimitedFrame;
+ int64 frameOffset = 0;
+
+ /*
+ * Check if we have a limited frame (ROWS ... N FOLLOWING).
+ * Each context needs its own frame end based on matchStartRow + offset.
+ */
+ hasLimitedFrame = (frameOptions & FRAMEOPTION_ROWS) &&
+ !(frameOptions & FRAMEOPTION_END_UNBOUNDED_FOLLOWING);
+ if (hasLimitedFrame && winstate->endOffsetValue != 0)
+ frameOffset = DatumGetInt64(winstate->endOffsetValue);
+
+ /*
+ * Case 1: pos is before any existing context's start position.
+ * This means the position was already processed and determined unmatched.
+ * Note: contexts are added at head with increasing positions, so we need
+ * to find the minimum matchStartRow (the oldest context).
+ */
+ {
+ int64 minStartRow = INT64_MAX;
+ for (firstCtx = winstate->nfaContext; firstCtx != NULL; firstCtx = firstCtx->next)
+ {
+ if (firstCtx->matchStartRow < minStartRow)
+ minStartRow = firstCtx->matchStartRow;
+ }
+ if (minStartRow != INT64_MAX && pos < minStartRow)
+ {
+ register_reduced_frame_map(winstate, pos, RF_UNMATCHED);
+ return;
+ }
+ }
+
+ /*
+ * Case 2: Find existing context for this pos, or create new one.
+ */
+ targetCtx = nfa_find_context_for_pos(winstate, pos);
+ if (targetCtx == NULL)
+ {
+ /* No context exists - create one and start fresh */
+ nfa_start_context(winstate, pos);
+ targetCtx = winstate->nfaContext;
+ }
+
+ /*
+ * Determine where to start processing.
+ * If we've already evaluated rows beyond pos, continue from there.
+ */
+ startPos = Max(pos, winstate->nfaLastProcessedRow + 1);
+
+ /*
+ * Process rows until target context completes or we hit boundaries.
+ * Each row evaluation is shared across all active contexts.
+ */
+ for (currentPos = startPos; targetCtx->states != NULL; currentPos++)
+ {
+ bool rowExists;
+ bool anyMatch;
+ RPRNFAContext *ctx;
+
+ /* Evaluate variables for this row - done only once, shared by all contexts */
+ rowExists = nfa_evaluate_row(winobj, currentPos, winstate->nfaVarMatched, &anyMatch);
+
+ /* No more rows in partition? Finalize all contexts */
+ if (!rowExists)
+ {
+ for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next)
+ {
+ if (ctx->states != NULL)
+ nfa_finalize_boundary(winstate, ctx, currentPos - 1);
+ }
+ break;
+ }
+
+ /* Update last processed row */
+ winstate->nfaLastProcessedRow = currentPos;
+
+ /*
+ * Process each active context with this row's evaluation results.
+ * Each context has its own frame boundary based on matchStartRow.
+ */
+ for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next)
+ {
+ int64 ctxFrameEnd;
+
+ /* Skip already-completed contexts */
+ if (ctx->states == NULL)
+ continue;
+
+ /*
+ * Calculate per-context frame end.
+ * For "ROWS BETWEEN CURRENT ROW AND N FOLLOWING", each context's
+ * frame end is matchStartRow + offset + 1 (exclusive).
+ */
+ if (hasLimitedFrame)
+ {
+ ctxFrameEnd = ctx->matchStartRow + frameOffset + 1;
+ if (currentPos >= ctxFrameEnd)
+ {
+ nfa_finalize_boundary(winstate, ctx, ctxFrameEnd - 1);
+ continue;
+ }
+ }
+
+ /* First row of this context must match at least one variable */
+ if (currentPos == ctx->matchStartRow && !anyMatch)
+ {
+ /* Clear states to mark as unmatched */
+ nfa_state_free_list(winstate, ctx->states);
+ ctx->states = NULL;
+ continue;
+ }
+
+ /* Skip if this row is before context's start */
+ if (currentPos < ctx->matchStartRow)
+ continue;
+
+ /* Process states for this context */
+ {
+ RPRNFAState *states = ctx->states;
+ RPRNFAState *state;
+ RPRNFAState *nextState;
+
+ ctx->states = NULL;
+
+ for (state = states; state != NULL; state = nextState)
+ {
+ nextState = state->next;
+ state->next = NULL;
+ nfa_step_single(winstate, ctx, state, winstate->nfaVarMatched, currentPos);
+ }
+ }
+ }
+
+ /*
+ * Create a new context for the next potential start position.
+ * This enables overlapping match detection for SKIP TO NEXT ROW.
+ */
+ if (anyMatch)
+ nfa_start_context(winstate, currentPos + 1);
+
+ /*
+ * Absorb redundant contexts.
+ * At the same elementIndex, if newer context's count <= older context's count,
+ * the newer context can be absorbed (for unbounded quantifiers).
+ * Note: Never absorb targetCtx - it's the context we're trying to complete.
+ */
+ nfa_absorb_contexts(winstate, targetCtx, currentPos);
+
+ /* Check if target context is now complete */
+ if (targetCtx->states == NULL)
+ break;
+ }
+
+ /*
+ * Get match result from target context.
+ */
+ if (targetCtx->matchEndRow >= pos)
+ matchLength = targetCtx->matchEndRow - pos + 1;
+
+ /*
+ * Register reduced frame map based on match result.
+ */
+ if (matchLength <= 0)
+ {
+ register_reduced_frame_map(winstate, pos, RF_UNMATCHED);
+ }
+ else
+ {
+ register_reduced_frame_map(winstate, pos, RF_FRAME_HEAD);
+ for (int64 i = pos + 1; i < pos + matchLength; i++)
+ {
+ register_reduced_frame_map(winstate, i, RF_SKIPPED);
+ }
+ }
+
+ /*
+ * Cleanup contexts based on SKIP mode.
+ */
+ if (matchLength > 0 && winstate->rpSkipTo == ST_PAST_LAST_ROW)
+ {
+ /* Remove all contexts with start <= matchEnd */
+ nfa_remove_contexts_up_to(winstate, pos + matchLength - 1);
+ }
+ else
+ {
+ /* SKIP TO NEXT ROW or no match: just remove the target context */
+ RPRNFAContext *ctx = winstate->nfaContext;
+
+ while (ctx != NULL)
+ {
+ if (ctx == targetCtx)
+ {
+ nfa_unlink_context(winstate, ctx);
+ nfa_context_free(winstate, ctx);
+ break;
+ }
+ ctx = ctx->next;
+ }
+ }
+}
+
+/*
+ * NFA-based pattern matching implementation
+ *
+ * These functions implement direct NFA execution using the compiled
+ * RPRPattern structure, avoiding regex compilation overhead.
+ */
+
+/*
+ * nfa_state_alloc
+ *
+ * Allocate an NFA state, reusing from freeList if available.
+ * freeList is stored in WindowAggState for reuse across match attempts.
+ * Uses flexible array member for counts[].
+ */
+static RPRNFAState *
+nfa_state_alloc(WindowAggState *winstate)
+{
+ RPRNFAState *state;
+ int maxDepth = winstate->rpPattern->maxDepth;
+
+ /* Try to reuse from free list first */
+ if (winstate->nfaStateFree != NULL)
+ {
+ state = winstate->nfaStateFree;
+ winstate->nfaStateFree = state->next;
+ }
+ else
+ {
+ /* Allocate in partition context for proper lifetime */
+ MemoryContext oldContext = MemoryContextSwitchTo(winstate->partcontext);
+ state = palloc(winstate->nfaStateSize);
+ MemoryContextSwitchTo(oldContext);
+ }
+
+ /* initialize state - clear all depth counts */
+ state->next = NULL;
+ state->elemIdx = 0;
+ state->altPriority = 0;
+ /* Initialize all depth counts to 0 using memset */
+ if (maxDepth > 0)
+ memset(state->counts, 0, sizeof(int16) * maxDepth);
+
+ return state;
+}
+
+/*
+ * nfa_state_free
+ *
+ * Return a state to the free list for later reuse.
+ */
+static void
+nfa_state_free(WindowAggState *winstate, RPRNFAState *state)
+{
+ state->next = winstate->nfaStateFree;
+ winstate->nfaStateFree = state;
+}
+
+/*
+ * nfa_state_free_list
+ *
+ * Free all states in a list using pfree.
+ */
+static void
+nfa_state_free_list(WindowAggState *winstate, RPRNFAState *list)
+{
+ RPRNFAState *state;
+
+ while(list != NULL)
+ {
+ state = list;
+ list = list->next;
+ nfa_state_free(winstate, state);
+ }
+}
+
+/*
+ * nfa_states_equal
+ *
+ * Check if two states are equivalent (same elemIdx and counts).
+ */
+static bool
+nfa_states_equal(WindowAggState *winstate, RPRNFAState *s1, RPRNFAState *s2)
+{
+ int maxDepth = winstate->rpPattern->maxDepth;
+
+ if (s1->elemIdx != s2->elemIdx)
+ return false;
+
+ if (maxDepth > 0 &&
+ memcmp(s1->counts, s2->counts, sizeof(int16) * maxDepth) != 0)
+ return false;
+
+ return true;
+}
+
+/*
+ * nfa_add_state_unique
+ *
+ * Add a state to ctx->states at the END, only if no duplicate exists.
+ * Returns true if state was added, false if duplicate found (state is freed).
+ */
+static bool
+nfa_add_state_unique(WindowAggState *winstate, RPRNFAContext *ctx, RPRNFAState *state)
+{
+ RPRNFAState *s;
+ RPRNFAState *prev = NULL;
+ RPRNFAState *tail = NULL;
+
+ /* Check for duplicate and find tail */
+ for (s = ctx->states; s != NULL; s = s->next)
+ {
+ if (nfa_states_equal(winstate, s, state))
+ {
+ /*
+ * Duplicate found - keep lower altPriority for lexical order.
+ * Lower altPriority means earlier alternative in pattern.
+ */
+ if (state->altPriority < s->altPriority)
+ {
+ /* New state has better priority, replace existing */
+ state->next = s->next;
+ if (prev == NULL)
+ ctx->states = state;
+ else
+ prev->next = state;
+ nfa_state_free(winstate, s);
+ return true;
+ }
+ /* Existing state has better/equal priority, discard new */
+ nfa_state_free(winstate, state);
+ return false;
+ }
+ prev = s;
+ tail = s;
+ }
+
+ /* No duplicate, add at end */
+ state->next = NULL;
+ if (tail == NULL)
+ ctx->states = state;
+ else
+ tail->next = state;
+
+ return true;
+}
+
+/*
+ * nfa_state_clone
+ *
+ * Clone a state with given elemIdx, altPriority and counts.
+ * Only copies counts up to elem->depth (not entire maxDepth).
+ * Prepends to the provided list and returns the new list head.
+ */
+static RPRNFAState *
+nfa_state_clone(WindowAggState *winstate, int16 elemIdx, int16 altPriority,
+ int16 *counts, RPRNFAState *list)
+{
+ RPRPattern *pattern = winstate->rpPattern;
+ int maxDepth = pattern->maxDepth;
+ RPRNFAState *state = nfa_state_alloc(winstate);
+
+ state->elemIdx = elemIdx;
+ state->altPriority = altPriority;
+ /* nfa_state_alloc already zeroed all counts, now copy all depth levels */
+ if (counts != NULL && maxDepth > 0)
+ memcpy(state->counts, counts, sizeof(int16) * maxDepth);
+ state->next = list;
+
+ return state;
+}
+
+/*
+ * nfa_add_matched_state
+ *
+ * Record a matched state following SQL standard lexical order preference.
+ * Priority: lower altPriority wins (lexical order), then longer match.
+ */
+static void
+nfa_add_matched_state(WindowAggState *winstate, RPRNFAContext *ctx,
+ RPRNFAState *state, int64 matchEndRow)
+{
+ bool shouldUpdate = false;
+
+ if (ctx->matchedState == NULL)
+ {
+ /* No previous match, always save */
+ shouldUpdate = true;
+ }
+ else if (state->altPriority < ctx->matchedState->altPriority)
+ {
+ /* New state has better lexical order priority */
+ shouldUpdate = true;
+ }
+ else if (state->altPriority == ctx->matchedState->altPriority &&
+ matchEndRow > ctx->matchEndRow)
+ {
+ /* Same priority, but longer match */
+ shouldUpdate = true;
+ }
+
+ if (shouldUpdate)
+ {
+ /* Reuse existing matchedState or allocate from free list */
+ if (ctx->matchedState == NULL)
+ ctx->matchedState = nfa_state_alloc(winstate);
+
+ /* Copy state data */
+ memcpy(ctx->matchedState, state, winstate->nfaStateSize);
+ ctx->matchedState->next = NULL;
+ ctx->matchEndRow = matchEndRow;
+ }
+}
+
+/*
+ * nfa_free_matched_state
+ *
+ * Return matchedState to free list for reuse.
+ */
+static void
+nfa_free_matched_state(WindowAggState *winstate, RPRNFAState *state)
+{
+ if (state != NULL)
+ nfa_state_free(winstate, state);
+}
+
+/*
+ * nfa_evaluate_row
+ *
+ * Evaluate all DEFINE variables for current row.
+ * Returns true if the row exists, false if out of partition.
+ * If row exists, fills varMatched array and sets *anyMatch if any variable matched.
+ * varMatched[i] = true if variable i matched at current row.
+ */
+static bool
+nfa_evaluate_row(WindowObject winobj, int64 pos, bool *varMatched, bool *anyMatchOut)
+{
+ WindowAggState *winstate = winobj->winstate;
+ ExprContext *econtext = winstate->ss.ps.ps_ExprContext;
+ int numDefineVars = list_length(winstate->defineVariableList);
+ ListCell *lc;
+ int varIdx = 0;
+ bool anyMatch = false;
+ TupleTableSlot *slot;
+
+ *anyMatchOut = false;
+
+ /*
+ * Set up slots for current, previous, and next rows.
+ * We don't call get_slots() here to avoid recursion through
+ * row_is_in_frame -> update_reduced_frame -> nfa_match_pattern.
+ */
+
+ /* Current row -> ecxt_outertuple */
+ slot = winstate->temp_slot_1;
+ if (!window_gettupleslot(winobj, pos, slot))
+ return false; /* No row exists */
+ econtext->ecxt_outertuple = slot;
+
+ /* Previous row -> ecxt_scantuple (for PREV) */
+ if (pos > 0)
+ {
+ slot = winstate->prev_slot;
+ if (!window_gettupleslot(winobj, pos - 1, slot))
+ econtext->ecxt_scantuple = winstate->null_slot;
+ else
+ econtext->ecxt_scantuple = slot;
+ }
+ else
+ econtext->ecxt_scantuple = winstate->null_slot;
+
+ /* Next row -> ecxt_innertuple (for NEXT) */
+ slot = winstate->next_slot;
+ if (!window_gettupleslot(winobj, pos + 1, slot))
+ econtext->ecxt_innertuple = winstate->null_slot;
+ else
+ econtext->ecxt_innertuple = slot;
+
+ foreach(lc, winstate->defineClauseList)
+ {
+ ExprState *exprState = (ExprState *) lfirst(lc);
+ Datum result;
+ bool isnull;
+
+ /* Evaluate DEFINE expression */
+ result = ExecEvalExpr(exprState, econtext, &isnull);
+
+ if (!isnull && DatumGetBool(result))
+ {
+ varMatched[varIdx] = true;
+ anyMatch = true;
+ }
+ else
+ {
+ varMatched[varIdx] = false;
+ }
+
+ varIdx++;
+ if (varIdx >= numDefineVars)
+ break;
+ }
+
+ *anyMatchOut = anyMatch;
+ return true; /* Row exists */
+}
+
+/*
+ * nfa_context_alloc
+ *
+ * Allocate an NFA context from free list or palloc.
+ */
+static RPRNFAContext *
+nfa_context_alloc(WindowAggState *winstate)
+{
+ RPRNFAContext *ctx;
+
+ if (winstate->nfaContextFree != NULL)
+ {
+ ctx = winstate->nfaContextFree;
+ winstate->nfaContextFree = ctx->next;
+ }
+ else
+ {
+ /* Allocate in partition context for proper lifetime */
+ MemoryContext oldContext = MemoryContextSwitchTo(winstate->partcontext);
+ ctx = palloc(sizeof(RPRNFAContext));
+ MemoryContextSwitchTo(oldContext);
+ }
+
+ ctx->next = NULL;
+ ctx->prev = NULL;
+ ctx->states = NULL;
+ ctx->matchStartRow = -1;
+ ctx->matchEndRow = -1;
+ ctx->matchedState = NULL;
+
+ return ctx;
+}
+
+/*
+ * nfa_unlink_context
+ *
+ * Remove a context from the doubly-linked active context list.
+ * Updates head (nfaContext) and tail (nfaContextTail) as needed.
+ */
+static void
+nfa_unlink_context(WindowAggState *winstate, RPRNFAContext *ctx)
+{
+ if (ctx->prev != NULL)
+ ctx->prev->next = ctx->next;
+ else
+ winstate->nfaContext = ctx->next; /* was head */
+
+ if (ctx->next != NULL)
+ ctx->next->prev = ctx->prev;
+ else
+ winstate->nfaContextTail = ctx->prev; /* was tail */
+
+ ctx->next = NULL;
+ ctx->prev = NULL;
+}
+
+/*
+ * nfa_context_free
+ *
+ * Return a context to free list. Also frees any states in the context.
+ * Note: Caller must unlink context from active list first using nfa_unlink_context.
+ */
+static void
+nfa_context_free(WindowAggState *winstate, RPRNFAContext *ctx)
+{
+ if (ctx->states != NULL)
+ nfa_state_free_list(winstate, ctx->states);
+ if (ctx->matchedState != NULL)
+ nfa_free_matched_state(winstate, ctx->matchedState);
+
+ ctx->states = NULL;
+ ctx->matchedState = NULL;
+ ctx->next = winstate->nfaContextFree;
+ winstate->nfaContextFree = ctx;
+}
+
+/*
+ * nfa_start_context
+ *
+ * Start a new match context at given position.
+ * Adds context to winstate->nfaContext list.
+ */
+static void
+nfa_start_context(WindowAggState *winstate, int64 startPos)
+{
+ RPRNFAContext *ctx;
+
+ ctx = nfa_context_alloc(winstate);
+ ctx->matchStartRow = startPos;
+ ctx->states = nfa_state_alloc(winstate); /* initial state at elem 0 */
+
+ /* Add to head of active context list (doubly-linked) */
+ ctx->next = winstate->nfaContext;
+ ctx->prev = NULL;
+ if (winstate->nfaContext != NULL)
+ winstate->nfaContext->prev = ctx;
+ else
+ winstate->nfaContextTail = ctx; /* first context becomes tail */
+ winstate->nfaContext = ctx;
+}
+
+/*
+ * nfa_find_context_for_pos
+ *
+ * Find a context with the given start position.
+ * Returns NULL if not found.
+ */
+static RPRNFAContext *
+nfa_find_context_for_pos(WindowAggState *winstate, int64 pos)
+{
+ RPRNFAContext *ctx;
+
+ for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next)
+ {
+ if (ctx->matchStartRow == pos)
+ return ctx;
+ }
+ return NULL;
+}
+
+/*
+ * nfa_remove_contexts_up_to
+ *
+ * Remove all contexts with matchStartRow <= endPos.
+ * Used by SKIP PAST LAST ROW to discard contexts within matched frame.
+ */
+static void
+nfa_remove_contexts_up_to(WindowAggState *winstate, int64 endPos)
+{
+ RPRNFAContext *ctx;
+ RPRNFAContext *next;
+
+ ctx = winstate->nfaContext;
+ while (ctx != NULL)
+ {
+ next = ctx->next;
+ if (ctx->matchStartRow <= endPos)
+ {
+ /* Remove this context */
+ nfa_unlink_context(winstate, ctx);
+ nfa_context_free(winstate, ctx);
+ }
+ ctx = next;
+ }
+}
+
+/*
+ * nfa_absorb_contexts
+ *
+ * Absorb newer contexts into older ones when states are fully covered.
+ * For pattern like A+, if older context (started earlier) has count >= newer
+ * context's count at the same element, the newer context produces subset matches.
+ *
+ * Absorption condition:
+ * - For unbounded quantifiers (max=INT32_MAX): older.counts >= newer.counts
+ * - For bounded quantifiers: older.counts == newer.counts
+ *
+ * Note: List is newest-first, so we check if later nodes (older contexts)
+ * can absorb earlier nodes (newer contexts).
+ */
+static void
+nfa_absorb_contexts(WindowAggState *winstate, RPRNFAContext *excludeCtx, int64 currentPos)
+{
+ RPRPattern *pattern = winstate->rpPattern;
+ RPRNFAContext *ctx;
+ int maxDepth;
+
+ /* Need at least 2 contexts for absorption */
+ if (winstate->nfaContext == NULL || winstate->nfaContext->next == NULL)
+ return;
+
+ if (pattern == NULL)
+ return;
+
+ /*
+ * Only absorb for patterns marked as absorbable during planning.
+ * See computeAbsorbability() in rpr.c for criteria.
+ */
+ if (!pattern->isAbsorbable)
+ return;
+
+ maxDepth = pattern->maxDepth;
+
+ /*
+ * Context absorption: A later context (started at higher row) can be
+ * absorbed by an earlier context if ALL states in the later context
+ * are "covered" by states in the earlier context.
+ *
+ * A later state is covered by an earlier state if:
+ * 1. They are at the same element
+ * 2. For unbounded elements (max == INT32_MAX): earlier.counts[d] >= later.counts[d]
+ * for all depths d
+ * 3. For bounded elements: counts must be exactly equal at all depths
+ *
+ * List is newest-first (head = highest matchStartRow).
+ * We iterate from head (newest) and check if older contexts can absorb it.
+ */
+ ctx = winstate->nfaContext;
+
+ while (ctx != NULL)
+ {
+ RPRNFAContext *nextCtx = ctx->next;
+ RPRNFAContext *older;
+ bool absorbed = false;
+
+ /* Never absorb the excluded context (it's the target being completed) */
+ if (ctx == excludeCtx)
+ {
+ ctx = nextCtx;
+ continue;
+ }
+
+ /* Skip contexts that haven't started processing yet (just created for future row) */
+ if (ctx->matchStartRow > currentPos)
+ {
+ ctx = nextCtx;
+ continue;
+ }
+
+ /*
+ * Handle completed contexts (states == NULL) separately.
+ * A completed context can be absorbed by an older completed context
+ * if the older one has the same or later matchEndRow.
+ */
+ if (ctx->states == NULL)
+ {
+ /* Only completed contexts with valid matchEndRow can be absorbed */
+ if (ctx->matchEndRow < 0)
+ {
+ ctx = nextCtx;
+ continue;
+ }
+
+ /* Check if any older completed context can absorb this one */
+ for (older = ctx->next; older != NULL && !absorbed; older = older->next)
+ {
+ /* Must have started earlier */
+ if (older->matchStartRow >= ctx->matchStartRow)
+ continue;
+
+ /* Older must also be completed with valid matchEndRow */
+ if (older->states != NULL || older->matchEndRow < 0)
+ continue;
+
+ /*
+ * Older context absorbs newer if it has the same or later
+ * matchEndRow, meaning all matches from newer are subsets.
+ */
+ if (older->matchEndRow >= ctx->matchEndRow)
+ {
+ /* Absorb: remove ctx (newer) */
+ nfa_unlink_context(winstate, ctx);
+ nfa_context_free(winstate, ctx);
+ absorbed = true;
+ }
+ }
+
+ ctx = nextCtx;
+ continue;
+ }
+
+ /*
+ * Check if all states in ctx are on absorbable branches.
+ * If any state is on a non-absorbable branch, skip this context.
+ */
+ {
+ RPRNFAState *s;
+ bool allAbsorbable = true;
+
+ for (s = ctx->states; s != NULL && allAbsorbable; s = s->next)
+ {
+ RPRPatternElement *branchFirst;
+
+ /* altPriority is the branch's first element index */
+ if (s->altPriority < 0 || s->altPriority >= pattern->numElements)
+ continue;
+
+ branchFirst = &pattern->elements[s->altPriority];
+ if (!(branchFirst->flags & RPR_ELEM_ABSORBABLE))
+ allAbsorbable = false;
+ }
+
+ if (!allAbsorbable)
+ {
+ ctx = nextCtx;
+ continue;
+ }
+ }
+
+ /*
+ * Check if any older context can absorb this one.
+ * Older contexts have lower matchStartRow.
+ */
+ for (older = ctx->next; older != NULL && !absorbed; older = older->next)
+ {
+ RPRNFAState *laterState;
+ RPRNFAState *earlierState;
+ bool canAbsorb;
+ int laterCount = 0;
+ int earlierCount = 0;
+
+ /* Skip if not started earlier */
+ if (older->matchStartRow >= ctx->matchStartRow)
+ continue;
+
+ /* Skip contexts that haven't started processing yet */
+ if (older->matchStartRow > currentPos)
+ continue;
+
+ /*
+ * For in-progress ctx, older must also be in-progress.
+ * (Completed older contexts are handled above for completed ctx.)
+ */
+ if (older->states == NULL)
+ continue;
+
+ /* Count states in both contexts */
+ for (laterState = ctx->states; laterState != NULL; laterState = laterState->next)
+ laterCount++;
+ for (earlierState = older->states; earlierState != NULL; earlierState = earlierState->next)
+ earlierCount++;
+
+ /* Must have same number of states (same structure) */
+ if (laterCount != earlierCount)
+ continue;
+
+ /*
+ * Check if ALL states in ctx (later) are covered by states in older.
+ * Both must have the same set of element indices.
+ */
+ canAbsorb = true;
+ for (laterState = ctx->states; laterState != NULL && canAbsorb; laterState = laterState->next)
+ {
+ bool found = false;
+
+ for (earlierState = older->states; earlierState != NULL && !found; earlierState = earlierState->next)
+ {
+ RPRPatternElement *elem;
+ bool countsMatch;
+ int d;
+
+ /* Must be at same element */
+ if (earlierState->elemIdx != laterState->elemIdx)
+ continue;
+
+ /* Must be on same branch (same altPriority) */
+ if (earlierState->altPriority != laterState->altPriority)
+ continue;
+
+ /* Handle invalid element index (terminal state) */
+ if (laterState->elemIdx < 0 || laterState->elemIdx >= pattern->numElements)
+ {
+ found = true;
+ break;
+ }
+
+ elem = &pattern->elements[laterState->elemIdx];
+ countsMatch = true;
+
+ if (elem->max == INT32_MAX)
+ {
+ /* Unbounded: earlier.count >= later.count at all depths */
+ for (d = 0; d <= maxDepth && countsMatch; d++)
+ {
+ if (earlierState->counts[d] < laterState->counts[d])
+ countsMatch = false;
+ }
+ }
+ else
+ {
+ /* Bounded: counts must be exactly equal at all depths */
+ for (d = 0; d <= maxDepth && countsMatch; d++)
+ {
+ if (earlierState->counts[d] != laterState->counts[d])
+ countsMatch = false;
+ }
+ }
+
+ if (countsMatch)
+ found = true;
+ }
+
+ if (!found)
+ canAbsorb = false;
+ }
+
+ if (canAbsorb)
+ {
+ /* Absorb: remove ctx (newer) */
+ nfa_unlink_context(winstate, ctx);
+ nfa_context_free(winstate, ctx);
+ absorbed = true;
+ }
+ }
+
+ ctx = nextCtx;
+ }
+}
+
+/*
+ * nfa_finalize_boundary
+ *
+ * Finalize NFA states at partition/frame boundary.
+ * Sets all varMatched to false and processes remaining states.
+ */
+static void
+nfa_finalize_boundary(WindowAggState *winstate, RPRNFAContext *ctx, int64 matchEndPos)
+{
+ RPRNFAState *states = ctx->states;
+ RPRNFAState *state;
+ RPRNFAState *nextState;
+ int numVars = list_length(winstate->defineVariableList);
+
+ ctx->states = NULL;
+
+ for (int i = 0; i < numVars; i++)
+ winstate->nfaVarMatched[i] = false;
+
+ for (state = states; state != NULL; state = nextState)
+ {
+ nextState = state->next;
+ state->next = NULL;
+ nfa_step_single(winstate, ctx, state, winstate->nfaVarMatched, matchEndPos);
+ }
+}
+
+/*
+ * nfa_step_single
+ *
+ * Process one state through NFA for one row.
+ * New states are added to ctx->states.
+ * Match (FIN) is recorded in ctx->matchedState.
+ * When FIN is reached by matching (not skipping), matchEndRow is updated.
+ */
+static void
+nfa_step_single(WindowAggState *winstate, RPRNFAContext *ctx,
+ RPRNFAState *state, bool *varMatched, int64 currentPos)
+{
+ RPRPattern *pattern = winstate->rpPattern;
+ RPRPatternElement *elements = pattern->elements;
+ RPRNFAState *pending = state; /* states to process in current row */
+
+ while (pending != NULL)
+ {
+ RPRPatternElement *elem;
+ bool matched;
+ int16 count;
+ int depth;
+
+ state = pending;
+ pending = pending->next;
+ state->next = NULL;
+
+ Assert(state->elemIdx >= 0 && state->elemIdx < pattern->numElements);
+ elem = &elements[state->elemIdx];
+ depth = elem->depth;
+
+ if (RPRElemIsVar(elem))
+ {
+ /*
+ * Variable: check if it matches current row.
+ * With DEFINE-first ordering, varId < numDefines has DEFINE expr,
+ * varId >= numDefines defaults to TRUE.
+ */
+ int numDefines = list_length(winstate->defineVariableList);
+
+ if (elem->varId >= numDefines)
+ matched = true; /* Not defined in DEFINE, defaults to TRUE */
+ else
+ matched = varMatched[elem->varId];
+
+ count = state->counts[depth];
+
+ if (matched)
+ {
+ count++;
+
+ if (elem->max != RPR_QUANTITY_INF && count > elem->max)
+ {
+ nfa_state_free(winstate, state);
+ continue;
+ }
+
+ state->counts[depth] = count;
+
+ /* Can repeat more? Clone for staying */
+ if (elem->max == RPR_QUANTITY_INF || count < elem->max)
+ {
+ RPRNFAState *clone = nfa_state_clone(winstate, state->elemIdx,
+ state->altPriority,
+ state->counts, NULL);
+ nfa_add_state_unique(winstate, ctx, clone);
+ }
+
+ /* Satisfied? Advance to next element */
+ if (count >= elem->min)
+ {
+ RPRPatternElement *nextElem;
+
+ state->counts[depth] = 0;
+ state->elemIdx = elem->next;
+ nextElem = &elements[state->elemIdx];
+
+ if (RPRElemIsFin(nextElem))
+ {
+ /* Match ends at current row since we matched */
+ nfa_add_matched_state(winstate, ctx, state, currentPos);
+ nfa_state_free(winstate, state);
+ }
+ else if (RPRElemIsEnd(nextElem))
+ {
+ /*
+ * END is epsilon transition - process immediately on same row.
+ * This ensures match end position is recorded at the row where
+ * the last VAR matched, not the next row.
+ */
+ state->next = pending;
+ pending = state;
+ }
+ else
+ {
+ /*
+ * VAR, ALT - wait for next row. ALT dispatches to VARs that
+ * need input, so it must wait for the next row after VAR
+ * consumed the current row.
+ */
+ nfa_add_state_unique(winstate, ctx, state);
+ }
+ }
+ else
+ {
+ nfa_state_free(winstate, state);
+ }
+ }
+ else
+ {
+ /* Not matched: can we skip? */
+ if (count >= elem->min)
+ {
+ RPRPatternElement *nextElem;
+
+ state->counts[depth] = 0;
+ state->elemIdx = elem->next;
+ nextElem = &elements[state->elemIdx];
+
+ if (RPRElemIsFin(nextElem))
+ {
+ /* Match ends at previous row since current didn't match */
+ nfa_add_matched_state(winstate, ctx, state, currentPos - 1);
+ nfa_state_free(winstate, state);
+ }
+ else if (RPRElemIsVar(nextElem))
+ {
+ /*
+ * Current row was NOT consumed (skip case), so next VAR
+ * must be tried on the SAME row via pending list
+ */
+ state->next = pending;
+ pending = state;
+ }
+ else
+ {
+ state->next = pending;
+ pending = state;
+ }
+ }
+ else
+ {
+ nfa_state_free(winstate, state);
+ }
+ }
+ }
+ else if (RPRElemIsFin(elem))
+ {
+ /* Already at FIN - match ends at current row */
+ nfa_add_matched_state(winstate, ctx, state, currentPos);
+ nfa_state_free(winstate, state);
+ }
+ else if (RPRElemIsAlt(elem))
+ {
+ RPRElemIdx altIdx = elem->next;
+ bool first = true;
+
+ /*
+ * ALT doesn't consume a row - it's just a dispatch point.
+ * All branches should be evaluated on the CURRENT row.
+ * Set altPriority to branch's elemIdx for lexical order tracking.
+ */
+ while (altIdx >= 0 && altIdx < pattern->numElements)
+ {
+ RPRPatternElement *altElem = &elements[altIdx];
+
+ if (first)
+ {
+ state->elemIdx = altIdx;
+ state->altPriority = altIdx; /* lexical order */
+ state->next = pending;
+ pending = state;
+ first = false;
+ }
+ else
+ {
+ pending = nfa_state_clone(winstate, altIdx, altIdx,
+ state->counts, pending);
+ }
+
+ altIdx = altElem->jump;
+ }
+
+ if (first)
+ nfa_state_free(winstate, state);
+ }
+ else if (RPRElemIsEnd(elem))
+ {
+ count = state->counts[depth] + 1;
+
+ if (count < elem->min)
+ {
+ /*
+ * Haven't reached minimum yet - must loop back.
+ * Add to ctx->states (next row) not pending (same row).
+ */
+ state->counts[depth] = count;
+ for (int d = depth + 1; d < pattern->maxDepth; d++)
+ state->counts[d] = 0;
+ state->elemIdx = elem->jump;
+ nfa_add_state_unique(winstate, ctx, state);
+ }
+ else if (elem->max != RPR_QUANTITY_INF && count >= elem->max)
+ {
+ /* Reached maximum - must exit, continue processing */
+ state->counts[depth] = 0;
+ state->elemIdx = elem->next;
+ state->next = pending;
+ pending = state;
+ }
+ else
+ {
+ /*
+ * Between min and max - can exit or continue.
+ * Exit state continues processing (pending).
+ * Loop state waits for next row (ctx->states).
+ */
+ RPRNFAState *exitState = nfa_state_clone(winstate, elem->next,
+ state->altPriority,
+ state->counts, pending);
+ exitState->counts[depth] = 0;
+ pending = exitState;
+
+ state->counts[depth] = count;
+ for (int d = depth + 1; d < pattern->maxDepth; d++)
+ state->counts[d] = 0;
+ state->elemIdx = elem->jump;
+ nfa_add_state_unique(winstate, ctx, state);
+ }
+ }
+ else
+ {
+ state->elemIdx = elem->next;
+ state->next = pending;
+ pending = state;
+ }
+ }
+}
+
diff --git a/src/backend/utils/adt/windowfuncs.c b/src/backend/utils/adt/windowfuncs.c
index 78b7f05aba2..723ebc91909 100644
--- a/src/backend/utils/adt/windowfuncs.c
+++ b/src/backend/utils/adt/windowfuncs.c
@@ -37,11 +37,19 @@ typedef struct
int64 remainder; /* (total rows) % (bucket num) */
} ntile_context;
+/*
+ * rpr process information.
+ * Used for AFTER MATCH SKIP PAST LAST ROW
+ */
+typedef struct SkipContext
+{
+ int64 pos; /* last row absolute position */
+} SkipContext;
+
static bool rank_up(WindowObject winobj);
static Datum leadlag_common(FunctionCallInfo fcinfo,
bool forward, bool withoffset, bool withdefault);
-
/*
* utility routine for *_rank functions.
*/
@@ -683,7 +691,7 @@ window_last_value(PG_FUNCTION_ARGS)
WinCheckAndInitializeNullTreatment(winobj, true, fcinfo);
result = WinGetFuncArgInFrame(winobj, 0,
- 0, WINDOW_SEEK_TAIL, true,
+ 0, WINDOW_SEEK_TAIL, false,
&isnull, NULL);
if (isnull)
PG_RETURN_NULL();
@@ -724,3 +732,25 @@ window_nth_value(PG_FUNCTION_ARGS)
PG_RETURN_DATUM(result);
}
+
+/*
+ * prev
+ * Dummy function to invoke RPR's navigation operator "PREV".
+ * This is *not* a window function.
+ */
+Datum
+window_prev(PG_FUNCTION_ARGS)
+{
+ PG_RETURN_DATUM(PG_GETARG_DATUM(0));
+}
+
+/*
+ * next
+ * Dummy function to invoke RPR's navigation operation "NEXT".
+ * This is *not* a window function.
+ */
+Datum
+window_next(PG_FUNCTION_ARGS)
+{
+ PG_RETURN_DATUM(PG_GETARG_DATUM(0));
+}
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 2ac69bf2df5..124884fce13 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -10853,6 +10853,12 @@
{ oid => '3114', descr => 'fetch the Nth row value',
proname => 'nth_value', prokind => 'w', prorettype => 'anyelement',
proargtypes => 'anyelement int4', prosrc => 'window_nth_value' },
+{ oid => '8126', descr => 'previous value',
+ proname => 'prev', provolatile => 's', prorettype => 'anyelement',
+ proargtypes => 'anyelement', prosrc => 'window_prev' },
+{ oid => '8127', descr => 'next value',
+ proname => 'next', provolatile => 's', prorettype => 'anyelement',
+ proargtypes => 'anyelement', prosrc => 'window_next' },
# functions for range types
{ oid => '3832', descr => 'I/O',
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index f8053d9e572..89139583855 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -41,6 +41,7 @@
#include "nodes/plannodes.h"
#include "nodes/tidbitmap.h"
#include "partitioning/partdefs.h"
+#include "regex/regex.h"
#include "storage/condition_variable.h"
#include "utils/hsearch.h"
#include "utils/queryenvironment.h"
@@ -2512,6 +2513,39 @@ typedef enum WindowAggStatus
* tuples during spool */
} WindowAggStatus;
+#define RF_NOT_DETERMINED 0
+#define RF_FRAME_HEAD 1
+#define RF_SKIPPED 2
+#define RF_UNMATCHED 3
+
+/*
+ * RPRNFAState - single NFA state for pattern matching
+ *
+ * counts[] tracks repetition counts at each nesting depth.
+ * altPriority tracks lexical order for alternation (lower = earlier alternative).
+ */
+typedef struct RPRNFAState
+{
+ struct RPRNFAState *next; /* next state in linked list */
+ int16 elemIdx; /* current pattern element index */
+ int16 altPriority; /* lexical order priority (lower = preferred) */
+ int16 counts[FLEXIBLE_ARRAY_MEMBER]; /* repetition counts by depth */
+} RPRNFAState;
+
+/*
+ * RPRNFAContext - context for NFA pattern matching execution
+ */
+typedef struct RPRNFAContext
+{
+ struct RPRNFAContext *next; /* next context in linked list */
+ struct RPRNFAContext *prev; /* previous context (for reverse traversal) */
+ RPRNFAState *states; /* active states (linked list) */
+
+ int64 matchStartRow; /* row where match started */
+ int64 matchEndRow; /* row where match ended (-1 = no match) */
+ RPRNFAState *matchedState; /* FIN state for greedy fallback (cloned) */
+} RPRNFAContext;
+
typedef struct WindowAggState
{
ScanState ss; /* its first field is NodeTag */
@@ -2571,6 +2605,24 @@ typedef struct WindowAggState
int64 groupheadpos; /* current row's peer group head position */
int64 grouptailpos; /* " " " " tail position (group end+1) */
+ /* these fields are used in Row pattern recognition: */
+ RPSkipTo rpSkipTo; /* Row Pattern Skip To type */
+ struct RPRPattern *rpPattern; /* compiled pattern for NFA execution */
+ List *defineVariableList; /* list of row pattern definition
+ * variables (list of String) */
+ List *defineClauseList; /* expression for row pattern definition
+ * search conditions ExprState list */
+ List *defineInitial; /* list of row pattern definition variable
+ * initials (list of String) */
+ RPRNFAContext *nfaContext; /* active matching contexts (head) */
+ RPRNFAContext *nfaContextTail; /* tail of active contexts (for reverse traversal) */
+ RPRNFAContext *nfaContextPending; /* matched but awaiting earlier starts */
+ RPRNFAContext *nfaContextFree; /* recycled NFA context nodes */
+ RPRNFAState *nfaStateFree; /* recycled NFA state nodes */
+ Size nfaStateSize; /* pre-calculated RPRNFAState size */
+ bool *nfaVarMatched; /* per-row cache: varMatched[varId] for varId < numDefines */
+ int64 nfaLastProcessedRow; /* last row processed by NFA (-1 = none) */
+
MemoryContext partcontext; /* context for partition-lifespan data */
MemoryContext aggcontext; /* shared context for aggregate working data */
MemoryContext curaggcontext; /* current aggregate's working data */
@@ -2598,6 +2650,18 @@ typedef struct WindowAggState
TupleTableSlot *agg_row_slot;
TupleTableSlot *temp_slot_1;
TupleTableSlot *temp_slot_2;
+
+ /* temporary slots for RPR */
+ TupleTableSlot *prev_slot; /* PREV row navigation operator */
+ TupleTableSlot *next_slot; /* NEXT row navigation operator */
+ TupleTableSlot *null_slot; /* all NULL slot */
+
+ /*
+ * Each byte corresponds to a row positioned at absolute its pos in
+ * partition. See above definition for RF_*. Used for RPR.
+ */
+ char *reduced_frame_map;
+ int64 alloc_sz; /* size of the map */
} WindowAggState;
/* ----------------
--
2.43.0