From 9544c24726aa4ac3b0ff2b91f47a6d64aa4ac925 Mon Sep 17 00:00:00 2001 From: jian he Date: Sun, 16 Aug 2026 14:29:10 +0900 Subject: [PATCH] Have the RPR executor read its own state instead of being handed it update_reduced_frame() computed whether the frame was limited and what its offset was, then threaded both through advance_reduced_frame_nfa() into ExecRPRProcessRow(). Neither value is the caller's to know: frameOptions and endOffsetValue sit in WindowAggState, which ExecRPRProcessRow() already has, and one frameOffset of -1 says "runs to the partition end" where a separate flag was needed before. The start position went the same way, being the target context's matchStartRow. nfa_update_absorption_flags() took one context and left the caller to test pattern->isAbsorbable and walk the list. Give it winstate and let it do both, as nfa_absorb_contexts() beside it now does, so the guard lives once at the top of each rather than at every call. ExecRPRGetHeadContext() had no caller left and is gone from the header. ExecRPRFreeContext() now clears every field of a context it returns to the free list, not the three that happened to matter. get_reduced_frame_status() classified a position through a cascade whose order carried the meaning: unmatched and empty match had to be tested before the range test because their lengths describe no range. Test the record's own row first instead. The three verdicts only that row can produce are then decided outright, and the range test needs no special casing -- lengths of -1 and 0 leave the range empty. nfa_match() kept the whole VAR case inside a conditional. Take the non-VAR case as an early exit and the VAR case unindents. ExecEvalRPRNavSet() asserted three times that a subtraction could not underflow, re-deriving the same fact each time. Assert the one precondition it rests on, that currentpos is non-negative. Phase 3 of ExecRPRProcessRow() recomputed the frame end under USE_ASSERT_CHECKING to assert what Phase 1 had just enforced; that block is gone. No behavior changes. --- src/backend/executor/README.rpr | 23 +- src/backend/executor/execExprInterp.c | 9 +- src/backend/executor/execRPR.c | 414 ++++++++++++-------------- src/backend/executor/nodeWindowAgg.c | 190 ++++++------ src/backend/parser/parse_rpr.c | 2 + src/include/executor/execRPR.h | 5 +- 6 files changed, 296 insertions(+), 347 deletions(-) diff --git a/src/backend/executor/README.rpr b/src/backend/executor/README.rpr index 05bbcd76240..eab06f775d2 100644 --- a/src/backend/executor/README.rpr +++ b/src/backend/executor/README.rpr @@ -678,8 +678,15 @@ Flow of update_reduced_frame(): Pseudocode of the row processing loop: - targetCtx = ExecRPRGetHeadContext(pos) + targetCtx = NULL + if nfaContext != NULL: -- head is the oldest context + if nfaContext->matchStartRow > pos: -- pos already skipped past + return + if nfaContext->matchStartRow == pos: + targetCtx = nfaContext if targetCtx == NULL: + if pos <= nfaLastProcessedRow: -- already unmatched or skipped + return targetCtx = ExecRPRStartContext(pos) for currentPos = startPos; targetCtx->states != NULL; currentPos++: @@ -1361,12 +1368,14 @@ X-4. Bounded Frame Handling rejected, since it would reduce the frame to the single current row. When the frame is bounded (e.g., ROWS BETWEEN CURRENT ROW AND 5 - FOLLOWING), ExecRPRProcessRow receives hasLimitedFrame=true and - frameOffset indicating the upper bound. Before the match phase, - any context whose match has exceeded the frame boundary - (currentPos >= matchStartRow + frameOffset + 1) is finalized early - by forcing a mismatch. This prevents matches from extending beyond - the window frame. The sum is clamped to PG_INT64_MAX on overflow. + FOLLOWING), ExecRPRProcessRow derives the upper bound itself from + winstate->frameOptions and winstate->endOffsetValue, using a + frameOffset of -1 to mean the frame runs to the partition end. + Before the match phase, any context whose match has exceeded the + frame boundary (currentPos >= matchStartRow + frameOffset + 1) is + finalized early by forcing a mismatch. This prevents matches from + extending beyond the window frame. The sum is clamped to + PG_INT64_MAX on overflow. Note that bounded frames also disable context absorption at the planner level (see VIII-3(b)), since the frame boundary breaks the diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c index d19e20a2eaa..f88547c7d94 100644 --- a/src/backend/executor/execExprInterp.c +++ b/src/backend/executor/execExprInterp.c @@ -6047,6 +6047,7 @@ ExecEvalRPRNavSet(ExprState *state, ExprEvalStep *op, ExprContext *econtext) offset = DatumGetInt64(rprnavstate->offset.value); compound_offset = DatumGetInt64(rprnavstate->compound_offset.value); + Assert(winstate->currentpos >= 0); Assert(offset >= 0 && compound_offset >= 0); /* @@ -6058,11 +6059,9 @@ ExecEvalRPRNavSet(ExprState *state, ExprEvalStep *op, ExprContext *econtext) case RPR_NAV_PREV: /* - * currentpos and offset are both non-negative, so the subtraction - * cannot underflow; assert the invariant rather than guarding an - * unreachable overflow. + * currentpos and offset are both non-negative, asserted above, so + * the subtraction cannot underflow. */ - Assert(!pg_sub_s64_overflow(winstate->currentpos, offset, &target_pos)); target_pos = winstate->currentpos - offset; break; case RPR_NAV_NEXT: @@ -6108,7 +6107,6 @@ ExecEvalRPRNavSet(ExprState *state, ExprEvalStep *op, ExprContext *econtext) * inner_pos is in [0, currentpos] and compound_offset is * non-negative, so this cannot underflow. */ - Assert(!pg_sub_s64_overflow(inner_pos, compound_offset, &target_pos)); target_pos = inner_pos - compound_offset; } else @@ -6144,7 +6142,6 @@ ExecEvalRPRNavSet(ExprState *state, ExprEvalStep *op, ExprContext *econtext) * and compound_offset is non-negative, so this cannot * underflow. */ - Assert(!pg_sub_s64_overflow(inner_pos, compound_offset, &target_pos)); target_pos = inner_pos - compound_offset; } else diff --git a/src/backend/executor/execRPR.c b/src/backend/executor/execRPR.c index b10aa7df60f..105abeee9d9 100644 --- a/src/backend/executor/execRPR.c +++ b/src/backend/executor/execRPR.c @@ -73,7 +73,7 @@ static void nfa_update_length_stats(int64 count, NFALengthStats *stats, int64 ne 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); @@ -512,7 +512,7 @@ nfa_record_context_absorbed(WindowAggState *winstate, int64 absorbedLen) /* * nfa_update_absorption_flags * - * Update context's absorption flags after state changes. + * Update every live context's absorption flags after state changes. * * Two flags control absorption behavior: * hasAbsorbableState: true if context has at least one absorbable state. @@ -527,55 +527,61 @@ 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) + { + 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; + } } /* @@ -710,10 +716,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; @@ -826,134 +834,131 @@ nfa_match(WindowAggState *winstate, RPRNFAContext *ctx, RPRVarMatch *varMatched, for (state = ctx->states; state != NULL; state = nextState) { RPRPatternElement *elem = &elements[state->elemIdx]; + int depth; + int32 count; CHECK_FOR_INTERRUPTS(); nextState = state->next; - if (RPRElemIsVar(elem)) + /* Non-VAR elements: keep as-is for advance phase */ + if (!RPRElemIsVar(elem)) { - bool matched; - int depth = elem->depth; - int32 count = state->counts[depth]; + prevPtr = &state->next; + continue; + } - matched = nfa_eval_var_match(winstate, elem, varMatched); + 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; + } - if (matched) - { - /* - * Increment count, saturating at RPR_COUNT_INF to avoid int32 - * overflow; a saturated count then compares as "unbounded". - */ - if (count < RPR_COUNT_INF) - count++; + prevPtr = &state->next; - /* Max constraint should not be exceeded */ - Assert(elem->max == RPR_QUANTITY_INF || count <= elem->max); + depth = elem->depth; + count = state->counts[depth]; - state->counts[depth] = count; + /* + * Increment count, saturating at RPR_COUNT_INF to avoid int32 + * overflow; a saturated count then compares as "unbounded". + */ + if (count < RPR_COUNT_INF) + count++; - /* - * For VAR at max count with END next, advance through END - * chain to reach the absorption comparison point. Only - * deterministic exits (count >= max, max finite) are handled; - * unbounded VARs stay for advance phase. - * - * In nested patterns like ((A (B C){2}){2})+, a VAR reaching - * its max triggers an exit cascade: inner END increments - * inner group count, which may itself reach max, requiring an - * exit to the next outer END. The loop below walks this - * chain. - * - * ABSORBABLE_BRANCH marks elements inside the absorbable - * region; ABSORBABLE marks the outermost comparison point - * where count-dominance is evaluated. We chain through - * BRANCH elements until reaching the ABSORBABLE point or an - * element that can still loop (count < max). - */ - if (RPRElemIsAbsorbableBranch(elem) && - !RPRElemIsAbsorbable(elem) && - count >= elem->max && - RPRElemIsEnd(&elements[elem->next])) - { - RPRPatternElement *endElem = &elements[elem->next]; - int endDepth = endElem->depth; - int32 endCount = state->counts[endDepth]; - - /* Increment group count */ - if (endCount < RPR_COUNT_INF) - endCount++; - Assert(endElem->max == RPR_QUANTITY_INF || - endCount <= endElem->max); - - state->elemIdx = elem->next; - state->counts[endDepth] = endCount; - - /* - * Leaf VAR exited (reached max): clear its own count so - * the next occupant enters with zero, as nfa_advance_var - * does on exit (this inline path replaces that exit). - * depth > endDepth, so this leaves the group count just - * written intact. - */ - Assert(endDepth < depth); - state->counts[depth] = 0; - - /* - * Chain through END elements within the absorbable region - * (ABSORBABLE_BRANCH) until reaching the comparison point - * (ABSORBABLE). Continue only on must-exit path (count - * >= max) with END next. - */ - while (RPRElemIsAbsorbableBranch(endElem) && - !RPRElemIsAbsorbable(endElem) && - endCount >= endElem->max && - RPRElemIsEnd(&elements[endElem->next])) - { - RPRPatternElement *outerEnd = &elements[endElem->next]; - int outerDepth = outerEnd->depth; - int32 outerCount = state->counts[outerDepth]; - - /* - * Exit this intermediate group: clear its own count - * (count-clear policy). It sits below the absorbable - * comparison point, so it is excluded from the - * dominance comparison; the comparison point where - * the chain stops keeps its count. - */ - state->counts[endDepth] = 0; - - /* Increment outer group count */ - if (outerCount < RPR_COUNT_INF) - outerCount++; - Assert(outerEnd->max == RPR_QUANTITY_INF || - outerCount <= outerEnd->max); - - state->elemIdx = endElem->next; - state->counts[outerDepth] = outerCount; - - /* Advance to next END in chain */ - endElem = outerEnd; - endDepth = outerDepth; - endCount = outerCount; - } - } - /* else: stay at VAR for advance phase */ - } - else + /* Max constraint should not be exceeded */ + Assert(elem->max == RPR_QUANTITY_INF || count <= elem->max); + + state->counts[depth] = count; + + /* + * For VAR at max count with END next, advance through END chain to + * reach the absorption comparison point. Only deterministic exits + * (count >= max, max finite) are handled; unbounded VARs stay for + * advance phase. + * + * In nested patterns like ((A (B C){2}){2})+, a VAR reaching its max + * triggers an exit cascade: inner END increments inner group count, + * which may itself reach max, requiring an exit to the next outer + * END. The loop below walks this chain. + * + * ABSORBABLE_BRANCH marks elements inside the absorbable region; + * ABSORBABLE marks the outermost comparison point where + * count-dominance is evaluated. We chain through BRANCH elements + * until reaching the ABSORBABLE point or an element that can still + * loop (count < max). + */ + if (RPRElemIsAbsorbableBranch(elem) && + !RPRElemIsAbsorbable(elem) && + count >= elem->max && + RPRElemIsEnd(&elements[elem->next])) + { + RPRPatternElement *endElem = &elements[elem->next]; + int endDepth = endElem->depth; + int32 endCount = state->counts[endDepth]; + + /* Increment group count */ + if (endCount < RPR_COUNT_INF) + endCount++; + Assert(endElem->max == RPR_QUANTITY_INF || + endCount <= endElem->max); + + state->elemIdx = elem->next; + state->counts[endDepth] = endCount; + + /* + * Leaf VAR exited (reached max): clear its own count so the next + * occupant enters with zero, as nfa_advance_var does on exit + * (this inline path replaces that exit). depth > endDepth, so + * this leaves the group count just written intact. + */ + Assert(endDepth < depth); + state->counts[depth] = 0; + + /* + * Chain through END elements within the absorbable region + * (ABSORBABLE_BRANCH) until reaching the comparison point + * (ABSORBABLE). Continue only on must-exit path (count >= max) + * with END next. + */ + while (RPRElemIsAbsorbableBranch(endElem) && + !RPRElemIsAbsorbable(endElem) && + endCount >= endElem->max && + RPRElemIsEnd(&elements[endElem->next])) { + RPRPatternElement *outerEnd = &elements[endElem->next]; + int outerDepth = outerEnd->depth; + int32 outerCount = state->counts[outerDepth]; + /* - * Not matched - remove state. Exit alternatives were already - * created by advance phase when count >= min was satisfied. + * Exit this intermediate group: clear its own count + * (count-clear policy). It sits below the absorbable + * comparison point, so it is excluded from the dominance + * comparison; the comparison point where the chain stops + * keeps its count. */ - *prevPtr = nextState; - nfa_state_free(winstate, state); - continue; + state->counts[endDepth] = 0; + + /* Increment outer group count */ + if (outerCount < RPR_COUNT_INF) + outerCount++; + Assert(outerEnd->max == RPR_QUANTITY_INF || + outerCount <= outerEnd->max); + + state->elemIdx = endElem->next; + state->counts[outerDepth] = outerCount; + + /* Advance to next END in chain */ + endElem = outerEnd; + endDepth = outerDepth; + endCount = outerCount; } } - /* Non-VAR elements: keep as-is for advance phase */ - - prevPtr = &state->next; } } @@ -1365,7 +1370,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); @@ -1375,7 +1379,8 @@ nfa_advance_var(WindowAggState *winstate, RPRNFAContext *ctx, Assert(canLoop || canExit); /* elem->next must be a valid index for any reachable VAR */ - Assert(elem->next >= 0 && elem->next < pattern->numElements); + Assert(elem->next >= 0 && + elem->next < winstate->rpPattern->numElements); if (canLoop && canExit) { @@ -1385,13 +1390,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, @@ -1438,7 +1442,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); @@ -1533,7 +1536,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; @@ -1550,7 +1552,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; @@ -1740,27 +1742,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 * @@ -1781,9 +1762,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; } @@ -1833,12 +1820,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 = -1; /* -1 = frame runs to the partition end */ + + /* + * 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(); @@ -1847,13 +1840,13 @@ 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 (ctx->states == NULL) continue; /* Check frame boundary - finalize the context when it is reached */ - if (hasLimitedFrame) + if (frameOffset >= 0) { int64 ctxFrameEnd; @@ -1911,46 +1904,18 @@ 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; - /* - * Phase 1 already handled frame boundary exceeded contexts by forcing - * mismatch (nfa_match with NULL), which removes all states (all - * states are at VAR positions after advance). So any surviving - * context here must be within its frame boundary. - * - * Compute the (clamped) frame end the same way as Phase 1, using two - * separately checked adds so that "frameOffset + 1" cannot overflow - * when frameOffset is near PG_INT64_MAX. - */ -#ifdef USE_ASSERT_CHECKING - if (hasLimitedFrame) - { - int64 ctxFrameEnd; - - if (pg_add_s64_overflow(ctx->matchStartRow, frameOffset, - &ctxFrameEnd) || - pg_add_s64_overflow(ctxFrameEnd, 1, &ctxFrameEnd)) - ctxFrameEnd = PG_INT64_MAX; - Assert(currentPos < ctxFrameEnd); - } -#endif - nfa_advance(winstate, ctx, currentPos); } } @@ -1994,9 +1959,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); diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c index 9ab73c614dc..f5ec842c45b 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 */ @@ -2555,12 +2554,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)) { + /* + * Under SKIP TO NEXT ROW, clear the recorded match so this + * row is matched again from its own start. + */ if (winstate->rpSkipTo == ST_NEXT_ROW) clear_reduced_frame(winstate); @@ -4545,9 +4544,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) @@ -4555,30 +4551,35 @@ get_reduced_frame_status(WindowAggState *winstate, int64 pos) int64 start = winstate->rpr_match_start; int64 length = winstate->rpr_match_length; + Assert(pos >= 0); + Assert(start < 0 || length >= -1); + + /* cleared slot: no result recorded yet */ if (start < 0) - return RF_NOT_DETERMINED; /* cleared slot: no result recorded yet */ + return RF_NOT_DETERMINED; /* - * 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. + * The record's own row: the length gives the verdict directly, and no + * other row can produce any of these three. */ - if (length == -1) - return (pos == start) ? RF_UNMATCHED : RF_NOT_DETERMINED; - if (length == 0) - return (pos == start) ? RF_EMPTY_MATCH : RF_NOT_DETERMINED; + if (pos == start) + { + if (length == -1) + return RF_UNMATCHED; + if (length == 0) + return RF_EMPTY_MATCH; + return RF_FRAME_HEAD; + } /* - * By here length >= 1, so [start, start + length) is a well-formed range. + * Any other row is covered only by a real match, over [start, start + + * length). The sentinels need no special casing: -1 and 0 leave that + * range empty, so every pos != start falls out here. */ if (pos < start || pos >= start + length) return RF_NOT_DETERMINED; - /* pos lies within a real match. */ - if (pos == start) - return RF_FRAME_HEAD; - + /* inside a real match, after its head */ return RF_SKIPPED; } @@ -4640,8 +4641,7 @@ 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; @@ -4650,11 +4650,12 @@ advance_reduced_frame_nfa(WindowObject winobj, RPRNFAContext *targetCtx, /* * 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). + * matchStartRow since contexts are created at currentPos+1 during + * processing. However, matchStartRow 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 @@ -4701,7 +4702,7 @@ advance_reduced_frame_nfa(WindowObject winobj, RPRNFAContext *targetCtx, * 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 @@ -4742,40 +4743,32 @@ 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; + + /* + * 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. Later contexts can't apply: the list + * ascends by matchStartRow. + */ + if (winstate->nfaContext->matchStartRow == pos) + targetCtx = winstate->nfaContext; } - /* - * Case 2: Find existing context for this pos, or create new one. - */ - targetCtx = ExecRPRGetHeadContext(winstate, pos); if (targetCtx == NULL) { /* @@ -4784,67 +4777,54 @@ 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); - -register_result: + /* + * Either branch above settles targetCtx on pos, which the driver relies + * on to resume from and which the result recorded at the top is keyed by. + */ 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. + * Drive the NFA forward, unless this context already finished in an + * earlier call. That happens in any skip mode: the driver runs rows on + * behalf of an older context, and an overlapping context can complete + * before the call for its own start row arrives. The result it recorded + * is registered below. */ - winstate->rpr_match_start = targetCtx->matchStartRow; + if (targetCtx->states != NULL) + advance_reduced_frame_nfa(winobj, targetCtx); - 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 */ - winstate->rpr_match_length = 0; - ExecRPRRecordContextSuccess(winstate, 0); - } - else - { - /* No match */ - winstate->rpr_match_length = -1; - ExecRPRRecordContextFailure(winstate, matchLen); - } - ExecRPRFreeContext(winstate, targetCtx); - return; + /* No match */ + winstate->rpr_match_length = -1; + ExecRPRRecordContextFailure(winstate, + targetCtx->lastProcessedRow - targetCtx->matchStartRow + 1); } + else + { + /* + * Match: an empty one ends at matchStartRow - 1, so the row count + * comes out 0 with no case of its own. Nothing ends earlier than + * that -- FIN either consumes rows or is reached before the first. + */ + Assert(targetCtx->matchEndRow >= targetCtx->matchStartRow - 1); - /* Match succeeded */ - matchLen = targetCtx->matchEndRow - targetCtx->matchStartRow + 1; + winstate->rpr_match_length = + 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 187666a59cd..2d058e63bf5 100644 --- a/src/backend/parser/parse_rpr.c +++ b/src/backend/parser/parse_rpr.c @@ -124,6 +124,8 @@ transformRPR(ParseState *pstate, WindowClause *wc, WindowDef *windef, parser_errposition(pstate, 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 fb7dc63a4c6..265a53efc57 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);