From ba1487af610372eb766eb6e2982eccbb64a086f1 Mon Sep 17 00:00:00 2001 From: Tatsuo Ishii Date: Sun, 18 Jan 2026 21:01:37 +0900 Subject: [PATCH v40 5/8] Row pattern recognition patch (executor and commands). --- src/backend/commands/explain.c | 279 ++++ src/backend/executor/nodeWindowAgg.c | 1914 +++++++++++++++++++++++++- src/backend/utils/adt/windowfuncs.c | 25 +- src/include/catalog/pg_proc.dat | 6 + src/include/nodes/execnodes.h | 93 ++ 5 files changed, 2306 insertions(+), 11 deletions(-) diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c index b7bb111688c..ba114bd491c 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 depth (content is at depth+1) */ + 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); + } + } } /* @@ -3499,6 +3642,7 @@ show_windowagg_info(WindowAggState *winstate, ExplainState *es) { char *maxStorageType; int64 maxSpaceUsed; + WindowAgg *wagg = (WindowAgg *) winstate->ss.ps.plan; Tuplestorestate *tupstore = winstate->buffer; @@ -3511,6 +3655,141 @@ show_windowagg_info(WindowAggState *winstate, ExplainState *es) tuplestore_get_stats(tupstore, &maxStorageType, &maxSpaceUsed); show_storage_info(maxStorageType, maxSpaceUsed, es); + + /* Show NFA statistics for Row Pattern Recognition */ + if (wagg->rpPattern != NULL) + { + if (es->format != EXPLAIN_FORMAT_TEXT) + { + ExplainPropertyInteger("NFA States Peak", NULL, winstate->nfaStatesMax, es); + ExplainPropertyInteger("NFA States Total", NULL, winstate->nfaStatesTotalCreated, es); + ExplainPropertyInteger("NFA States Merged", NULL, winstate->nfaStatesMerged, es); + ExplainPropertyInteger("NFA Contexts Peak", NULL, winstate->nfaContextsMax, es); + ExplainPropertyInteger("NFA Contexts Total", NULL, winstate->nfaContextsTotalCreated, es); + ExplainPropertyInteger("NFA Contexts Absorbed", NULL, winstate->nfaContextsAbsorbed, es); + ExplainPropertyInteger("NFA Contexts Skipped", NULL, winstate->nfaContextsSkipped, es); + ExplainPropertyInteger("NFA Contexts Pruned", NULL, winstate->nfaContextsPruned, es); + ExplainPropertyInteger("NFA Matched", NULL, winstate->nfaMatchesSucceeded, es); + ExplainPropertyInteger("NFA Mismatched", NULL, winstate->nfaMatchesFailed, es); + if (winstate->nfaMatchesSucceeded > 0) + { + ExplainPropertyInteger("NFA Match Length Min", NULL, winstate->nfaMatchLen.min, es); + ExplainPropertyInteger("NFA Match Length Max", NULL, winstate->nfaMatchLen.max, es); + ExplainPropertyInteger("NFA Match Length Avg", NULL, + (int64) ((winstate->nfaMatchLen.total + winstate->nfaMatchesSucceeded / 2) / winstate->nfaMatchesSucceeded), + es); + } + if (winstate->nfaMatchesFailed > 0) + { + ExplainPropertyInteger("NFA Mismatch Length Min", NULL, winstate->nfaFailLen.min, es); + ExplainPropertyInteger("NFA Mismatch Length Max", NULL, winstate->nfaFailLen.max, es); + ExplainPropertyInteger("NFA Mismatch Length Avg", NULL, + (int64) ((winstate->nfaFailLen.total + winstate->nfaMatchesFailed / 2) / winstate->nfaMatchesFailed), + es); + } + if (winstate->nfaContextsAbsorbed > 0) + { + ExplainPropertyInteger("NFA Absorbed Length Min", NULL, winstate->nfaAbsorbedLen.min, es); + ExplainPropertyInteger("NFA Absorbed Length Max", NULL, winstate->nfaAbsorbedLen.max, es); + ExplainPropertyInteger("NFA Absorbed Length Avg", NULL, + (int64) ((winstate->nfaAbsorbedLen.total + winstate->nfaContextsAbsorbed / 2) / winstate->nfaContextsAbsorbed), + es); + } + if (winstate->nfaContextsSkipped > 0) + { + ExplainPropertyInteger("NFA Skipped Length Min", NULL, winstate->nfaSkippedLen.min, es); + ExplainPropertyInteger("NFA Skipped Length Max", NULL, winstate->nfaSkippedLen.max, es); + ExplainPropertyInteger("NFA Skipped Length Avg", NULL, + (int64) ((winstate->nfaSkippedLen.total + winstate->nfaContextsSkipped / 2) / winstate->nfaContextsSkipped), + es); + } + } + else + { + ExplainIndentText(es); + appendStringInfo(es->str, + "NFA States: " INT64_FORMAT " peak, " INT64_FORMAT " total, " INT64_FORMAT " merged\n", + winstate->nfaStatesMax, + winstate->nfaStatesTotalCreated, + winstate->nfaStatesMerged); + ExplainIndentText(es); + appendStringInfo(es->str, + "NFA Contexts: " INT64_FORMAT " peak, " INT64_FORMAT " total, " INT64_FORMAT " pruned\n", + winstate->nfaContextsMax, + winstate->nfaContextsTotalCreated, + winstate->nfaContextsPruned); + ExplainIndentText(es); + appendStringInfo(es->str, "NFA: "); + if (winstate->nfaMatchesSucceeded > 0) + { + double avgLen = (double) winstate->nfaMatchLen.total / winstate->nfaMatchesSucceeded; + appendStringInfo(es->str, + INT64_FORMAT " matched (len " INT64_FORMAT "/" INT64_FORMAT "/%.0f)", + winstate->nfaMatchesSucceeded, + winstate->nfaMatchLen.min, + winstate->nfaMatchLen.max, + avgLen); + } + else + { + appendStringInfo(es->str, "0 matched"); + } + if (winstate->nfaMatchesFailed > 0) + { + double avgFail = (double) winstate->nfaFailLen.total / winstate->nfaMatchesFailed; + appendStringInfo(es->str, + ", " INT64_FORMAT " mismatched (len " INT64_FORMAT "/" INT64_FORMAT "/%.0f)", + winstate->nfaMatchesFailed, + winstate->nfaFailLen.min, + winstate->nfaFailLen.max, + avgFail); + } + else + { + appendStringInfo(es->str, ", 0 mismatched"); + } + appendStringInfoChar(es->str, '\n'); + + /* Show absorbed and skipped context length statistics */ + if (winstate->nfaContextsAbsorbed > 0 || winstate->nfaContextsSkipped > 0) + { + ExplainIndentText(es); + appendStringInfo(es->str, "NFA: "); + + if (winstate->nfaContextsAbsorbed > 0) + { + double avgAbsorbed = (double) winstate->nfaAbsorbedLen.total / winstate->nfaContextsAbsorbed; + appendStringInfo(es->str, + INT64_FORMAT " absorbed (len " INT64_FORMAT "/" INT64_FORMAT "/%.0f)", + winstate->nfaContextsAbsorbed, + winstate->nfaAbsorbedLen.min, + winstate->nfaAbsorbedLen.max, + avgAbsorbed); + } + else + { + appendStringInfo(es->str, "0 absorbed"); + } + + if (winstate->nfaContextsSkipped > 0) + { + double avgSkipped = (double) winstate->nfaSkippedLen.total / winstate->nfaContextsSkipped; + appendStringInfo(es->str, + ", " INT64_FORMAT " skipped (len " INT64_FORMAT "/" INT64_FORMAT "/%.0f)", + winstate->nfaContextsSkipped, + winstate->nfaSkippedLen.min, + winstate->nfaSkippedLen.max, + avgSkipped); + } + else + { + appendStringInfo(es->str, ", 0 skipped"); + } + + appendStringInfoChar(es->str, '\n'); + } + } + } } /* diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c index d9b64b0f465..6002ca05c90 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,46 @@ 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, int32 *counts); +static bool nfa_evaluate_row(WindowObject winobj, int64 pos, bool *varMatched); +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 RPRNFAContext *nfa_start_context(WindowAggState *winstate, int64 startPos); +static void nfa_step(WindowAggState *winstate, RPRNFAContext *ctx, + bool *varMatched, int64 pos); +static void nfa_finalize_all_contexts(WindowAggState *winstate, int64 lastPos); +static void nfa_process_context(WindowAggState *winstate, RPRNFAContext *ctx, + int64 currentPos, bool hasLimitedFrame, int64 frameOffset); +static void nfa_step_single(WindowAggState *winstate, RPRNFAContext *ctx, + RPRNFAState *state, bool *varMatched, int64 currentPos); +static RPRNFAContext *nfa_find_context_for_pos(WindowAggState *winstate, int64 pos); +static void nfa_remove_contexts_up_to(WindowAggState *winstate, int64 endPos, + RPRNFAContext *excludeCtx); +static void nfa_cleanup_dead_contexts(WindowAggState *winstate, RPRNFAContext *excludeCtx); +static void nfa_absorb_contexts(WindowAggState *winstate, RPRNFAContext *excludeCtx, int64 currentPos); +static void update_nfa_length_stats(int64 count, NFALengthStats *stats, int64 newLen); +static void record_nfa_context_failure(WindowAggState *winstate, int64 failedLen); /* * Not null info bit array consists of 2-bit items @@ -817,6 +874,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 +889,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 +964,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 +1034,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 +1055,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 +1077,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 +1152,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 +1376,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 +1598,15 @@ 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; + winstate->nfaStatesActive = 0; + winstate->nfaContextsActive = 0; } /* @@ -2237,6 +2380,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 +2493,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 +2670,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 +2771,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 +2967,65 @@ 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(int32) * node->rpPattern->maxDepth; + } + + /* Set up row pattern recognition DEFINE clause */ + 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 +3034,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 +3196,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 +3558,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); @@ -3337,6 +3676,7 @@ ignorenulls_getfuncarginframe(WindowObject winobj, int argno, int notnull_offset; int notnull_relpos; int forward; + int num_reduced_frame; Assert(WindowObjectIsValid(winobj)); winstate = winobj->winstate; @@ -3365,6 +3705,12 @@ ignorenulls_getfuncarginframe(WindowObject winobj, int argno, /* 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; mark_pos = 0; /* keep compiler quiet */ @@ -3380,6 +3726,35 @@ ignorenulls_getfuncarginframe(WindowObject winobj, int argno, * Get the next nonnull value in the frame, moving forward or backward * until we find a value or reach the frame's end. */ + + /* + * Check whether current row is in reduced frame. + */ + num_reduced_frame = row_is_in_reduced_frame(winobj, winstate->frameheadpos); + if (num_reduced_frame < 0) /* unmatched or skipped row */ + goto out_of_frame; + else if (num_reduced_frame > 0) /* the first row of the reduced frame */ + { + /* + * Early check if row could be out of reduced frame. When RPR is + * enabled, EXCUDE clause cannot be specified and the frame is always + * contiguous. So we can do the check followings safely. Note, + * however, it is possible that a row is out of reduced frame if + * there's a NULL in the middle. So we need to check it in the + * following do lopp. + */ + if (seektype == WINDOW_SEEK_HEAD && relpos >= num_reduced_frame) + goto out_of_frame; + if (seektype == WINDOW_SEEK_TAIL) + { + if (notnull_relpos >= num_reduced_frame) + goto out_of_frame; + + /* not out of reduced frame. Set abspos as a starting point */ + abs_pos = winstate->frameheadpos + num_reduced_frame - 1; + } + } + do { int inframe; @@ -3441,6 +3816,16 @@ ignorenulls_getfuncarginframe(WindowObject winobj, int argno, } advance: abs_pos += forward; + if (rpr_is_defined(winstate)) + { + /* + * Check whether we are still in the reduced frame. (also check + * if we succeeded in getting the target row). + */ + num_reduced_frame--; + if (num_reduced_frame <= 0 && notnull_offset <= notnull_relpos) + goto out_of_frame; + } } while (notnull_offset <= notnull_relpos); if (set_mark) @@ -3922,8 +4307,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 +4317,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 +4425,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 +4510,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 +4536,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 +4573,1454 @@ 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; + 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. + * Head is the oldest context (lowest matchStartRow) since contexts are + * added at tail with increasing positions. + */ + if (winstate->nfaContext != NULL && + pos < winstate->nfaContext->matchStartRow) + { + 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. If pos is already processed, it means this row + * was already determined to be unmatched or skipped - no need to + * reprocess. + */ + if (pos <= winstate->nfaLastProcessedRow) + { + register_reduced_frame_map(winstate, pos, RF_UNMATCHED); + return; + } + /* Not yet processed - create new context and start fresh */ + targetCtx = nfa_start_context(winstate, pos); + } + else if (targetCtx->states == NULL) + { + /* Context already completed - skip to result registration */ + goto register_result; + } + + /* + * Determine where to start processing. + * Usually nfaLastProcessedRow+1 >= pos since contexts are created at + * currentPos+1 during processing. However, pos can exceed this when + * rows are skipped (e.g., unmatched rows don't update nfaLastProcessedRow). + */ + 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; + RPRNFAContext *ctx; + + /* Evaluate variables for this row - done only once, shared by all contexts */ + rowExists = nfa_evaluate_row(winobj, currentPos, winstate->nfaVarMatched); + + /* No more rows in partition? Finalize all contexts */ + if (!rowExists) + { + nfa_finalize_all_contexts(winstate, currentPos - 1); + /* Clean up dead contexts from finalization */ + nfa_cleanup_dead_contexts(winstate, targetCtx); + /* Absorb contexts at partition boundary */ + if (winstate->rpSkipTo == ST_PAST_LAST_ROW && !hasLimitedFrame && + winstate->rpPattern->isAbsorbable) + nfa_absorb_contexts(winstate, NULL, currentPos - 1); + break; + } + + /* Update last processed row */ + winstate->nfaLastProcessedRow = currentPos; + + /* Process each active context with this row's evaluation results */ + for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) + nfa_process_context(winstate, ctx, currentPos, hasLimitedFrame, frameOffset); + + /* + * Create a new context for the next potential start position. + * This enables overlapping match detection for SKIP TO NEXT ROW. + */ + nfa_start_context(winstate, currentPos + 1); + + /* + * Clean up dead contexts (failed with no active states and no match). + * This removes contexts that failed during processing and counts them + * appropriately as pruned or mismatched. + */ + nfa_cleanup_dead_contexts(winstate, targetCtx); + + /* + * Absorb redundant contexts using simplified algorithm. + * Only compares absorbable element counts (e.g., A+ or (A B)+). + */ + if (winstate->rpSkipTo == ST_PAST_LAST_ROW && !hasLimitedFrame && + winstate->rpPattern->isAbsorbable) + nfa_absorb_contexts(winstate, targetCtx, currentPos); + + /* Check if target context is now complete */ + if (targetCtx->states == NULL) + break; + } + +register_result: + Assert(pos == targetCtx->matchStartRow); + + /* + * Register reduced frame map based on match result. + */ + if (targetCtx->matchEndRow < targetCtx->matchStartRow) + { + int64 mismatchLen = targetCtx->lastProcessedRow - targetCtx->matchStartRow + 1; + + register_reduced_frame_map(winstate, targetCtx->matchStartRow, RF_UNMATCHED); + record_nfa_context_failure(winstate, mismatchLen); + } + else + { + int64 matchLen = targetCtx->matchEndRow - targetCtx->matchStartRow + 1; + + register_reduced_frame_map(winstate, targetCtx->matchStartRow, RF_FRAME_HEAD); + for (int64 i = targetCtx->matchStartRow + 1; i <= targetCtx->matchEndRow; i++) + { + register_reduced_frame_map(winstate, i, RF_SKIPPED); + } + winstate->nfaMatchesSucceeded++; + update_nfa_length_stats(winstate->nfaMatchesSucceeded, + &winstate->nfaMatchLen, + matchLen); + } + + /* + * Cleanup contexts based on SKIP mode. + */ + if (targetCtx->matchEndRow >= targetCtx->matchStartRow && + winstate->rpSkipTo == ST_PAST_LAST_ROW) + { + /* Remove all contexts with start <= matchEnd, excluding targetCtx from skip count */ + nfa_remove_contexts_up_to(winstate, targetCtx->matchEndRow, targetCtx); + } + else + { + /* SKIP TO NEXT ROW or no match: just remove the target context */ + nfa_context_free(winstate, targetCtx); + } +} + +/* + * 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; + + /* 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 */ + state = MemoryContextAlloc(winstate->partcontext, winstate->nfaStateSize); + } + + /* Initialize entire state to zero */ + memset(state, 0, winstate->nfaStateSize); + + /* Update statistics */ + winstate->nfaStatesActive++; + winstate->nfaStatesTotalCreated++; + if (winstate->nfaStatesActive > winstate->nfaStatesMax) + winstate->nfaStatesMax = winstate->nfaStatesActive; + + return state; +} + +/* + * nfa_state_free + * + * Return a state to the free list for later reuse. + */ +static void +nfa_state_free(WindowAggState *winstate, RPRNFAState *state) +{ + winstate->nfaStatesActive--; + 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(int32) * 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). + * Earlier states have lower altPriority (lexical order), so existing wins. + */ +static bool +nfa_add_state_unique(WindowAggState *winstate, RPRNFAContext *ctx, RPRNFAState *state) +{ + RPRNFAState *s; + 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 - existing has better lexical order, discard new */ + nfa_state_free(winstate, state); + winstate->nfaStatesMerged++; + return false; + } + 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. + * Caller is responsible for linking the returned state. + */ +static RPRNFAState * +nfa_state_clone(WindowAggState *winstate, int16 elemIdx, int16 altPriority, + int32 *counts) +{ + RPRPattern *pattern = winstate->rpPattern; + int maxDepth = pattern->maxDepth; + RPRNFAState *state = nfa_state_alloc(winstate); + + state->elemIdx = elemIdx; + state->altPriority = altPriority; + if (counts != NULL && maxDepth > 0) + memcpy(state->counts, counts, sizeof(int32) * maxDepth); + + return state; +} + +/* + * nfa_add_matched_state + * + * Record a matched state following SQL standard semantics. + * Lexical order (lower altPriority) wins first. Among same lexical order, + * longer match wins (greedy). + */ +static void +nfa_add_matched_state(WindowAggState *winstate, RPRNFAContext *ctx, + RPRNFAState *state, int64 matchEndRow) +{ + bool shouldUpdate = false; + + if (ctx->matchedState == NULL) + shouldUpdate = true; + else if (state->altPriority < ctx->matchedState->altPriority) + shouldUpdate = true; /* Better lexical order wins */ + else if (state->altPriority == ctx->matchedState->altPriority && + matchEndRow > ctx->matchEndRow) + shouldUpdate = true; /* Same lexical order, longer wins */ + + if (shouldUpdate) + { + /* Free old matchedState if exists */ + if (ctx->matchedState != NULL) + nfa_state_free(winstate, ctx->matchedState); + + /* Take ownership of the new state */ + ctx->matchedState = state; + state->next = NULL; + ctx->matchEndRow = matchEndRow; + } + else + { + /* This state didn't win, free it */ + 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. + * varMatched[i] = true if variable i matched at current row. + */ +static bool +nfa_evaluate_row(WindowObject winobj, int64 pos, bool *varMatched) +{ + WindowAggState *winstate = winobj->winstate; + ExprContext *econtext = winstate->ss.ps.ps_ExprContext; + int numDefineVars = list_length(winstate->defineVariableList); + ListCell *lc; + int varIdx = 0; + TupleTableSlot *slot; + + /* + * 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); + + varMatched[varIdx] = (!isnull && DatumGetBool(result)); + + varIdx++; + if (varIdx >= numDefineVars) + break; + } + + 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 */ + ctx = MemoryContextAlloc(winstate->partcontext, sizeof(RPRNFAContext)); + } + + ctx->next = NULL; + ctx->prev = NULL; + ctx->states = NULL; + ctx->matchStartRow = -1; + ctx->matchEndRow = -1; + ctx->lastProcessedRow = -1; + ctx->matchedState = NULL; + + /* Update statistics */ + winstate->nfaContextsActive++; + winstate->nfaContextsTotalCreated++; + if (winstate->nfaContextsActive > winstate->nfaContextsMax) + winstate->nfaContextsMax = winstate->nfaContextsActive; + + 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 + * + * Unlink context from active list and return it to free list. + * Also frees any states in the context. + */ +static void +nfa_context_free(WindowAggState *winstate, RPRNFAContext *ctx) +{ + /* Unlink from active list first */ + nfa_unlink_context(winstate, ctx); + + /* Update statistics */ + winstate->nfaContextsActive--; + + if (ctx->states != NULL) + nfa_state_free_list(winstate, ctx->states); + if (ctx->matchedState != NULL) + nfa_state_free(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 and returns the new context. + */ +static RPRNFAContext * +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 tail of active context list (doubly-linked, oldest-first) */ + ctx->prev = winstate->nfaContextTail; + ctx->next = NULL; + if (winstate->nfaContextTail != NULL) + winstate->nfaContextTail->next = ctx; + else + winstate->nfaContext = ctx; /* first context becomes head */ + winstate->nfaContextTail = ctx; + + return 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; + + /* + * List is sorted by matchStartRow ascending (oldest/smallest at head). + * Stop early if we pass the target position. + */ + for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) + { + if (ctx->matchStartRow == pos) + return ctx; + if (ctx->matchStartRow > pos) + break; /* won't find it, list is sorted */ + } + return NULL; +} + +/* + * update_nfa_length_stats + * + * Helper function to update min/max/total length statistics. + * Called when tracking match/mismatch/absorbed/skipped lengths. + */ +static void +update_nfa_length_stats(int64 count, NFALengthStats *stats, int64 newLen) +{ + if (count == 1) + { + stats->min = newLen; + stats->max = newLen; + } + else + { + if (newLen < stats->min) + stats->min = newLen; + if (newLen > stats->max) + stats->max = newLen; + } + stats->total += newLen; +} + +/* + * record_nfa_context_failure + * + * Record a failed context in statistics. + * If failedLen == 1, count as pruned (failed on first row). + * If failedLen > 1, count as mismatched and update length stats. + */ +static void +record_nfa_context_failure(WindowAggState *winstate, int64 failedLen) +{ + if (failedLen == 1) + { + winstate->nfaContextsPruned++; + } + else + { + winstate->nfaMatchesFailed++; + update_nfa_length_stats(winstate->nfaMatchesFailed, + &winstate->nfaFailLen, + failedLen); + } +} + +/* + * nfa_remove_contexts_up_to + * + * Remove all contexts with matchStartRow <= endPos. + * Used by SKIP PAST LAST ROW to discard contexts within matched frame. + * + * excludeCtx: if not NULL, this context should not be counted in statistics + * (typically the matched context that triggered this removal). + */ +static void +nfa_remove_contexts_up_to(WindowAggState *winstate, int64 endPos, + RPRNFAContext *excludeCtx) +{ + RPRNFAContext *ctx; + RPRNFAContext *next; + + /* Contexts are sorted by matchStartRow ascending, so we can stop early */ + for (ctx = winstate->nfaContext; ctx != NULL; ctx = next) + { + next = ctx->next; + if (ctx->matchStartRow > endPos) + break; + + /* Track skipped context length statistics, excluding the matched context */ + if (ctx != excludeCtx && ctx->lastProcessedRow >= ctx->matchStartRow) + { + int64 skippedLen = ctx->lastProcessedRow - ctx->matchStartRow + 1; + + winstate->nfaContextsSkipped++; + update_nfa_length_stats(winstate->nfaContextsSkipped, + &winstate->nfaSkippedLen, + skippedLen); + } + + nfa_context_free(winstate, ctx); + } +} + +/* + * nfa_cleanup_dead_contexts + * + * Remove contexts that have failed (no active states and no match). + * These are contexts that failed during normal processing and should be + * counted as pruned (if length 1) or mismatched (if length > 1). + */ +static void +nfa_cleanup_dead_contexts(WindowAggState *winstate, RPRNFAContext *excludeCtx) +{ + RPRNFAContext *ctx; + RPRNFAContext *next; + + for (ctx = winstate->nfaContext; ctx != NULL; ctx = next) + { + next = ctx->next; + + /* Skip the target context and contexts still processing */ + if (ctx == excludeCtx || ctx->states != NULL) + continue; + + /* Skip successfully matched contexts (will be handled by SKIP logic) */ + if (ctx->matchEndRow >= ctx->matchStartRow) + continue; + + /* + * This is a failed context - count and remove it. + * Only count if it actually processed its start row. + * Contexts created for beyond-partition rows are silently removed. + */ + if (ctx->lastProcessedRow >= ctx->matchStartRow) + { + int64 failedLen = ctx->lastProcessedRow - ctx->matchStartRow + 1; + record_nfa_context_failure(winstate, failedLen); + } + /* else: context was never processed (beyond-partition), just remove */ + + nfa_context_free(winstate, ctx); + } +} + +/* + * 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 oldest-first (head=oldest, tail=newest), so we start from + * tail (newest) and check if older contexts (via prev) can absorb it. + */ +static void +nfa_absorb_contexts(WindowAggState *winstate, RPRNFAContext *excludeCtx, int64 currentPos) +{ + RPRPattern *pattern = winstate->rpPattern; + RPRNFAContext *ctx; + + /* 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; + + /* + * 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 oldest-first (head = lowest matchStartRow, tail = highest). + * We iterate from tail (newest) via prev and check if older contexts can absorb it. + */ + ctx = winstate->nfaContextTail; + + while (ctx != NULL) + { + RPRNFAContext *nextCtx = ctx->prev; /* traverse toward older (head) */ + 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->prev; older != NULL && !absorbed; older = older->prev) + { + /* 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) + { + /* Track absorbed context length statistics */ + int64 absorbedLen = ctx->matchEndRow - ctx->matchStartRow + 1; + + /* Absorb: remove ctx (newer) */ + nfa_context_free(winstate, ctx); + winstate->nfaContextsAbsorbed++; + update_nfa_length_stats(winstate->nfaContextsAbsorbed, + &winstate->nfaAbsorbedLen, + absorbedLen); + absorbed = true; + } + } + + ctx = nextCtx; + continue; + } + + /* + * SIMPLIFIED ABSORPTION: Compare counts at depth 0 only. + * + * For absorbable patterns (A+ or (A B)+), we use a simple rule: + * If older context has count[0] >= newer context's count[0] for + * ANY state, older can absorb newer. + * + * This works because: + * - A+: count[0] tracks how many A's matched + * - (A B)+: count[0] tracks how many groups matched + * + * If older is ahead or equal, newer's future matches are subsets. + */ + for (older = ctx->prev; older != NULL && !absorbed; older = older->prev) + { + int32 newerMaxCount = -1; + int32 olderMaxCount = -1; + RPRNFAState *s; + + /* 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 */ + if (older->states == NULL) + continue; + + /* Find maximum count[0] in newer context */ + for (s = ctx->states; s != NULL; s = s->next) + { + if (s->counts[0] > newerMaxCount) + newerMaxCount = s->counts[0]; + } + + /* Find maximum count[0] in older context */ + for (s = older->states; s != NULL; s = s->next) + { + if (s->counts[0] > olderMaxCount) + olderMaxCount = s->counts[0]; + } + + /* If older is ahead or equal, absorb newer */ + if (olderMaxCount >= 0 && newerMaxCount >= 0 && + olderMaxCount >= newerMaxCount) + { + int64 absorbedLen = ctx->lastProcessedRow - ctx->matchStartRow + 1; + + nfa_context_free(winstate, ctx); + winstate->nfaContextsAbsorbed++; + update_nfa_length_stats(winstate->nfaContextsAbsorbed, + &winstate->nfaAbsorbedLen, + absorbedLen); + absorbed = true; + } + } + + ctx = nextCtx; + } +} + +/* + * nfa_step + * + * Process all states in context through NFA for one row. + * Calls nfa_step_single for each state. + */ +static void +nfa_step(WindowAggState *winstate, RPRNFAContext *ctx, bool *varMatched, int64 pos) +{ + RPRNFAState *states = ctx->states; + RPRNFAState *state; + RPRNFAState *nextState; + + ctx->states = NULL; + + /* Track last processed row for failed match statistics */ + if (pos > ctx->lastProcessedRow) + ctx->lastProcessedRow = pos; + + for (state = states; state != NULL; state = nextState) + { + nextState = state->next; + state->next = NULL; + nfa_step_single(winstate, ctx, state, varMatched, pos); + } +} + +/* + * nfa_finalize_all_contexts + * + * Finalize all active contexts when partition ends. + * Calls nfa_step with NULL varMatched to complete without new row data. + */ +static void +nfa_finalize_all_contexts(WindowAggState *winstate, int64 lastPos) +{ + RPRNFAContext *ctx; + + for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) + { + if (ctx->states != NULL) + nfa_step(winstate, ctx, NULL, lastPos); + } +} + +/* + * nfa_process_context + * + * Process one context for the current row. + * Handles frame boundary check and NFA step. + */ +static void +nfa_process_context(WindowAggState *winstate, RPRNFAContext *ctx, + int64 currentPos, bool hasLimitedFrame, int64 frameOffset) +{ + /* Skip already-completed contexts */ + if (ctx->states == NULL) + return; + + /* Check frame boundary */ + if (hasLimitedFrame) + { + int64 ctxFrameEnd = ctx->matchStartRow + frameOffset + 1; + if (currentPos >= ctxFrameEnd) + { + nfa_step(winstate, ctx, NULL, ctxFrameEnd - 1); + return; + } + } + + /* Process states for this context */ + nfa_step(winstate, ctx, winstate->nfaVarMatched, currentPos); +} + +/* + * 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 */ + RPRNFAState *pending_tail = state; /* tail for FIFO append */ + + while (pending != NULL) + { + RPRPatternElement *elem; + bool matched; + int32 count; + int depth; + + /* Pop from head */ + state = pending; + pending = pending->next; + if (pending == NULL) + pending_tail = NULL; + 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. + * If varMatched is NULL (boundary), all variables are forced to false. + * Otherwise, varId < numDefines uses DEFINE expr, + * varId >= numDefines defaults to TRUE. + */ + if (varMatched == NULL) + matched = false; + else + { + 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) + { + /* + * Clamp count to prevent overflow with very large partitions. + * Once count reaches RPR_COUNT_MAX, we stop incrementing. + * This is safe because for unbounded quantifiers, only the + * comparison with min matters, and for bounded quantifiers, + * max is always < RPR_COUNT_MAX. + */ + if (count < RPR_COUNT_MAX) + 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); + 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); + } + 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 = NULL; + if (pending_tail) + pending_tail->next = state; + else + pending = state; + pending_tail = 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); + } + 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 = NULL; + if (pending_tail) + pending_tail->next = state; + else + pending = state; + pending_tail = state; + } + else + { + state->next = NULL; + if (pending_tail) + pending_tail->next = state; + else + pending = state; + pending_tail = 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); + } + 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. + * Append to pending_tail to maintain lexical order. + */ + while (altIdx >= 0 && altIdx < pattern->numElements) + { + RPRPatternElement *altElem = &elements[altIdx]; + RPRNFAState *newState; + + if (first) + { + state->elemIdx = altIdx; + state->altPriority = altIdx; + newState = state; + first = false; + } + else + { + newState = nfa_state_clone(winstate, altIdx, altIdx, + state->counts); + } + + /* Append to tail for lexical order */ + newState->next = NULL; + if (pending_tail) + pending_tail->next = newState; + else + pending = newState; + pending_tail = newState; + + 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 = NULL; + if (pending_tail) + pending_tail->next = state; + else + pending = state; + pending_tail = state; + } + else + { + /* + * Between min and max - can exit or continue. + * Exit state handling depends on what comes next: + * - If next is FIN: process on same row (match ends here) + * - If next is VAR/ALT: wait for next row (needs new input) + * Loop state always waits for next row. + */ + RPRPatternElement *nextElem = &elements[elem->next]; + RPRElemIdx exitAltPriority; + RPRNFAState *exitState; + + /* + * For greedy extension: if context already has a match recorded, + * preserve its altPriority. This ensures that iterations of a + * quantified group don't compete on lexical order - the first + * iteration's choice sets the preference, and subsequent + * iterations can extend it if they produce a longer match. + */ + exitAltPriority = state->altPriority; + if (ctx->matchedState != NULL) + exitAltPriority = ctx->matchedState->altPriority; + + exitState = nfa_state_clone(winstate, elem->next, + exitAltPriority, + state->counts); + exitState->counts[depth] = 0; + + if (RPRElemIsFin(nextElem)) + { + /* Match ends at current row - append to pending */ + exitState->next = NULL; + if (pending_tail) + pending_tail->next = exitState; + else + pending = exitState; + pending_tail = exitState; + } + else + { + /* Next element (VAR/ALT) needs new row input */ + nfa_add_state_unique(winstate, ctx, 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 = NULL; + if (pending_tail) + pending_tail->next = state; + else + pending = state; + pending_tail = state; + } + } +} + diff --git a/src/backend/utils/adt/windowfuncs.c b/src/backend/utils/adt/windowfuncs.c index 78b7f05aba2..efb60c99052 100644 --- a/src/backend/utils/adt/windowfuncs.c +++ b/src/backend/utils/adt/windowfuncs.c @@ -41,7 +41,6 @@ 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 +682,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 +723,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 894b6a1b6d6..7736a2c3360 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..a25d7642d3a 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,53 @@ 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) */ + int32 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) */ + int64 lastProcessedRow; /* last row processed (for fail depth) */ + RPRNFAState *matchedState; /* FIN state for greedy fallback (cloned) */ +} RPRNFAContext; + +/* + * NFALengthStats + * + * Statistics for length measurements (min/max/total) used for computing + * average lengths in EXPLAIN ANALYZE output. + */ +typedef struct NFALengthStats +{ + int64 min; /* minimum length */ + int64 max; /* maximum length */ + int64 total; /* total length (for computing average) */ +} NFALengthStats; + typedef struct WindowAggState { ScanState ss; /* its first field is NodeTag */ @@ -2571,6 +2619,39 @@ 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 */ + RPRNFAContext *nfaContext; /* active matching contexts (head) */ + RPRNFAContext *nfaContextTail; /* tail of active contexts (for reverse traversal) */ + 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) */ + + /* NFA statistics for EXPLAIN ANALYZE */ + int64 nfaStatesActive; /* current active states (internal) */ + int64 nfaStatesMax; /* peak active states */ + int64 nfaStatesTotalCreated; /* total states allocated */ + int64 nfaStatesMerged; /* states merged (deduplicated) */ + int64 nfaContextsActive; /* current active contexts (internal) */ + int64 nfaContextsMax; /* peak active contexts */ + int64 nfaContextsTotalCreated; /* total contexts allocated */ + int64 nfaContextsAbsorbed; /* contexts absorbed (optimization) */ + int64 nfaContextsSkipped; /* contexts skipped (SKIP PAST LAST ROW) */ + int64 nfaContextsPruned; /* contexts pruned on first row */ + int64 nfaMatchesSucceeded; /* successful pattern matches */ + int64 nfaMatchesFailed; /* failed pattern matches */ + NFALengthStats nfaMatchLen; /* successful match length stats */ + NFALengthStats nfaFailLen; /* mismatch length stats */ + NFALengthStats nfaAbsorbedLen; /* absorbed context length stats */ + NFALengthStats nfaSkippedLen; /* skipped context length stats */ + MemoryContext partcontext; /* context for partition-lifespan data */ MemoryContext aggcontext; /* shared context for aggregate working data */ MemoryContext curaggcontext; /* current aggregate's working data */ @@ -2598,6 +2679,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