From c3381f476a6987e36650df7799f3e1d8e53e30ad Mon Sep 17 00:00:00 2001 From: jian he Date: Wed, 19 Aug 2026 12:52:57 +0800 Subject: [PATCH v51 1/1] refactor nodeWindowAgg.c and execRPR.c 1. ExecRPRFreeContext, we should also reset other field value. 2. nfa_update_absorption_flags() now takes the WindowAggState and walks the context list itself. This is more intuitive, I think. I did the similar thing for nfa_absorb_contexts. 3. Remove some duplicated Asserts, other preceding caller have already did the equivalent Asserts. 4. Some local variable, we only use once, it would be better just remove these. 5. Add some elog(ERROR) to avoid circular winstate->nfaContext, RPRNFAContext->states. 6. Only ExecRPRProcessRow use variable frameOffset, refactor to make it as a local variable. 7. ExecRPRStartContext->nfa_context_make. nfa_context_make() used to return a bare struct, leaving the initial state, matchStartRow, and active-list linkage for the caller to fill in, that is not intuitive, it looks like a half-built context. Have nfa_context_make() return a fully-formed one: allocate the initial state, set the start row, and append it to the list tail. 8. ExecRPRGetHeadContext(pos) external function can be removed. 9. Now update_reduced_frame code flow is more intuitive: look up or create the context for pos, drive the NFA forward with advance_reduced_frame_nfa(), then record the match result. 10. Restructure get_reduced_frame_status as a single decision tree: first "no record" (start < 0), then "the record's own row" (pos == start), where length alone gives the verdict (-1 unmatched, 0 empty match, >= 1 frame head), then the range test for everything else. --- src/backend/executor/README.rpr | 2 +- src/backend/executor/execRPR.c | 298 +++++++++++++-------------- src/backend/executor/nodeWindowAgg.c | 213 +++++++++---------- src/backend/parser/parse_rpr.c | 2 + src/include/executor/execRPR.h | 5 +- 5 files changed, 248 insertions(+), 272 deletions(-) diff --git a/src/backend/executor/README.rpr b/src/backend/executor/README.rpr index b2118a3493..121c5aa923 100644 --- a/src/backend/executor/README.rpr +++ b/src/backend/executor/README.rpr @@ -674,7 +674,7 @@ Flow of update_reduced_frame(): Pseudocode of the row processing loop: - targetCtx = ExecRPRGetHeadContext(pos) + targetCtx = winstate->nfaContext if targetCtx == NULL: targetCtx = ExecRPRStartContext(pos) diff --git a/src/backend/executor/execRPR.c b/src/backend/executor/execRPR.c index aa034bcefe..75142b9f52 100644 --- a/src/backend/executor/execRPR.c +++ b/src/backend/executor/execRPR.c @@ -66,14 +66,14 @@ static void nfa_add_state_unique(WindowAggState *winstate, RPRNFAContext *ctx, static void nfa_add_matched_state(WindowAggState *winstate, RPRNFAContext *ctx, RPRNFAState *state, int64 matchEndRow); -static RPRNFAContext *nfa_context_make(WindowAggState *winstate); +static RPRNFAContext *nfa_context_make(WindowAggState *winstate, int64 startPos); static void nfa_unlink_context(WindowAggState *winstate, RPRNFAContext *ctx); static void nfa_update_length_stats(int64 count, NFALengthStats *stats, int64 newLen); static void nfa_record_context_skipped(WindowAggState *winstate, int64 skippedLen); static void nfa_record_context_absorbed(WindowAggState *winstate, int64 absorbedLen); -static void nfa_update_absorption_flags(RPRNFAContext *ctx); +static void nfa_update_absorption_flags(WindowAggState *winstate); static bool nfa_states_covered(RPRPattern *pattern, RPRNFAContext *older, RPRNFAContext *newer); static void nfa_try_absorb_context(WindowAggState *winstate, RPRNFAContext *ctx); @@ -141,6 +141,7 @@ nfa_state_make(WindowAggState *winstate) /* Initialize entire state to zero */ memset(state, 0, winstate->nfaStateSize); + state->isAbsorbable = winstate->rpPattern->isAbsorbable; /* Update statistics */ winstate->nfaStatesActive++; @@ -270,6 +271,9 @@ nfa_states_equal(WindowAggState *winstate, RPRNFAState *s1, RPRNFAState *s2) if (s1->elemIdx != s2->elemIdx) return false; + if (s1->isAbsorbable != s2->isAbsorbable) + return false; + /* * Compare counts up to current element's depth. Two states sharing * elemIdx are equivalent iff every enclosing-or-current depth count @@ -320,6 +324,9 @@ nfa_add_state_unique(WindowAggState *winstate, RPRNFAContext *ctx, RPRNFAState * { CHECK_FOR_INTERRUPTS(); + if (unlikely(s->next == ctx->states)) + elog(ERROR, "circular link in RPR NFA state list"); + if (nfa_states_equal(winstate, s, state)) { /* @@ -402,7 +409,7 @@ nfa_add_matched_state(WindowAggState *winstate, RPRNFAContext *ctx, * Allocate an NFA context, reusing from free list if available. */ static RPRNFAContext * -nfa_context_make(WindowAggState *winstate) +nfa_context_make(WindowAggState *winstate, int64 startPos) { RPRNFAContext *ctx; @@ -426,7 +433,6 @@ nfa_context_make(WindowAggState *winstate) ctx->matchedState = NULL; ctx->matchUpdated = false; - /* Initialize two-flag absorption design based on pattern */ ctx->hasAbsorbableState = winstate->rpPattern->isAbsorbable; ctx->allStatesAbsorbable = winstate->rpPattern->isAbsorbable; @@ -436,6 +442,32 @@ nfa_context_make(WindowAggState *winstate) winstate->nfaContextsMax = Max(winstate->nfaContextsMax, winstate->nfaContextsActive); + ctx->matchStartRow = startPos; + + /* initial state at elem 0 */ + ctx->states = nfa_state_make(winstate); + + Assert(RPRElemIsAbsorbableBranch(&winstate->rpPattern->elements[0]) == + winstate->rpPattern->isAbsorbable); + + /* + * Add to tail of active context list (doubly-linked, oldest-first). + * matchStartRow increases along the list, so the head holds the smallest + * -- an ordering other code relies on. At most one context starts at a + * row: the on-demand path in update_reduced_frame creates one only where + * none exists. + */ + Assert(winstate->nfaContextTail == NULL || + startPos > winstate->nfaContextTail->matchStartRow); + 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; } @@ -530,55 +562,64 @@ nfa_record_context_absorbed(WindowAggState *winstate, int64 absorbedLen) * permanently, so we skip recalculation. */ static void -nfa_update_absorption_flags(RPRNFAContext *ctx) +nfa_update_absorption_flags(WindowAggState *winstate) { - RPRNFAState *state; - bool hasAbsorbable = false; - bool allAbsorbable = true; - - /* - * Optimization: Once hasAbsorbableState becomes false, it stays false. No - * need to recalculate - both flags remain false permanently. - */ - if (!ctx->hasAbsorbableState) - { - ctx->allStatesAbsorbable = false; + if (!winstate->rpPattern->isAbsorbable) return; - } - /* No states means no absorbable states */ - if (ctx->states == NULL) + for (RPRNFAContext *ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) { - ctx->hasAbsorbableState = false; - ctx->allStatesAbsorbable = false; - return; - } + bool hasAbsorbable = false; + bool allAbsorbable = true; - /* - * Iterate through all states to check absorption status. Uses - * state->isAbsorbable which tracks if state is in absorbable region. This - * is different from RPRElemIsAbsorbable(elem) which checks comparison - * point. - */ - for (state = ctx->states; state != NULL; state = state->next) - { - CHECK_FOR_INTERRUPTS(); + /* + * Optimization: Once hasAbsorbableState becomes false, it stays + * false. No need to recalculate - both flags remain false + * permanently. + */ + if (!ctx->hasAbsorbableState) + { + ctx->allStatesAbsorbable = false; + continue; + } - if (state->isAbsorbable) - hasAbsorbable = true; - else - allAbsorbable = false; - } + /* No states means no absorbable states */ + if (ctx->states == NULL) + { + ctx->hasAbsorbableState = false; + ctx->allStatesAbsorbable = false; + continue;; + } - /* - * A recorded match makes this context non-absorbable: absorption would - * free the match, which no absorbing context can reproduce. - */ - if (ctx->matchedState != NULL) - allAbsorbable = false; + /* + * Iterate through all states to check absorption status. Uses + * state->isAbsorbable which tracks if state is in absorbable region. + * This is different from RPRElemIsAbsorbable(elem) which checks + * comparison point. + */ + for (RPRNFAState *state = ctx->states; state != NULL; state = state->next) + { + if (unlikely(state->next == ctx->states)) + elog(ERROR, "circular link in RPR NFA state list"); + + CHECK_FOR_INTERRUPTS(); - ctx->hasAbsorbableState = hasAbsorbable; - ctx->allStatesAbsorbable = allAbsorbable; + if (state->isAbsorbable) + hasAbsorbable = true; + else + allAbsorbable = false; + } + + /* + * A recorded match makes this context non-absorbable: absorption + * would free the match, which no absorbing context can reproduce. + */ + if (ctx->matchedState != NULL) + allAbsorbable = false; + + ctx->hasAbsorbableState = hasAbsorbable; + ctx->allStatesAbsorbable = allAbsorbable; + } } /* @@ -713,10 +754,12 @@ nfa_try_absorb_context(WindowAggState *winstate, RPRNFAContext *ctx) static void nfa_absorb_contexts(WindowAggState *winstate) { - RPRNFAContext *ctx; RPRNFAContext *nextCtx; - for (ctx = winstate->nfaContextTail; ctx != NULL; ctx = nextCtx) + if (!winstate->rpPattern->isAbsorbable) + return; + + for (RPRNFAContext *ctx = winstate->nfaContextTail; ctx != NULL; ctx = nextCtx) { nextCtx = ctx->prev; @@ -757,9 +800,6 @@ nfa_eval_var_match(WindowAggState *winstate, RPRPatternElement *elem, { int varId; - /* This function should only be called for VAR elements */ - Assert(RPRElemIsVar(elem)); - if (varMatched == NULL) return false; @@ -827,21 +867,29 @@ nfa_match(WindowAggState *winstate, RPRNFAContext *ctx, RPRVarMatch *varMatched, nextState = state->next; - if (RPRElemIsVar(elem)) + /* Non-VAR elements: keep as-is for advance phase */ + if (!RPRElemIsVar(elem)) + { + prevPtr = &state->next; + continue; + } + + if (!nfa_eval_var_match(winstate, elem, varMatched)) + { + /* + * Not matched - remove state. Exit alternatives were already + * created by advance phase when count >= min was satisfied. + */ + *prevPtr = nextState; + nfa_state_free(winstate, state); + continue; + } + else { int depth = elem->depth; int32 count = state->counts[depth]; - if (!nfa_eval_var_match(winstate, elem, varMatched)) - { - /* - * Not matched - remove state. Exit alternatives were already - * created by advance phase when count >= min was satisfied. - */ - *prevPtr = nextState; - nfa_state_free(winstate, state); - continue; - } + prevPtr = &state->next; /* * Increment count, saturating at RPR_COUNT_INF to avoid int32 @@ -939,11 +987,7 @@ nfa_match(WindowAggState *winstate, RPRNFAContext *ctx, RPRVarMatch *varMatched, endCount = outerCount; } } - /* else: stay at VAR for advance phase */ } - /* Non-VAR elements: keep as-is for advance phase */ - - prevPtr = &state->next; } } @@ -1077,7 +1121,6 @@ nfa_advance_alt(WindowAggState *winstate, RPRNFAContext *ctx, if (ctx->matchUpdated) break; - Assert(sepIdx >= 0 && sepIdx < pattern->numElements); sepElem = &elements[sepIdx]; Assert(RPRElemIsSep(sepElem)); @@ -1355,7 +1398,6 @@ nfa_advance_var(WindowAggState *winstate, RPRNFAContext *ctx, RPRNFAState *state, RPRPatternElement *elem, int64 currentPos) { - RPRPattern *pattern = winstate->rpPattern; int depth = elem->depth; int32 count = state->counts[depth]; bool canLoop = (elem->max == RPR_QUANTITY_INF || count < elem->max); @@ -1364,9 +1406,6 @@ nfa_advance_var(WindowAggState *winstate, RPRNFAContext *ctx, /* min <= max, so !canExit (count < min) implies canLoop (count < max) */ Assert(canLoop || canExit); - /* elem->next must be a valid index for any reachable VAR */ - Assert(elem->next >= 0 && elem->next < pattern->numElements); - if (canLoop && canExit) { /* @@ -1375,13 +1414,12 @@ nfa_advance_var(WindowAggState *winstate, RPRNFAContext *ctx, */ RPRNFAState *cloneState; RPRPatternElement *nextElem; - bool reluctant = RPRElemIsReluctant(elem); /* * Clone state for the first-priority path. For greedy, clone is the * loop state; for reluctant, clone is the exit state. */ - if (reluctant) + if (RPRElemIsReluctant(elem)) { /* Clone for exit, original stays for loop */ cloneState = nfa_state_clone(winstate, elem->next, @@ -1428,7 +1466,6 @@ nfa_advance_var(WindowAggState *winstate, RPRNFAContext *ctx, /* Exit only: advance to next element (canExit necessarily true) */ RPRPatternElement *nextElem; - Assert(canExit); nextElem = nfa_exit_to(winstate, state, depth, elem->next); nfa_route_to_elem(winstate, ctx, state, nextElem, currentPos); @@ -1448,8 +1485,6 @@ nfa_advance_state(WindowAggState *winstate, RPRNFAContext *ctx, RPRPattern *pattern = winstate->rpPattern; RPRPatternElement *elem; - Assert(state->elemIdx >= 0 && state->elemIdx < pattern->numElements); - /* Protect against stack overflow for deeply complex patterns */ check_stack_depth(); @@ -1511,7 +1546,6 @@ nfa_advance_state(WindowAggState *winstate, RPRNFAContext *ctx, switch (elem->varId) { case RPR_VARID_FIN: - /* FIN: record match */ nfa_add_matched_state(winstate, ctx, state, currentPos); break; @@ -1528,7 +1562,7 @@ nfa_advance_state(WindowAggState *winstate, RPRNFAContext *ctx, break; default: - /* VAR element; a SEP would land here, so see fillRPRPatternAlt */ + /* VAR element; a SEP should not land here */ Assert(!RPRElemIsSep(elem) && RPRElemIsVar(elem)); nfa_advance_var(winstate, ctx, state, elem, currentPos); break; @@ -1666,39 +1700,7 @@ nfa_reevaluate_dependent_vars(WindowAggState *winstate, RPRNFAContext *ctx, RPRNFAContext * ExecRPRStartContext(WindowAggState *winstate, int64 startPos) { - RPRNFAContext *ctx; - RPRPattern *pattern = winstate->rpPattern; - - ctx = nfa_context_make(winstate); - ctx->matchStartRow = startPos; - ctx->states = nfa_state_make(winstate); /* initial state at elem 0 */ - - /* - * The only state so far sits on element 0, and computeAbsorbability() - * marks that element ABSORBABLE_BRANCH exactly when it calls the pattern - * absorbable, so the pattern's flag answers for the state -- as it - * already did for the context flags nfa_context_make() set. - */ - Assert(RPRElemIsAbsorbableBranch(&pattern->elements[0]) == - pattern->isAbsorbable); - ctx->states->isAbsorbable = pattern->isAbsorbable; - - /* - * Add to tail of active context list (doubly-linked, oldest-first). - * matchStartRow increases along the list, so the head holds the smallest - * -- an ordering other code relies on. At most one context starts at a - * row: the on-demand path in update_reduced_frame creates one only where - * none exists. - */ - Assert(winstate->nfaContextTail == NULL || - startPos > winstate->nfaContextTail->matchStartRow); - 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; + RPRNFAContext *ctx = nfa_context_make(winstate, startPos); /* * Initial advance (divergence): expand ALT branches and create exit @@ -1714,27 +1716,6 @@ ExecRPRStartContext(WindowAggState *winstate, int64 startPos) return ctx; } -/* - * ExecRPRGetHeadContext - * - * Return the head context if its start position matches pos. - * Returns NULL if no context exists or head doesn't match pos. - */ -RPRNFAContext * -ExecRPRGetHeadContext(WindowAggState *winstate, int64 pos) -{ - RPRNFAContext *ctx = winstate->nfaContext; - - /* - * Contexts are sorted by matchStartRow ascending. If the head context - * doesn't match pos, no context exists for this position. - */ - if (ctx == NULL || ctx->matchStartRow != pos) - return NULL; - - return ctx; -} - /* * ExecRPRFreeContext * @@ -1755,9 +1736,15 @@ ExecRPRFreeContext(WindowAggState *winstate, RPRNFAContext *ctx) if (ctx->matchedState != NULL) nfa_state_free(winstate, ctx->matchedState); + ctx->next = winstate->nfaContextFree; ctx->states = NULL; + ctx->matchStartRow = -1; + ctx->matchEndRow = -1; + ctx->lastProcessedRow = -1; ctx->matchedState = NULL; - ctx->next = winstate->nfaContextFree; + ctx->matchUpdated = false; + ctx->hasAbsorbableState = false; + ctx->allStatesAbsorbable = false; winstate->nfaContextFree = ctx; } @@ -1807,12 +1794,18 @@ ExecRPRRecordContextFailure(WindowAggState *winstate, int64 failedLen) * 3. Advance all contexts (divergence) - create new states for next row */ void -ExecRPRProcessRow(WindowAggState *winstate, int64 currentPos, - bool hasLimitedFrame, int64 frameOffset) +ExecRPRProcessRow(WindowAggState *winstate, int64 currentPos) { - RPRNFAContext *ctx; RPRVarMatch *varMatched = winstate->nfaVarMatched; bool hasDependent = !bms_is_empty(winstate->defineMatchStartDependent); + int64 frameOffset = 0; + + /* + * Check if we have a limited frame (ROWS ... N FOLLOWING). Each context + * needs its own frame end based on matchStartRow + offset. + */ + if (!(winstate->frameOptions & FRAMEOPTION_END_UNBOUNDED_FOLLOWING)) + frameOffset = DatumGetInt64(winstate->endOffsetValue); /* Allow query cancellation once per row for simple/low-state patterns */ CHECK_FOR_INTERRUPTS(); @@ -1821,13 +1814,16 @@ ExecRPRProcessRow(WindowAggState *winstate, int64 currentPos, * Phase 1: Match all contexts (convergence). Evaluate VAR elements, * update counts, remove dead states. */ - for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) + for (RPRNFAContext *ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) { + if (unlikely(ctx->next == winstate->nfaContext)) + elog(ERROR, "circular link in RPR NFA context list"); + if (ctx->states == NULL) continue; /* Check frame boundary - finalize the context when it is reached */ - if (hasLimitedFrame) + if (frameOffset > 0) { int64 ctxFrameEnd; @@ -1885,19 +1881,14 @@ ExecRPRProcessRow(WindowAggState *winstate, int64 currentPos, * converged - ideal for absorption. First update absorption flags that * may have changed due to state removal. */ - if (winstate->rpPattern->isAbsorbable) - { - for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) - nfa_update_absorption_flags(ctx); - - nfa_absorb_contexts(winstate); - } + nfa_update_absorption_flags(winstate); + nfa_absorb_contexts(winstate); /* * Phase 3: Advance all contexts (divergence). Create new states * (loop/exit) from surviving matched states. */ - for (ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) + for (RPRNFAContext *ctx = winstate->nfaContext; ctx != NULL; ctx = ctx->next) { if (ctx->states == NULL) continue; @@ -1913,7 +1904,7 @@ ExecRPRProcessRow(WindowAggState *winstate, int64 currentPos, * when frameOffset is near PG_INT64_MAX. */ #ifdef USE_ASSERT_CHECKING - if (hasLimitedFrame) + if (frameOffset > 0) { int64 ctxFrameEnd; @@ -1948,10 +1939,17 @@ ExecRPRCleanupDeadContexts(WindowAggState *winstate, RPRNFAContext *excludeCtx) next = ctx->next; + if (unlikely(next == winstate->nfaContext)) + elog(ERROR, "circular link in RPR NFA context list"); + /* Skip the target context and contexts still processing */ if (ctx == excludeCtx || ctx->states != NULL) continue; + /* Future context should be skipped */ + if (excludeCtx->matchStartRow > winstate->currentpos) + continue; + /* * Skip contexts that recorded a match (handled by SKIP logic). Test * matchedState, not matchEndRow: an empty match ends at matchStartRow @@ -1968,9 +1966,8 @@ ExecRPRCleanupDeadContexts(WindowAggState *winstate, RPRNFAContext *excludeCtx) */ if (ctx->lastProcessedRow >= ctx->matchStartRow) { - int64 failedLen = ctx->lastProcessedRow - ctx->matchStartRow + 1; - - ExecRPRRecordContextFailure(winstate, failedLen); + ExecRPRRecordContextFailure(winstate, + ctx->lastProcessedRow - ctx->matchStartRow + 1); } ExecRPRFreeContext(winstate, ctx); @@ -2014,6 +2011,9 @@ ExecRPRFinalizeAllContexts(WindowAggState *winstate, int64 lastPos) { CHECK_FOR_INTERRUPTS(); + if (unlikely(ctx->next == winstate->nfaContext)) + elog(ERROR, "circular link in RPR NFA context list"); + if (ctx->states != NULL) { nfa_match(winstate, ctx, NULL, lastPos); diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c index 25e67cd8ac..40092fe5de 100644 --- a/src/backend/executor/nodeWindowAgg.c +++ b/src/backend/executor/nodeWindowAgg.c @@ -255,8 +255,7 @@ static void clear_reduced_frame(WindowAggState *winstate); static int get_reduced_frame_status(WindowAggState *winstate, int64 pos); static void advance_nav_mark(WindowAggState *winstate, int64 currentPos); static void advance_reduced_frame_nfa(WindowObject winobj, - RPRNFAContext *targetCtx, int64 pos, - bool hasLimitedFrame, int64 frameOffset); + RPRNFAContext *targetCtx); static void update_reduced_frame(WindowObject winobj, int64 pos); /* Forward declarations - DEFINE row evaluation */ @@ -2541,12 +2540,12 @@ 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, clear the current - * match so the next row triggers re-evaluation. - */ if (rpr_is_defined(winstate)) { + /* + * If RPR is defined and skip mode is next row, clear the + * current match so the next row triggers re-evaluation. + */ if (winstate->rpSkipTo == ST_NEXT_ROW) clear_reduced_frame(winstate); @@ -4512,9 +4511,6 @@ clear_reduced_frame(WindowAggState *winstate) * start >= 0, length == -1 unmatched (covers only the start row) * start >= 0, length == 0 empty match (zero-length match at start) * start >= 0, length >= 1 real match spanning [start, start + length) - * - * The tests below form a cascade with early returns, so their order is - * significant. */ static int get_reduced_frame_status(WindowAggState *winstate, int64 pos) @@ -4522,31 +4518,38 @@ get_reduced_frame_status(WindowAggState *winstate, int64 pos) int64 start = winstate->rpr_match_start; int64 length = winstate->rpr_match_length; - if (start < 0) - return RF_NOT_DETERMINED; /* cleared slot: no result recorded yet */ + Assert(pos >= 0); - /* - * Unmatched (length -1) and empty match (length 0) do not describe a - * positive-length range, so they are classified before the range test. - * Each covers only its own start row; any other position is not part of - * this record and is still undetermined. - */ - if (length == -1) - return (pos == start) ? RF_UNMATCHED : RF_NOT_DETERMINED; - if (length == 0) - return (pos == start) ? RF_EMPTY_MATCH : RF_NOT_DETERMINED; - - /* - * By here length >= 1, so [start, start + length) is a well-formed range. - */ - if (pos < start || pos >= start + length) + if (start < 0) + /* cleared slot: no result recorded yet */ return RF_NOT_DETERMINED; - - /* pos lies within a real match. */ - if (pos == start) - return RF_FRAME_HEAD; - - return RF_SKIPPED; + else if (pos == start) + { + /* + * The record's own row: the length gives the verdict directly (-1 + * unmatched, 0 empty match, >= 1 head of a real match). + */ + if (length < 0) + return RF_UNMATCHED; + else if (length == 0) + return RF_EMPTY_MATCH; + else + return RF_FRAME_HEAD; + } + else if (pos < start || pos >= start + length) + { + /* + * Any other row is either inside the recorded match's range, [start, + * start + length), or not covered by this record at all. The + * sentinel lengths need no special casing: for -1 and 0 the range is + * empty (or ends before start), so every pos != start correctly falls + * out as not determined. + */ + return RF_NOT_DETERMINED; + } + else + /* inside a real match, after its head */ + return RF_SKIPPED; } /* @@ -4607,21 +4610,30 @@ advance_nav_mark(WindowAggState *winstate, int64 currentPos) * evaluations are shared across all active contexts. */ static void -advance_reduced_frame_nfa(WindowObject winobj, RPRNFAContext *targetCtx, - int64 pos, bool hasLimitedFrame, int64 frameOffset) +advance_reduced_frame_nfa(WindowObject winobj, RPRNFAContext *targetCtx) { WindowAggState *winstate = winobj->winstate; int64 currentPos; int64 startPos; int64 saved_currentpos = winstate->currentpos; + /* + * An empty state list means the context already completed in an earlier + * call, its result recorded in matchedState, nothing left to drive. It + * can be reachable under SKIP TO NEXT ROW, where an overlapping context + * can finish before the call for its own start row arrives. + */ + if (targetCtx->states == NULL) + return; + /* * 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); + startPos = Max(targetCtx->matchStartRow, + winstate->nfaLastProcessedRow + 1); /* * Process rows until target context completes or we hit boundaries. Each @@ -4656,16 +4668,13 @@ advance_reduced_frame_nfa(WindowObject winobj, RPRNFAContext *targetCtx, break; } - /* Update last processed row */ - winstate->nfaLastProcessedRow = currentPos; - /*-------------------------- * Process all contexts for this row: * 1. Match all (convergence) * 2. Absorb redundant * 3. Advance all (divergence) */ - ExecRPRProcessRow(winstate, currentPos, hasLimitedFrame, frameOffset); + ExecRPRProcessRow(winstate, currentPos); /* * Create a new context for the next potential start position. This @@ -4682,6 +4691,9 @@ advance_reduced_frame_nfa(WindowObject winobj, RPRNFAContext *targetCtx, /* Advance the nav mark to the frontier so trim can free old rows. */ advance_nav_mark(winstate, currentPos); + + /* Update last processed row */ + winstate->nfaLastProcessedRow = currentPos; } /* Restore the output row position borrowed for the NFA scan. */ @@ -4706,40 +4718,33 @@ static void update_reduced_frame(WindowObject winobj, int64 pos) { WindowAggState *winstate = winobj->winstate; - RPRNFAContext *targetCtx; - int frameOptions = winstate->frameOptions; - bool hasLimitedFrame; - int64 frameOffset = 0; - int64 matchLen; + RPRNFAContext *targetCtx = NULL; - /* - * 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) - frameOffset = DatumGetInt64(winstate->endOffsetValue); + winstate->rpr_match_start = pos; + winstate->rpr_match_length = -1; - /* - * 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) + if (winstate->nfaContext != NULL) { - /* already processed, unmatched */ - winstate->rpr_match_start = pos; - winstate->rpr_match_length = -1; - return; + /* + * 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->matchStartRow > pos) + return; + else if (winstate->nfaContext->matchStartRow == pos) + { + /* + * Case 2: the head context starts exactly at pos, it holds this + * row's pending result: either still in flight, or already + * completed by an earlier call's driver loop (SKIP TO NEXT ROW). + * Later contexts can't apply: the list ascends by matchStartRow. + */ + targetCtx = winstate->nfaContext; + } } - /* - * Case 2: Find existing context for this pos, or create new one. - */ - targetCtx = ExecRPRGetHeadContext(winstate, pos); if (targetCtx == NULL) { /* @@ -4748,67 +4753,39 @@ update_reduced_frame(WindowObject winobj, int64 pos) * reprocess. */ if (pos <= winstate->nfaLastProcessedRow) - { - /* already processed, unmatched */ - winstate->rpr_match_start = pos; - winstate->rpr_match_length = -1; return; - } + /* Not yet processed - create new context and start fresh */ targetCtx = ExecRPRStartContext(winstate, pos); } - else if (targetCtx->states == NULL) - { - /* - * The head context already completed in an earlier call. Reachable - * under SKIP TO NEXT ROW, where overlapping contexts let one reach - * FIN -- recording its result -- before the call for its own start - * row arrives. Register that result. - */ - goto register_result; - } - /* Drive the NFA forward until pos's match is resolved. */ - advance_reduced_frame_nfa(winobj, targetCtx, pos, hasLimitedFrame, - frameOffset); + /* Drive the NFA forward */ + advance_reduced_frame_nfa(winobj, targetCtx); -register_result: - Assert(pos == targetCtx->matchStartRow); - - /* - * Record match result. A determined slot has rpr_match_start >= 0; the - * length then gives the kind: -1 unmatched, 0 empty match, >= 1 real - * match. A cleared slot keeps rpr_match_start < 0. - */ - winstate->rpr_match_start = targetCtx->matchStartRow; - - if (targetCtx->matchEndRow < targetCtx->matchStartRow) + if (targetCtx->matchedState == NULL) { - matchLen = targetCtx->lastProcessedRow - targetCtx->matchStartRow + 1; - - if (targetCtx->matchedState != NULL) - { - /* Empty match: FIN reached but 0 rows consumed */ + /* No match */ + winstate->rpr_match_length = -1; + ExecRPRRecordContextFailure(winstate, + targetCtx->lastProcessedRow - targetCtx->matchStartRow + 1); + } + else + { + /* Empty match: FIN reached but 0 rows consumed */ + if (targetCtx->matchEndRow < targetCtx->matchStartRow) winstate->rpr_match_length = 0; - ExecRPRRecordContextSuccess(winstate, 0); - } else - { - /* No match */ - winstate->rpr_match_length = -1; - ExecRPRRecordContextFailure(winstate, matchLen); - } - ExecRPRFreeContext(winstate, targetCtx); - return; - } + /* Match succeeded */ + winstate->rpr_match_length = + targetCtx->matchEndRow - targetCtx->matchStartRow + 1; - /* Match succeeded */ - matchLen = targetCtx->matchEndRow - targetCtx->matchStartRow + 1; - - winstate->rpr_match_length = matchLen; - ExecRPRRecordContextSuccess(winstate, matchLen); + ExecRPRRecordContextSuccess(winstate, winstate->rpr_match_length); + } - /* Remove the matched context */ + /* + * The result for pos is recorded; matched or not, this context is + * consumed, so release it. + */ ExecRPRFreeContext(winstate, targetCtx); } diff --git a/src/backend/parser/parse_rpr.c b/src/backend/parser/parse_rpr.c index 3411369922..d1cc36ee0a 100644 --- a/src/backend/parser/parse_rpr.c +++ b/src/backend/parser/parse_rpr.c @@ -172,6 +172,8 @@ transformRPR(ParseState *pstate, WindowClause *wc, WindowDef *windef, windef->frameLocation >= 0 ? windef->frameLocation : windef->location)); + Assert(wc->frameOptions & FRAMEOPTION_ROWS); + /* Assign AFTER MATCH SKIP TO flag */ wc->rpSkipTo = windef->rpCommonSyntax->rpSkipTo; diff --git a/src/include/executor/execRPR.h b/src/include/executor/execRPR.h index fb7dc63a4c..265a53efc5 100644 --- a/src/include/executor/execRPR.h +++ b/src/include/executor/execRPR.h @@ -19,13 +19,10 @@ /* NFA context management */ extern RPRNFAContext *ExecRPRStartContext(WindowAggState *winstate, int64 startPos); -extern RPRNFAContext *ExecRPRGetHeadContext(WindowAggState *winstate, - int64 pos); extern void ExecRPRFreeContext(WindowAggState *winstate, RPRNFAContext *ctx); /* NFA processing */ -extern void ExecRPRProcessRow(WindowAggState *winstate, int64 currentPos, - bool hasLimitedFrame, int64 frameOffset); +extern void ExecRPRProcessRow(WindowAggState *winstate, int64 currentPos); extern void ExecRPRCleanupDeadContexts(WindowAggState *winstate, RPRNFAContext *excludeCtx); extern void ExecRPRFinalizeAllContexts(WindowAggState *winstate, int64 lastPos); -- 2.34.1