public inbox for [email protected]
help / color / mirror / Atom feed[PATCH v17 5/8] Row pattern recognition patch (executor).
3+ messages / 2 participants
[nested] [flat]
* [PATCH v17 5/8] Row pattern recognition patch (executor).
@ 2024-04-28 11:00 Tatsuo Ishii <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Tatsuo Ishii @ 2024-04-28 11:00 UTC (permalink / raw)
---
src/backend/executor/nodeWindowAgg.c | 1610 +++++++++++++++++++++++++-
src/backend/utils/adt/windowfuncs.c | 37 +-
src/include/catalog/pg_proc.dat | 6 +
src/include/nodes/execnodes.h | 30 +
4 files changed, 1671 insertions(+), 12 deletions(-)
diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c
index 3221fa1522..140bb3941e 100644
--- a/src/backend/executor/nodeWindowAgg.c
+++ b/src/backend/executor/nodeWindowAgg.c
@@ -36,6 +36,7 @@
#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"
@@ -48,6 +49,7 @@
#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"
@@ -159,6 +161,43 @@ typedef struct WindowStatePerAggData
bool restart; /* need to restart this agg in this cycle? */
} WindowStatePerAggData;
+/*
+ * Set of StringInfo. Used in RPR.
+ */
+typedef struct StringSet
+{
+ StringInfo *str_set;
+ Size set_size; /* current array allocation size in number of
+ * items */
+ int set_index; /* current used size */
+} StringSet;
+
+/*
+ * Allowed subsequent PATTERN variables positions.
+ * Used in RPR.
+ *
+ * pos represents the pattern variable defined order in DEFINE caluase. For
+ * example. "DEFINE START..., UP..., DOWN ..." and "PATTERN START UP DOWN UP"
+ * will create:
+ * VariablePos[0].pos[0] = 0; START
+ * VariablePos[1].pos[0] = 1; UP
+ * VariablePos[1].pos[1] = 3; UP
+ * VariablePos[2].pos[0] = 2; DOWN
+ *
+ * Note that UP has two pos because UP appears in PATTERN twice.
+ *
+ * By using this strucrture, we can know which pattern variable can be followed
+ * by which pattern variable(s). For example, START can be followed by UP and
+ * DOWN since START's pos is 0, and UP's pos is 1 or 3, DOWN's pos is 2.
+ * DOWN can be followed by UP since UP's pos is either 1 or 3.
+ *
+ */
+#define NUM_ALPHABETS 26 /* we allow [a-z] variable initials */
+typedef struct VariablePos
+{
+ int pos[NUM_ALPHABETS]; /* postion(s) in PATTERN */
+} VariablePos;
+
static void initialize_windowaggregate(WindowAggState *winstate,
WindowStatePerFunc perfuncstate,
WindowStatePerAgg peraggstate);
@@ -184,6 +223,7 @@ static void release_partition(WindowAggState *winstate);
static int row_is_in_frame(WindowAggState *winstate, int64 pos,
TupleTableSlot *slot);
+
static void update_frameheadpos(WindowAggState *winstate);
static void update_frametailpos(WindowAggState *winstate);
static void update_grouptailpos(WindowAggState *winstate);
@@ -195,9 +235,48 @@ 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);
+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 int64 evaluate_pattern(WindowObject winobj, int64 current_pos,
+ char *vname, StringInfo encoded_str, bool *result);
+
+static bool get_slots(WindowObject winobj, int64 current_pos);
+
+static int search_str_set(char *pattern, StringSet * str_set,
+ VariablePos * variable_pos);
+static char pattern_initial(WindowAggState *winstate, char *vname);
+static int do_pattern_match(char *pattern, char *encoded_str);
+
+static StringSet * string_set_init(void);
+static void string_set_add(StringSet * string_set, StringInfo str);
+static StringInfo string_set_get(StringSet * string_set, int index);
+static int string_set_get_size(StringSet * string_set);
+static void string_set_discard(StringSet * string_set);
+static VariablePos * variable_pos_init(void);
+static void variable_pos_register(VariablePos * variable_pos, char initial,
+ int pos);
+static bool variable_pos_compare(VariablePos * variable_pos,
+ char initial1, char initial2);
+static int variable_pos_fetch(VariablePos * variable_pos, char initial,
+ int index);
+static void variable_pos_discard(VariablePos * variable_pos);
/*
* initialize_windowaggregate
@@ -774,10 +853,12 @@ 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
* surely is faster.
+ * we restart aggregation too.
*----------
*/
numaggs_restart = 0;
@@ -788,7 +869,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++;
@@ -862,7 +944,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.
@@ -917,6 +1014,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;
}
/*
@@ -930,6 +1035,12 @@ eval_windowaggregates(WindowAggState *winstate)
{
int ret;
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "===== loop in frame starts: aggregatedupto: " INT64_FORMAT " aggregatedbase: " INT64_FORMAT,
+ winstate->aggregatedupto,
+ winstate->aggregatedbase);
+#endif
+
/* Fetch next row if we didn't already */
if (TupIsNull(agg_row_slot))
{
@@ -945,9 +1056,52 @@ eval_windowaggregates(WindowAggState *winstate)
ret = row_is_in_frame(winstate, winstate->aggregatedupto, agg_row_slot);
if (ret < 0)
break;
+
if (ret == 0)
goto next_tuple;
+ if (rpr_is_defined(winstate))
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "reduced_frame_map: %d aggregatedupto: " INT64_FORMAT " aggregatedbase: " INT64_FORMAT,
+ 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
+ elog(DEBUG1, "skip current row for aggregation");
+#endif
+ break;
+ }
+ }
+
/* Set tuple context for evaluation of aggregate arguments */
winstate->tmpcontext->ecxt_outertuple = agg_row_slot;
@@ -976,6 +1130,7 @@ next_tuple:
ExecClearTuple(agg_row_slot);
}
+
/* The frame's end is not supposed to move backwards, ever */
Assert(aggregatedupto_nonrestarted <= winstate->aggregatedupto);
@@ -995,7 +1150,6 @@ next_tuple:
&winstate->perfunc[wfuncno],
peraggstate,
result, isnull);
-
/*
* save the result in case next row shares the same frame.
*
@@ -1090,6 +1244,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;
@@ -2053,6 +2208,11 @@ ExecWindowAgg(PlanState *pstate)
CHECK_FOR_INTERRUPTS();
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "ExecWindowAgg called. pos: " INT64_FORMAT,
+ winstate->currentpos);
+#endif
+
if (winstate->status == WINDOWAGG_DONE)
return NULL;
@@ -2221,6 +2381,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
*/
@@ -2388,6 +2559,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)));
@@ -2486,6 +2660,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
@@ -2667,6 +2851,43 @@ 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 */
+ winstate->patternVariableList = node->patternVariable;
+ winstate->patternRegexpList = node->patternRegexp;
+
+ /* Set up row pattern recognition DEFINE clause */
+ winstate->defineInitial = node->defineInitial;
+ winstate->defineVariableList = NIL;
+ winstate->defineClauseList = NIL;
+ if (node->defineClause != NIL)
+ {
+ /*
+ * Tweak arg var of PREV/NEXT so that it refers to scan/inner slot.
+ */
+ foreach(l, node->defineClause)
+ {
+ char *name;
+ ExprState *exps;
+
+ te = lfirst(l);
+ name = te->resname;
+ expr = te->expr;
+
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "defineVariable name: %s", 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);
+ }
+ }
+
winstate->all_first = true;
winstate->partition_spooled = false;
winstate->more_partitions = false;
@@ -2674,6 +2895,64 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags)
return winstate;
}
+/*
+ * Rewrite varno of Var node that is the argument of PREV/NET so that it sees
+ * scan tuple (PREV) or inner tuple (NEXT).
+ */
+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;
+ Expr *expr;
+ Var *var;
+
+ if (node == NULL)
+ return false;
+
+ if (IsA(node, FuncExpr))
+ {
+ func = (FuncExpr *) node;
+
+ if (func->funcid == F_PREV || func->funcid == F_NEXT)
+ {
+ /* sanity check */
+ 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);
+
+ expr = (Expr *) lfirst(list_head(func->args));
+ if (!IsA(expr, Var))
+ elog(ERROR, "PREV/NEXT's arg is not Var"); /* XXX: is it possible
+ * that arg type is
+ * Const? */
+ var = (Var *) expr;
+
+ if (func->funcid == F_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;
+
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "PREV/NEXT's varno is rewritten to: %d", var->varno);
+#endif
+ }
+ }
+ return expression_tree_walker(node, attno_map_walker, NULL);
+}
+
/* -----------------
* ExecEndWindowAgg
* -----------------
@@ -2723,6 +3002,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)
@@ -3083,7 +3364,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);
@@ -3403,14 +3685,54 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno,
WindowAggState *winstate;
ExprContext *econtext;
TupleTableSlot *slot;
- int64 abs_pos;
- int64 mark_pos;
Assert(WindowObjectIsValid(winobj));
winstate = winobj->winstate;
econtext = winstate->ss.ps.ps_ExprContext;
slot = winstate->temp_slot_1;
+ 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:
@@ -3477,11 +3799,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;
@@ -3548,6 +3884,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);
@@ -3566,15 +3910,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;
}
/*
@@ -3605,3 +3947,1251 @@ 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->patternVariableList != NIL;
+}
+
+/*
+ * -----------------
+ * 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
+ elog(DEBUG1, "row_is_in_reduced_frame returns %d: pos: " INT64_FORMAT,
+ 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
+ elog(DEBUG1, "row_is_in_reduced_frame returns %d: pos: " INT64_FORMAT,
+ 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);
+
+ if (pos < 0 || pos >= winstate->alloc_sz)
+ elog(ERROR, "wrong pos: " INT64_FORMAT, pos);
+
+ 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.
+ */
+static
+void
+update_reduced_frame(WindowObject winobj, int64 pos)
+{
+ WindowAggState *winstate = winobj->winstate;
+ ListCell *lc1,
+ *lc2;
+ bool expression_result;
+ int num_matched_rows;
+ int64 original_pos;
+ bool anymatch;
+ StringInfo encoded_str;
+ StringInfo pattern_str = makeStringInfo();
+ StringSet *str_set;
+ int initial_index;
+ VariablePos *variable_pos;
+ bool greedy = false;
+ int64 result_pos,
+ i;
+
+ /*
+ * Set of pattern variables evaluated to true. Each character corresponds
+ * to pattern variable. Example: str_set[0] = "AB"; str_set[1] = "AC"; In
+ * this case at row 0 A and B are true, and A and C are true in row 1.
+ */
+
+ /* initialize pattern variables set */
+ str_set = string_set_init();
+
+ /* save original pos */
+ original_pos = pos;
+
+ /*
+ * Check if the pattern does not include any greedy quantifier. If it does
+ * not, we can just apply the pattern to each row. If it succeeds, we are
+ * done.
+ */
+ foreach(lc1, winstate->patternRegexpList)
+ {
+ char *quantifier = strVal(lfirst(lc1));
+
+ if (*quantifier == '+' || *quantifier == '*')
+ {
+ greedy = true;
+ break;
+ }
+ }
+
+ /*
+ * Non greedy case
+ */
+ if (!greedy)
+ {
+ num_matched_rows = 0;
+
+ foreach(lc1, winstate->patternVariableList)
+ {
+ char *vname = strVal(lfirst(lc1));
+
+ encoded_str = makeStringInfo();
+
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "pos: " INT64_FORMAT " pattern vname: %s",
+ pos, vname);
+#endif
+ expression_result = false;
+
+ /* evaluate row pattern against current row */
+ result_pos = evaluate_pattern(winobj, pos, vname,
+ encoded_str, &expression_result);
+ if (!expression_result || result_pos < 0)
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "expression result is false or out of frame");
+#endif
+ register_reduced_frame_map(winstate, original_pos,
+ RF_UNMATCHED);
+ return;
+ }
+ /* move to next row */
+ pos++;
+ num_matched_rows++;
+ }
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "pattern matched");
+#endif
+ register_reduced_frame_map(winstate, original_pos, RF_FRAME_HEAD);
+
+ for (i = original_pos + 1; i < original_pos + num_matched_rows; i++)
+ {
+ register_reduced_frame_map(winstate, i, RF_SKIPPED);
+ }
+ return;
+ }
+
+ /*
+ * Greedy quantifiers included. Loop over until none of pattern matches or
+ * encounters end of frame.
+ */
+ for (;;)
+ {
+ result_pos = -1;
+
+ /*
+ * Loop over each PATTERN variable.
+ */
+ anymatch = false;
+ encoded_str = makeStringInfo();
+
+ forboth(lc1, winstate->patternVariableList, lc2,
+ winstate->patternRegexpList)
+ {
+ char *vname = strVal(lfirst(lc1));
+#ifdef RPR_DEBUG
+ char *quantifier = strVal(lfirst(lc2));
+
+ elog(DEBUG1, "pos: " INT64_FORMAT " pattern vname: %s quantifier: %s",
+ pos, vname, quantifier);
+#endif
+ expression_result = false;
+
+ /* evaluate row pattern against current row */
+ result_pos = evaluate_pattern(winobj, pos, vname,
+ encoded_str, &expression_result);
+ if (expression_result)
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "expression result is true");
+#endif
+ anymatch = true;
+ }
+
+ /*
+ * If out of frame, we are done.
+ */
+ if (result_pos < 0)
+ break;
+ }
+
+ if (!anymatch)
+ {
+ /* none of patterns matched. */
+ break;
+ }
+
+ string_set_add(str_set, encoded_str);
+
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "pos: " INT64_FORMAT " encoded_str: %s",
+ encoded_str->data);
+#endif
+
+ /* move to next row */
+ pos++;
+
+ if (result_pos < 0)
+ {
+ /* out of frame */
+ break;
+ }
+ }
+
+ if (string_set_get_size(str_set) == 0)
+ {
+ /* no match found in the first row */
+ register_reduced_frame_map(winstate, original_pos, RF_UNMATCHED);
+ return;
+ }
+
+#ifdef RPR_DEBUG
+ elog(DEBUG2, "pos: " INT64_FORMAT " encoded_str: %s",
+ pos, encoded_str->data);
+#endif
+
+ /* build regular expression */
+ pattern_str = makeStringInfo();
+ appendStringInfoChar(pattern_str, '^');
+ initial_index = 0;
+
+ variable_pos = variable_pos_init();
+
+ forboth(lc1, winstate->patternVariableList,
+ lc2, winstate->patternRegexpList)
+ {
+ char *vname = strVal(lfirst(lc1));
+ char *quantifier = strVal(lfirst(lc2));
+ char initial;
+
+ initial = pattern_initial(winstate, vname);
+ Assert(initial != 0);
+ appendStringInfoChar(pattern_str, initial);
+ if (quantifier[0])
+ appendStringInfoChar(pattern_str, quantifier[0]);
+
+ /*
+ * Register the initial at initial_index. If the initial appears more
+ * than once, all of it's initial_index will be recorded. This could
+ * happen if a pattern variable appears in the PATTERN clause more
+ * than once like "UP DOWN UP" "UP UP UP".
+ */
+ variable_pos_register(variable_pos, initial, initial_index);
+
+ initial_index++;
+ }
+
+#ifdef RPR_DEBUG
+ elog(DEBUG2, "pos: " INT64_FORMAT " pattern: %s",
+ pos, pattern_str->data);
+#endif
+
+ /* look for matching pattern variable sequence */
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "search_str_set started");
+#endif
+ num_matched_rows = search_str_set(pattern_str->data,
+ str_set, variable_pos);
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "search_str_set returns: %d", num_matched_rows);
+#endif
+ variable_pos_discard(variable_pos);
+ string_set_discard(str_set);
+
+ /*
+ * We are at the first row in the reduced frame. Save the number of
+ * matched rows as the number of rows in the reduced frame.
+ */
+ if (num_matched_rows <= 0)
+ {
+ /* no match */
+ register_reduced_frame_map(winstate, original_pos, RF_UNMATCHED);
+ }
+ else
+ {
+ register_reduced_frame_map(winstate, original_pos, RF_FRAME_HEAD);
+
+ for (i = original_pos + 1; i < original_pos + num_matched_rows; i++)
+ {
+ register_reduced_frame_map(winstate, i, RF_SKIPPED);
+ }
+ }
+
+ return;
+}
+
+/*
+ * search_str_set
+ * Perform pattern matching using "pattern" against str_set. pattern is a
+ * regular expression derived from PATTERN clause. Note that the regular
+ * expression string is prefixed by '^' and followed by initials represented
+ * in a same way as str_set. str_set is a set of StringInfo. Each StringInfo
+ * has a string comprising initials of pattern variable strings being true in
+ * a row. The initials are one of [a-y], parallel to the order of variable
+ * names in DEFINE clause. Suppose DEFINE has variables START, UP and DOWN. If
+ * PATTERN has START, UP+ and DOWN, then the initials in PATTERN will be 'a',
+ * 'b' and 'c'. The "pattern" will be "^ab+c".
+ *
+ * variable_pos is an array representing the order of pattern variable string
+ * initials in PATTERN clause. For example initial 'a' potion is in
+ * variable_pos[0].pos[0] = 0. Note that if the pattern is "START UP DOWN UP"
+ * (UP appears twice), then "UP" (initial is 'b') has two position 1 and
+ * 3. Thus variable_pos for b is variable_pos[1].pos[0] = 1 and
+ * variable_pos[1].pos[1] = 3.
+ *
+ * Returns the longest number of the matching rows (greedy matching) if
+ * quatifier '+' or '*' is included in "pattern".
+ */
+static
+int
+search_str_set(char *pattern, StringSet * str_set, VariablePos * variable_pos)
+{
+#define MAX_CANDIDATE_NUM 10000 /* max pattern match candidate size */
+#define FREEZED_CHAR 'Z' /* a pattern is freezed if it ends with the
+ * char */
+#define DISCARD_CHAR 'z' /* a pattern is not need to keep */
+
+ int set_size; /* number of rows in the set */
+ int resultlen;
+ int index;
+ StringSet *old_str_set,
+ *new_str_set;
+ int new_str_size;
+ int len;
+
+ set_size = string_set_get_size(str_set);
+ new_str_set = string_set_init();
+ len = 0;
+ resultlen = 0;
+
+ /*
+ * Generate all possible pattern variable name initials as a set of
+ * StringInfo named "new_str_set". For example, if we have two rows
+ * having "ab" (row 0) and "ac" (row 1) in the input str_set, new_str_set
+ * will have set of StringInfo "aa", "ac", "ba" and "bc" in the end.
+ */
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "pattern: %s set_size: %d", pattern, set_size);
+#endif
+ for (index = 0; index < set_size; index++)
+ {
+ StringInfo str; /* search target row */
+ char *p;
+ int old_set_size;
+ int i;
+
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "index: %d", index);
+#endif
+ if (index == 0)
+ {
+ /* copy variables in row 0 */
+ str = string_set_get(str_set, index);
+ p = str->data;
+
+ /*
+ * Loop over each new pattern variable char.
+ */
+ while (*p)
+ {
+ StringInfo new = makeStringInfo();
+
+ /* add pattern variable char */
+ appendStringInfoChar(new, *p);
+ /* add new one to string set */
+ string_set_add(new_str_set, new);
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "old_str: NULL new_str: %s", new->data);
+#endif
+ p++; /* next pattern variable */
+ }
+ }
+ else /* index != 0 */
+ {
+ old_str_set = new_str_set;
+ new_str_set = string_set_init();
+ str = string_set_get(str_set, index);
+ old_set_size = string_set_get_size(old_str_set);
+
+ /*
+ * Loop over each rows in the previous result set.
+ */
+ for (i = 0; i < old_set_size; i++)
+ {
+ StringInfo new;
+ char last_old_char;
+ int old_str_len;
+ StringInfo old = string_set_get(old_str_set, i);
+
+ p = old->data;
+ old_str_len = strlen(p);
+ if (old_str_len > 0)
+ last_old_char = p[old_str_len - 1];
+ else
+ last_old_char = '\0';
+
+ /* Is this old set freezed? */
+ if (last_old_char == FREEZED_CHAR)
+ {
+ /* if shorter match. we can discard it */
+ if ((old_str_len - 1) < resultlen)
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "discard this old set because shorter match: %s",
+ old->data);
+#endif
+ continue;
+ }
+
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "keep this old set: %s", old->data);
+#endif
+
+ /* move the old set to new_str_set */
+ string_set_add(new_str_set, old);
+ old_str_set->str_set[i] = NULL;
+ continue;
+ }
+ /* Can this old set be discarded? */
+ else if (last_old_char == DISCARD_CHAR)
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "discard this old set: %s", old->data);
+#endif
+ continue;
+ }
+
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "str->data: %s", str->data);
+#endif
+
+ /*
+ * loop over each pattern variable initial char in the input
+ * set.
+ */
+ for (p = str->data; *p; p++)
+ {
+ /*
+ * Optimization. Check if the row's pattern variable
+ * initial character position is greater than or equal to
+ * the old set's last pattern variable initial character
+ * position. For example, if the old set's last pattern
+ * variable initials are "ab", then the new pattern
+ * variable initial can be "b" or "c" but can not be "a",
+ * if the initials in PATTERN is something like "a b c" or
+ * "a b+ c+" etc. This optimization is possible when we
+ * only allow "+" quantifier.
+ */
+ if (variable_pos_compare(variable_pos, last_old_char, *p))
+ {
+ /* copy source string */
+ new = makeStringInfo();
+ enlargeStringInfo(new, old->len + 1);
+ appendStringInfoString(new, old->data);
+ /* add pattern variable char */
+ appendStringInfoChar(new, *p);
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "old_str: %s new_str: %s",
+ old->data, new->data);
+#endif
+
+ /*
+ * Adhoc optimization. If the first letter in the
+ * input string is the first and second position one
+ * and there's no associated quatifier '+', then we
+ * can dicard the input because there's no chace to
+ * expand the string further.
+ *
+ * For example, pattern "abc" cannot match "aa".
+ */
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "pattern[1]:%c pattern[2]:%c new[0]:%c new[1]:%c",
+ pattern[1], pattern[2], new->data[0], new->data[1]);
+#endif
+ if (pattern[1] == new->data[0] &&
+ pattern[1] == new->data[1] &&
+ pattern[2] != '+' &&
+ pattern[1] != pattern[2])
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "discard this new data: %s",
+ new->data);
+#endif
+ pfree(new->data);
+ pfree(new);
+ continue;
+ }
+
+ /* add new one to string set */
+ string_set_add(new_str_set, new);
+ }
+ else
+ {
+ /*
+ * We are freezing this pattern string. Since there's
+ * no chance to expand the string further, we perform
+ * pattern matching against the string. If it does not
+ * match, we can discard it.
+ */
+ len = do_pattern_match(pattern, old->data);
+
+ if (len <= 0)
+ {
+ /* no match. we can discard it */
+ continue;
+ }
+
+ else if (len <= resultlen)
+ {
+ /* shorter match. we can discard it */
+ continue;
+ }
+ else
+ {
+ /* match length is the longest so far */
+
+ int new_index;
+
+ /* remember the longest match */
+ resultlen = len;
+
+ /* freeze the pattern string */
+ new = makeStringInfo();
+ enlargeStringInfo(new, old->len + 1);
+ appendStringInfoString(new, old->data);
+ /* add freezed mark */
+ appendStringInfoChar(new, FREEZED_CHAR);
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "old_str: %s new_str: %s", old->data, new->data);
+#endif
+ string_set_add(new_str_set, new);
+
+ /*
+ * Search new_str_set to find out freezed entries
+ * that have shorter match length. Mark them as
+ * "discard" so that they are discarded in the
+ * next round.
+ */
+
+ /* new_index_size should be the one before */
+ new_str_size =
+ string_set_get_size(new_str_set) - 1;
+
+ /* loop over new_str_set */
+ for (new_index = 0; new_index < new_str_size;
+ new_index++)
+ {
+ char new_last_char;
+ int new_str_len;
+
+ new = string_set_get(new_str_set, new_index);
+ new_str_len = strlen(new->data);
+ if (new_str_len > 0)
+ {
+ new_last_char =
+ new->data[new_str_len - 1];
+ if (new_last_char == FREEZED_CHAR &&
+ (new_str_len - 1) <= len)
+ {
+ /*
+ * mark this set to discard in the
+ * next round
+ */
+ appendStringInfoChar(new, DISCARD_CHAR);
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "add discard char: %s", new->data);
+#endif
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ /* we no longer need old string set */
+ string_set_discard(old_str_set);
+ }
+ }
+
+ /*
+ * Perform pattern matching to find out the longest match.
+ */
+ new_str_size = string_set_get_size(new_str_set);
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "new_str_size: %d", new_str_size);
+#endif
+ len = 0;
+ resultlen = 0;
+
+ for (index = 0; index < new_str_size; index++)
+ {
+ StringInfo s;
+
+ s = string_set_get(new_str_set, index);
+ if (s == NULL)
+ continue; /* no data */
+
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "target string: %s", s->data);
+#endif
+ len = do_pattern_match(pattern, s->data);
+ if (len > resultlen)
+ {
+ /* remember the longest match */
+ resultlen = len;
+
+ /*
+ * If the size of result set is equal to the number of rows in the
+ * set, we are done because it's not possible that the number of
+ * matching rows exceeds the number of rows in the set.
+ */
+ if (resultlen >= set_size)
+ break;
+ }
+ }
+
+ /* we no longer need new string set */
+ string_set_discard(new_str_set);
+
+ return resultlen;
+}
+
+/*
+ * do_pattern_match
+ * perform pattern match using pattern against encoded_str.
+ * returns matching number of rows if matching is succeeded.
+ * Otherwise returns 0.
+ */
+static
+int
+do_pattern_match(char *pattern, char *encoded_str)
+{
+ Datum d;
+ text *res;
+ char *substr;
+ int len = 0;
+ text *pattern_text,
+ *encoded_str_text;
+
+ pattern_text = cstring_to_text(pattern);
+ encoded_str_text = cstring_to_text(encoded_str);
+
+ /*
+ * We first perform pattern matching using regexp_instr, then call
+ * textregexsubstr to get matched substring to know how long the matched
+ * string is. That is the number of rows in the reduced window frame. The
+ * reason why we can't call textregexsubstr in the first place is, it
+ * errors out if pattern does not match.
+ */
+ if (DatumGetInt32(DirectFunctionCall2Coll(
+ regexp_instr, DEFAULT_COLLATION_OID,
+ PointerGetDatum(encoded_str_text),
+ PointerGetDatum(pattern_text))))
+ {
+ d = DirectFunctionCall2Coll(textregexsubstr,
+ DEFAULT_COLLATION_OID,
+ PointerGetDatum(encoded_str_text),
+ PointerGetDatum(pattern_text));
+ if (d != 0)
+ {
+ res = DatumGetTextPP(d);
+ substr = text_to_cstring(res);
+ len = strlen(substr);
+ pfree(substr);
+ }
+ }
+ pfree(encoded_str_text);
+ pfree(pattern_text);
+
+ return len;
+}
+
+/*
+ * evaluate_pattern
+ * Evaluate expression associated with PATTERN variable vname. current_pos is
+ * relative row position in a frame (starting from 0). If vname is evaluated
+ * to true, initial letters associated with vname is appended to
+ * encode_str. result is out paramater representing the expression evaluation
+ * result is true of false.
+ *---------
+ * Return values are:
+ * >=0: the last match absolute row position
+ * otherwise out of frame.
+ *---------
+ */
+static
+int64
+evaluate_pattern(WindowObject winobj, int64 current_pos,
+ char *vname, StringInfo encoded_str, bool *result)
+{
+ WindowAggState *winstate = winobj->winstate;
+ ExprContext *econtext = winstate->ss.ps.ps_ExprContext;
+ ListCell *lc1,
+ *lc2,
+ *lc3;
+ ExprState *pat;
+ Datum eval_result;
+ bool out_of_frame = false;
+ bool isnull;
+ TupleTableSlot *slot;
+
+ forthree(lc1, winstate->defineVariableList,
+ lc2, winstate->defineClauseList,
+ lc3, winstate->defineInitial)
+ {
+ char initial; /* initial letter associated with vname */
+ char *name = strVal(lfirst(lc1));
+
+ if (strcmp(vname, name))
+ continue;
+
+ initial = *(strVal(lfirst(lc3)));
+
+ /* set expression to evaluate */
+ pat = lfirst(lc2);
+
+ /* get current, previous and next tuples */
+ if (!get_slots(winobj, current_pos))
+ {
+ out_of_frame = true;
+ }
+ else
+ {
+ /* evaluate the expression */
+ eval_result = ExecEvalExpr(pat, econtext, &isnull);
+ if (isnull)
+ {
+ /* expression is NULL */
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "expression for %s is NULL at row: " INT64_FORMAT,
+ vname, current_pos);
+#endif
+ *result = false;
+ }
+ else
+ {
+ if (!DatumGetBool(eval_result))
+ {
+ /* expression is false */
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "expression for %s is false at row: " INT64_FORMAT,
+ vname, current_pos);
+#endif
+ *result = false;
+ }
+ else
+ {
+ /* expression is true */
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "expression for %s is true at row: " INT64_FORMAT,
+ vname, current_pos);
+#endif
+ appendStringInfoChar(encoded_str, initial);
+ *result = true;
+ }
+ }
+
+ slot = winstate->temp_slot_1;
+ if (slot != winstate->null_slot)
+ ExecClearTuple(slot);
+ slot = winstate->prev_slot;
+ if (slot != winstate->null_slot)
+ ExecClearTuple(slot);
+ slot = winstate->next_slot;
+ if (slot != winstate->null_slot)
+ ExecClearTuple(slot);
+
+ break;
+ }
+
+ if (out_of_frame)
+ {
+ *result = false;
+ return -1;
+ }
+ }
+ return current_pos;
+}
+
+/*
+ * get_slots
+ * Get current, previous and next tuples.
+ * Returns false if current row is out of partition/full frame.
+ */
+static
+bool
+get_slots(WindowObject winobj, int64 current_pos)
+{
+ WindowAggState *winstate = winobj->winstate;
+ TupleTableSlot *slot;
+ int ret;
+ ExprContext *econtext;
+
+ econtext = winstate->ss.ps.ps_ExprContext;
+
+ /* set up current row tuple slot */
+ slot = winstate->temp_slot_1;
+ if (!window_gettupleslot(winobj, current_pos, slot))
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "current row is out of partition at:" INT64_FORMAT,
+ current_pos);
+#endif
+ return false;
+ }
+ ret = row_is_in_frame(winstate, current_pos, slot);
+ if (ret <= 0)
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "current row is out of frame at: " INT64_FORMAT,
+ current_pos);
+#endif
+ ExecClearTuple(slot);
+ return false;
+ }
+ econtext->ecxt_outertuple = slot;
+
+ /* for PREV */
+ if (current_pos > 0)
+ {
+ slot = winstate->prev_slot;
+ if (!window_gettupleslot(winobj, current_pos - 1, slot))
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "previous row is out of partition at: " INT64_FORMAT,
+ current_pos - 1);
+#endif
+ econtext->ecxt_scantuple = winstate->null_slot;
+ }
+ else
+ {
+ ret = row_is_in_frame(winstate, current_pos - 1, slot);
+ if (ret <= 0)
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "previous row is out of frame at: " INT64_FORMAT,
+ current_pos - 1);
+#endif
+ ExecClearTuple(slot);
+ econtext->ecxt_scantuple = winstate->null_slot;
+ }
+ else
+ {
+ econtext->ecxt_scantuple = slot;
+ }
+ }
+ }
+ else
+ econtext->ecxt_scantuple = winstate->null_slot;
+
+ /* for NEXT */
+ slot = winstate->next_slot;
+ if (!window_gettupleslot(winobj, current_pos + 1, slot))
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "next row is out of partiton at: " INT64_FORMAT,
+ current_pos + 1);
+#endif
+ econtext->ecxt_innertuple = winstate->null_slot;
+ }
+ else
+ {
+ ret = row_is_in_frame(winstate, current_pos + 1, slot);
+ if (ret <= 0)
+ {
+#ifdef RPR_DEBUG
+ elog(DEBUG1, "next row is out of frame at: " INT64_FORMAT,
+ current_pos + 1);
+#endif
+ ExecClearTuple(slot);
+ econtext->ecxt_innertuple = winstate->null_slot;
+ }
+ else
+ econtext->ecxt_innertuple = slot;
+ }
+ return true;
+}
+
+/*
+ * pattern_initial
+ * Return pattern variable initial character
+ * matching with pattern variable name vname.
+ * If not found, return 0.
+ */
+static
+char
+pattern_initial(WindowAggState *winstate, char *vname)
+{
+ char initial;
+ char *name;
+ ListCell *lc1,
+ *lc2;
+
+ forboth(lc1, winstate->defineVariableList,
+ lc2, winstate->defineInitial)
+ {
+ name = strVal(lfirst(lc1)); /* DEFINE variable name */
+ initial = *(strVal(lfirst(lc2))); /* DEFINE variable initial */
+
+
+ if (!strcmp(name, vname))
+ return initial; /* found */
+ }
+ return 0;
+}
+
+/*
+ * string_set_init
+ * Create dynamic set of StringInfo.
+ */
+static
+StringSet * string_set_init(void)
+{
+/* Initial allocation size of str_set */
+#define STRING_SET_ALLOC_SIZE 1024
+
+ StringSet *string_set;
+ Size set_size;
+
+ string_set = palloc0(sizeof(StringSet));
+ string_set->set_index = 0;
+ set_size = STRING_SET_ALLOC_SIZE;
+ string_set->str_set = palloc(set_size * sizeof(StringInfo));
+ string_set->set_size = set_size;
+
+ return string_set;
+}
+
+/*
+ * string_set_add
+ * Add StringInfo str to StringSet string_set.
+ */
+static
+void
+string_set_add(StringSet * string_set, StringInfo str)
+{
+ Size set_size;
+
+ set_size = string_set->set_size;
+ if (string_set->set_index >= set_size)
+ {
+ set_size *= 2;
+ string_set->str_set = repalloc(string_set->str_set,
+ set_size * sizeof(StringInfo));
+ string_set->set_size = set_size;
+ }
+
+ string_set->str_set[string_set->set_index++] = str;
+
+ return;
+}
+
+/*
+ * string_set_get
+ * Returns StringInfo specified by index.
+ * If there's no data yet, returns NULL.
+ */
+static
+StringInfo
+string_set_get(StringSet * string_set, int index)
+{
+ /* no data? */
+ if (index == 0 && string_set->set_index == 0)
+ return NULL;
+
+ if (index < 0 || index >= string_set->set_index)
+ elog(ERROR, "invalid index: %d", index);
+
+ return string_set->str_set[index];
+}
+
+/*
+ * string_set_get_size
+ * Returns the size of StringSet.
+ */
+static
+int
+string_set_get_size(StringSet * string_set)
+{
+ return string_set->set_index;
+}
+
+/*
+ * string_set_discard
+ * Discard StringSet.
+ * All memory including StringSet itself is freed.
+ */
+static
+void
+string_set_discard(StringSet * string_set)
+{
+ int i;
+
+ for (i = 0; i < string_set->set_index; i++)
+ {
+ StringInfo str = string_set->str_set[i];
+
+ if (str)
+ {
+ pfree(str->data);
+ pfree(str);
+ }
+ }
+ pfree(string_set->str_set);
+ pfree(string_set);
+}
+
+/*
+ * variable_pos_init
+ * Create and initialize variable postion structure
+ */
+static
+VariablePos * variable_pos_init(void)
+{
+ VariablePos *variable_pos;
+
+ variable_pos = palloc(sizeof(VariablePos) * NUM_ALPHABETS);
+ MemSet(variable_pos, -1, sizeof(VariablePos) * NUM_ALPHABETS);
+ return variable_pos;
+}
+
+/*
+ * variable_pos_register
+ * Register pattern variable whose initial is initial into postion index.
+ * pos is position of initial.
+ * If pos is already registered, register it at next empty slot.
+ */
+static
+void
+variable_pos_register(VariablePos * variable_pos, char initial, int pos)
+{
+ int index = initial - 'a';
+ int slot;
+ int i;
+
+ if (pos < 0 || pos > NUM_ALPHABETS)
+ elog(ERROR, "initial is not valid char: %c", initial);
+
+ for (i = 0; i < NUM_ALPHABETS; i++)
+ {
+ slot = variable_pos[index].pos[i];
+ if (slot < 0)
+ {
+ /* empty slot found */
+ variable_pos[index].pos[i] = pos;
+ return;
+ }
+ }
+ elog(ERROR, "no empty slot for initial: %c", initial);
+}
+
+/*
+ * variable_pos_compare
+ * Returns true if initial1 can be followed by initial2
+ */
+static
+bool
+variable_pos_compare(VariablePos * variable_pos, char initial1, char initial2)
+{
+ int index1,
+ index2;
+ int pos1,
+ pos2;
+
+ for (index1 = 0;; index1++)
+ {
+ pos1 = variable_pos_fetch(variable_pos, initial1, index1);
+ if (pos1 < 0)
+ break;
+
+ for (index2 = 0;; index2++)
+ {
+ pos2 = variable_pos_fetch(variable_pos, initial2, index2);
+ if (pos2 < 0)
+ break;
+ if (pos1 <= pos2)
+ return true;
+ }
+ }
+ return false;
+}
+
+/*
+ * variable_pos_fetch
+ * Fetch position of pattern variable whose initial is initial, and whose index
+ * is index. If no postion was registered by initial, index, returns -1.
+ */
+static
+int
+variable_pos_fetch(VariablePos * variable_pos, char initial, int index)
+{
+ int pos = initial - 'a';
+
+ if (pos < 0 || pos > NUM_ALPHABETS)
+ elog(ERROR, "initial is not valid char: %c", initial);
+
+ if (index < 0 || index > NUM_ALPHABETS)
+ elog(ERROR, "index is not valid: %d", index);
+
+ return variable_pos[pos].pos[index];
+}
+
+/*
+ * variable_pos_discard
+ * Discard VariablePos
+ */
+static
+void
+variable_pos_discard(VariablePos * variable_pos)
+{
+ pfree(variable_pos);
+}
diff --git a/src/backend/utils/adt/windowfuncs.c b/src/backend/utils/adt/windowfuncs.c
index 473c61569f..92c528d38c 100644
--- a/src/backend/utils/adt/windowfuncs.c
+++ b/src/backend/utils/adt/windowfuncs.c
@@ -13,6 +13,9 @@
*/
#include "postgres.h"
+#include "catalog/pg_collation_d.h"
+#include "executor/executor.h"
+#include "nodes/execnodes.h"
#include "nodes/parsenodes.h"
#include "nodes/supportnodes.h"
#include "utils/fmgrprotos.h"
@@ -37,11 +40,19 @@ typedef struct
int64 remainder; /* (total rows) % (bucket num) */
} ntile_context;
+/*
+ * rpr process information.
+ * Used for AFTER MATCH SKIP PAST LAST ROW
+ */
+typedef struct SkipContext
+{
+ int64 pos; /* last row absolute position */
+} SkipContext;
+
static bool rank_up(WindowObject winobj);
static Datum leadlag_common(FunctionCallInfo fcinfo,
bool forward, bool withoffset, bool withdefault);
-
/*
* utility routine for *_rank functions.
*/
@@ -674,7 +685,7 @@ window_last_value(PG_FUNCTION_ARGS)
bool isnull;
result = WinGetFuncArgInFrame(winobj, 0,
- 0, WINDOW_SEEK_TAIL, true,
+ 0, WINDOW_SEEK_TAIL, false,
&isnull, NULL);
if (isnull)
PG_RETURN_NULL();
@@ -714,3 +725,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 134e3b22fd..5f7fb538f9 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -10477,6 +10477,12 @@
{ oid => '3114', descr => 'fetch the Nth row value',
proname => 'nth_value', prokind => 'w', prorettype => 'anyelement',
proargtypes => 'anyelement int4', prosrc => 'window_nth_value' },
+{ oid => '6122', descr => 'previous value',
+ proname => 'prev', provolatile => 's', prorettype => 'anyelement',
+ proargtypes => 'anyelement', prosrc => 'window_prev' },
+{ oid => '6123', 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 d927ac44a8..971d8682b1 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2550,6 +2550,11 @@ 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
+
typedef struct WindowAggState
{
ScanState ss; /* its first field is NodeTag */
@@ -2598,6 +2603,19 @@ 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 */
+ List *patternVariableList; /* list of row pattern variables names
+ * (list of String) */
+ List *patternRegexpList; /* list of row pattern regular expressions
+ * ('+' or ''. list of String) */
+ List *defineVariableList; /* list of row pattern definition
+ * variables (list of String) */
+ List *defineClauseList; /* expression for row pattern definition
+ * search conditions ExprState list */
+ List *defineInitial; /* list of row pattern definition variable
+ * initials (list of String) */
+
MemoryContext partcontext; /* context for partition-lifespan data */
MemoryContext aggcontext; /* shared context for aggregate working data */
MemoryContext curaggcontext; /* current aggregate's working data */
@@ -2634,6 +2652,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_*
+ */
+ char *reduced_frame_map;
+ int64 alloc_sz; /* size of the map */
} WindowAggState;
/* ----------------
--
2.25.1
----Next_Part(Sun_Apr_28_20_28_26_2024_444)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v17-0006-Row-pattern-recognition-patch-docs.patch"
^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Limit memory usage by postgres_fdw batches
@ 2026-01-13 10:44 Alexander Pyhalov <[email protected]>
0 siblings, 1 reply; 3+ messages in thread
From: Alexander Pyhalov @ 2026-01-13 10:44 UTC (permalink / raw)
To: Tomas Vondra <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>
Hi.
I've looked at the third patch more and found some evident issues.
1) While using tuplestore we get too much garbage from tuple conversion,
which was not cleared properly. Tried to fix it, but now we come to the
second problem.
2) While receiving tuples we've already allocated memory in libpq, so it
can be too late to do something. In simple examples I can see libpq
result sets which have several GB size (and likely are not limited by
anything).
I've used the attached script to create some tables.
select create_partitioned_rel('t1', 128, true, 1);
select create_partitioned_rel('t2', 4, true, 100);
insert into t1 select i, pg_read_binary_file('/some/100mb/file') from
generate_series(1,128) i;
insert into t2 select i, pg_read_binary_file('/some/100mb/file') from
generate_series(1,128) i;
And with simple queries like
select sum(length(s)) from TABLE;
we can see that backend can easily consume up to several GBs of RAM.
For now I start thinking we need some form of FETCH, which stops
fetching data based on batch size...
--
Best regards,
Alexander Pyhalov,
Postgres Professional
SELECT current_database() AS current_database,
current_setting('port') AS current_port
\gset
CREATE EXTENSION postgres_fdw;
CREATE SERVER loopback1 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (dbname :'current_database', port :'current_port', async_capable 'on', fetch_size '1');
CREATE SERVER loopback2 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (dbname :'current_database', port :'current_port', async_capable 'on', fetch_size '1');
CREATE SERVER loopback3 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (dbname :'current_database', port :'current_port', async_capable 'on', fetch_size '1');
CREATE SERVER loopback4 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (dbname :'current_database', port :'current_port', async_capable 'on', fetch_size '1');
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback1;
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback2;
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback3;
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback4;
create or replace function create_partitioned_rel(relname text, nparts int, foreign_parts bool, fetch_size int) returns void as $$
declare
i int;
sql text;
part_name text;
fdw_part_name text;
begin
execute format('create table %I (i int, s bytea) partition by hash(i)', relname);
for i in 0..nparts-1 loop
if foreign_parts then
part_name := format('%s_%s', relname, i);
fdw_part_name := format('%s_%s_fdw', relname, i);
execute format('create foreign table %I partition of %I for values with (modulus %s, remainder %s) server loopback%s options (table_name %L, fetch_size %L)', fdw_part_name, relname, nparts, i, i%4+1, part_name, fetch_size);
execute format('create table %I (i int, s bytea)', part_name);
else
part_name := format('%s_%s', relname, i);
execute format('create table %I partition of %I for values with (modulus %s, remainder %s)', part_name, relname, nparts, i);
end if;
end loop;
end
$$ language plpgsql;
Attachments:
[text/plain] create_test_example.sql (2.0K, ../../[email protected]/2-create_test_example.sql)
download | inline:
SELECT current_database() AS current_database,
current_setting('port') AS current_port
\gset
CREATE EXTENSION postgres_fdw;
CREATE SERVER loopback1 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (dbname :'current_database', port :'current_port', async_capable 'on', fetch_size '1');
CREATE SERVER loopback2 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (dbname :'current_database', port :'current_port', async_capable 'on', fetch_size '1');
CREATE SERVER loopback3 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (dbname :'current_database', port :'current_port', async_capable 'on', fetch_size '1');
CREATE SERVER loopback4 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (dbname :'current_database', port :'current_port', async_capable 'on', fetch_size '1');
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback1;
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback2;
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback3;
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback4;
create or replace function create_partitioned_rel(relname text, nparts int, foreign_parts bool, fetch_size int) returns void as $$
declare
i int;
sql text;
part_name text;
fdw_part_name text;
begin
execute format('create table %I (i int, s bytea) partition by hash(i)', relname);
for i in 0..nparts-1 loop
if foreign_parts then
part_name := format('%s_%s', relname, i);
fdw_part_name := format('%s_%s_fdw', relname, i);
execute format('create foreign table %I partition of %I for values with (modulus %s, remainder %s) server loopback%s options (table_name %L, fetch_size %L)', fdw_part_name, relname, nparts, i, i%4+1, part_name, fetch_size);
execute format('create table %I (i int, s bytea)', part_name);
else
part_name := format('%s_%s', relname, i);
execute format('create table %I partition of %I for values with (modulus %s, remainder %s)', part_name, relname, nparts, i);
end if;
end loop;
end
$$ language plpgsql;
[text/x-diff] v3-0001-Limit-batch_size-for-foreign-insert-with-work_mem.patch (2.7K, ../../[email protected]/3-v3-0001-Limit-batch_size-for-foreign-insert-with-work_mem.patch)
download | inline diff:
From 7efc88ffb4d90e935ca1e17ac3d4c753a2de4ef0 Mon Sep 17 00:00:00 2001
From: Alexander Pyhalov <[email protected]>
Date: Tue, 16 Dec 2025 18:46:27 +0300
Subject: [PATCH 1/3] Limit batch_size for foreign insert with work_mem
Option batch_size can be set on foreign server level. It can be very optimistic
even for one table with different tuple lengths. To prevent large memory usage
limit effective batch size with work_mem.
---
src/backend/executor/nodeModifyTable.c | 15 ++++++++++++---
src/include/nodes/execnodes.h | 1 +
2 files changed, 13 insertions(+), 3 deletions(-)
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 874b71e6608..a28beca63ff 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -943,12 +943,18 @@ ExecInsert(ModifyTableContext *context,
if (resultRelInfo->ri_BatchSize > 1)
{
bool flushed = false;
+ Size tuple_len;
+
+ /* Compute length of the current tuple */
+ slot_getallattrs(slot);
+ tuple_len = heap_compute_data_size(slot->tts_tupleDescriptor, slot->tts_values, slot->tts_isnull);
/*
- * When we've reached the desired batch size, perform the
- * insertion.
+ * When we've reached the desired batch size or exceeded work_mem,
+ * perform the insertion.
*/
- if (resultRelInfo->ri_NumSlots == resultRelInfo->ri_BatchSize)
+ if (resultRelInfo->ri_NumSlots == resultRelInfo->ri_BatchSize ||
+ ((tuple_len + resultRelInfo->ri_BatchMemoryUsed > work_mem * 1024) && (resultRelInfo->ri_NumSlots > 0)))
{
ExecBatchInsert(mtstate, resultRelInfo,
resultRelInfo->ri_Slots,
@@ -1019,6 +1025,8 @@ ExecInsert(ModifyTableContext *context,
resultRelInfo));
resultRelInfo->ri_NumSlots++;
+ /* Batch size can grow up to tuple length even if it exceeds work_mem. */
+ resultRelInfo->ri_BatchMemoryUsed += tuple_len;
MemoryContextSwitchTo(oldContext);
@@ -1418,6 +1426,7 @@ ExecBatchInsert(ModifyTableState *mtstate,
ExecClearTuple(planSlots[i]);
}
resultRelInfo->ri_NumSlots = 0;
+ resultRelInfo->ri_BatchMemoryUsed = 0;
}
/*
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 3968429f991..b82ec938228 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -542,6 +542,7 @@ typedef struct ResultRelInfo
int ri_NumSlots; /* number of slots in the array */
int ri_NumSlotsInitialized; /* number of initialized slots */
int ri_BatchSize; /* max slots inserted in a single batch */
+ int ri_BatchMemoryUsed; /* memory used by batch */
TupleTableSlot **ri_Slots; /* input tuples for batch insert */
TupleTableSlot **ri_PlanSlots;
--
2.43.0
[text/x-diff] v3-0002-Fall-back-to-foreign-insert-when-tuple-is-too-big.patch (10.2K, ../../[email protected]/4-v3-0002-Fall-back-to-foreign-insert-when-tuple-is-too-big.patch)
download | inline diff:
From f9a2ec48d907c37259fc6658a6a0ea5925143a42 Mon Sep 17 00:00:00 2001
From: Alexander Pyhalov <[email protected]>
Date: Tue, 30 Dec 2025 10:35:22 +0300
Subject: [PATCH 2/3] Fall back to foreign insert when tuple is too big
When tuple is more then work_mem, we don't form batch,
but fall through to usual foreign insert. This means
that executor can form batch, insert it and finish with
a usual insert, so there will be no pending batch inserts
to process. To avoid manipulation with possibly long lists,
we don't clear es_insert_pending_result_relations and
es_insert_pending_modifytables lists, but check in
ExecPendingInserts() if there are any pending tuples.
---
.../postgres_fdw/expected/postgres_fdw.out | 11 ++
contrib/postgres_fdw/sql/postgres_fdw.sql | 9 ++
src/backend/executor/nodeModifyTable.c | 138 ++++++++++--------
src/include/nodes/execnodes.h | 1 +
4 files changed, 96 insertions(+), 63 deletions(-)
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 48e3185b227..a8bfec9413c 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -7640,6 +7640,17 @@ select count(*) from tab_batch_sharded;
45
(1 row)
+delete from tab_batch_sharded;
+-- test batch insert with large tuples and switching from batch to usual insert
+set work_mem to 64;
+insert into tab_batch_sharded select 3, case when i%4 = 0 then lpad('a',65*1024,'a') else 'test'|| i end from generate_series(1, 100) i;
+select count(*) from tab_batch_sharded;
+ count
+-------
+ 100
+(1 row)
+
+reset work_mem;
drop table tab_batch_local;
drop table tab_batch_sharded;
drop table tab_batch_sharded_p1_remote;
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 9a8f9e28135..467943ca32e 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -1936,10 +1936,19 @@ create foreign table tab_batch_sharded_p1 partition of tab_batch_sharded
server loopback options (table_name 'tab_batch_sharded_p1_remote');
insert into tab_batch_sharded select * from tab_batch_local;
select count(*) from tab_batch_sharded;
+delete from tab_batch_sharded;
+-- test batch insert with large tuples and switching from batch to usual insert
+set work_mem to 64;
+insert into tab_batch_sharded select 3, case when i%4 = 0 then lpad('a',65*1024,'a') else 'test'|| i end from generate_series(1, 100) i;
+select count(*) from tab_batch_sharded;
+reset work_mem;
+
drop table tab_batch_local;
drop table tab_batch_sharded;
drop table tab_batch_sharded_p1_remote;
+
+
alter server loopback options (drop batch_size);
-- ===================================================================
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index a28beca63ff..2e6dcdff7bf 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -942,7 +942,6 @@ ExecInsert(ModifyTableContext *context,
*/
if (resultRelInfo->ri_BatchSize > 1)
{
- bool flushed = false;
Size tuple_len;
/* Compute length of the current tuple */
@@ -961,76 +960,82 @@ ExecInsert(ModifyTableContext *context,
resultRelInfo->ri_PlanSlots,
resultRelInfo->ri_NumSlots,
estate, canSetTag);
- flushed = true;
}
- oldContext = MemoryContextSwitchTo(estate->es_query_cxt);
-
- if (resultRelInfo->ri_Slots == NULL)
+ if (tuple_len < work_mem * 1024)
{
- resultRelInfo->ri_Slots = palloc_array(TupleTableSlot *, resultRelInfo->ri_BatchSize);
- resultRelInfo->ri_PlanSlots = palloc_array(TupleTableSlot *, resultRelInfo->ri_BatchSize);
- }
+ oldContext = MemoryContextSwitchTo(estate->es_query_cxt);
- /*
- * Initialize the batch slots. We don't know how many slots will
- * be needed, so we initialize them as the batch grows, and we
- * keep them across batches. To mitigate an inefficiency in how
- * resource owner handles objects with many references (as with
- * many slots all referencing the same tuple descriptor) we copy
- * the appropriate tuple descriptor for each slot.
- */
- if (resultRelInfo->ri_NumSlots >= resultRelInfo->ri_NumSlotsInitialized)
- {
- TupleDesc tdesc = CreateTupleDescCopy(slot->tts_tupleDescriptor);
- TupleDesc plan_tdesc =
- CreateTupleDescCopy(planSlot->tts_tupleDescriptor);
+ if (resultRelInfo->ri_Slots == NULL)
+ {
+ resultRelInfo->ri_Slots = palloc_array(TupleTableSlot *, resultRelInfo->ri_BatchSize);
+ resultRelInfo->ri_PlanSlots = palloc_array(TupleTableSlot *, resultRelInfo->ri_BatchSize);
+ }
- resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots] =
- MakeSingleTupleTableSlot(tdesc, slot->tts_ops);
+ /*
+ * Initialize the batch slots. We don't know how many slots
+ * will be needed, so we initialize them as the batch grows,
+ * and we keep them across batches. To mitigate an
+ * inefficiency in how resource owner handles objects with
+ * many references (as with many slots all referencing the
+ * same tuple descriptor) we copy the appropriate tuple
+ * descriptor for each slot.
+ */
+ if (resultRelInfo->ri_NumSlots >= resultRelInfo->ri_NumSlotsInitialized)
+ {
+ TupleDesc tdesc = CreateTupleDescCopy(slot->tts_tupleDescriptor);
+ TupleDesc plan_tdesc =
+ CreateTupleDescCopy(planSlot->tts_tupleDescriptor);
- resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots] =
- MakeSingleTupleTableSlot(plan_tdesc, planSlot->tts_ops);
+ resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots] =
+ MakeSingleTupleTableSlot(tdesc, slot->tts_ops);
- /* remember how many batch slots we initialized */
- resultRelInfo->ri_NumSlotsInitialized++;
- }
+ resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots] =
+ MakeSingleTupleTableSlot(plan_tdesc, planSlot->tts_ops);
- ExecCopySlot(resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots],
- slot);
+ /* remember how many batch slots we initialized */
+ resultRelInfo->ri_NumSlotsInitialized++;
+ }
- ExecCopySlot(resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots],
- planSlot);
+ ExecCopySlot(resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots],
+ slot);
- /*
- * If these are the first tuples stored in the buffers, add the
- * target rel and the mtstate to the
- * es_insert_pending_result_relations and
- * es_insert_pending_modifytables lists respectively, except in
- * the case where flushing was done above, in which case they
- * would already have been added to the lists, so no need to do
- * this.
- */
- if (resultRelInfo->ri_NumSlots == 0 && !flushed)
- {
- Assert(!list_member_ptr(estate->es_insert_pending_result_relations,
- resultRelInfo));
- estate->es_insert_pending_result_relations =
- lappend(estate->es_insert_pending_result_relations,
- resultRelInfo);
- estate->es_insert_pending_modifytables =
- lappend(estate->es_insert_pending_modifytables, mtstate);
- }
- Assert(list_member_ptr(estate->es_insert_pending_result_relations,
- resultRelInfo));
+ ExecCopySlot(resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots],
+ planSlot);
- resultRelInfo->ri_NumSlots++;
- /* Batch size can grow up to tuple length even if it exceeds work_mem. */
- resultRelInfo->ri_BatchMemoryUsed += tuple_len;
+ /*
+ * If these are the first tuples stored in the buffers, add
+ * the target rel and the mtstate to the
+ * es_insert_pending_result_relations and
+ * es_insert_pending_modifytables lists respectively, except
+ * in the case where flushing was done above, in which case
+ * they would already have been added to the lists, so no need
+ * to do this.
+ */
+ if (resultRelInfo->ri_NumSlots == 0 && !resultRelInfo->ri_ExecutorPendingStateModified)
+ {
+ Assert(!list_member_ptr(estate->es_insert_pending_result_relations,
+ resultRelInfo));
+ estate->es_insert_pending_result_relations =
+ lappend(estate->es_insert_pending_result_relations,
+ resultRelInfo);
+ estate->es_insert_pending_modifytables =
+ lappend(estate->es_insert_pending_modifytables, mtstate);
+ resultRelInfo->ri_ExecutorPendingStateModified = true;
+ }
+ Assert(list_member_ptr(estate->es_insert_pending_result_relations,
+ resultRelInfo));
- MemoryContextSwitchTo(oldContext);
+ resultRelInfo->ri_NumSlots++;
+ resultRelInfo->ri_BatchMemoryUsed += tuple_len;
+ /* We've flushed batch if it is too big. */
+ Assert(resultRelInfo->ri_BatchMemoryUsed < work_mem * 1024);
- return NULL;
+ MemoryContextSwitchTo(oldContext);
+
+ return NULL;
+ }
+ /* else do usual foreign insert */
}
/*
@@ -1445,11 +1450,18 @@ ExecPendingInserts(EState *estate)
ModifyTableState *mtstate = (ModifyTableState *) lfirst(l2);
Assert(mtstate);
- ExecBatchInsert(mtstate, resultRelInfo,
- resultRelInfo->ri_Slots,
- resultRelInfo->ri_PlanSlots,
- resultRelInfo->ri_NumSlots,
- estate, mtstate->canSetTag);
+
+ /*
+ * Batch insert could switch to non-batched insert, in this case we
+ * could have no filled slots.
+ */
+ if (resultRelInfo->ri_NumSlots > 0)
+ ExecBatchInsert(mtstate, resultRelInfo,
+ resultRelInfo->ri_Slots,
+ resultRelInfo->ri_PlanSlots,
+ resultRelInfo->ri_NumSlots,
+ estate, mtstate->canSetTag);
+ resultRelInfo->ri_ExecutorPendingStateModified = false;
}
list_free(estate->es_insert_pending_result_relations);
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index b82ec938228..dcc23774a4c 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -543,6 +543,7 @@ typedef struct ResultRelInfo
int ri_NumSlotsInitialized; /* number of initialized slots */
int ri_BatchSize; /* max slots inserted in a single batch */
int ri_BatchMemoryUsed; /* memory used by batch */
+ bool ri_ExecutorPendingStateModified; /* true if member of estate->es_insert_pending_result_relations */
TupleTableSlot **ri_Slots; /* input tuples for batch insert */
TupleTableSlot **ri_PlanSlots;
--
2.43.0
[text/x-diff] v3-0003-WIP-Use-tuplestore-in-PgFdwScanState-scan-state-to-l.patch (6.1K, ../../[email protected]/5-v3-0003-WIP-Use-tuplestore-in-PgFdwScanState-scan-state-to-l.patch)
download | inline diff:
From 1d89167cbb6b13bd3e73348fd6d8e8cd33c180ea Mon Sep 17 00:00:00 2001
From: Alexander Pyhalov <[email protected]>
Date: Thu, 18 Dec 2025 09:17:42 +0300
Subject: [PATCH 3/3] WIP: Use tuplestore in PgFdwScanState scan state to limit
memory usage
A tuplestore doesn't preserve ctids, so we need to store them
separately.
---
contrib/postgres_fdw/postgres_fdw.c | 82 ++++++++++++++++++++++++++---
1 file changed, 74 insertions(+), 8 deletions(-)
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 5e178c21b39..f160f1a3f62 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -158,8 +158,12 @@ typedef struct PgFdwScanState
const char **param_values; /* textual values of query parameters */
/* for storing result tuples */
- HeapTuple *tuples; /* array of currently-retrieved tuples */
- int num_tuples; /* # of tuples in array */
+ Tuplestorestate *result_store; /* currently-retrieved tuples */
+ ItemPointerData *ctids; /* separate store for ctids */
+
+ TupleTableSlot *tupstore_slot; /* slot needed for retrieving tuples from
+ * tuplestore */
+ int num_tuples; /* # of tuples in tuple store */
int next_tuple; /* index of next one to return */
/* batch-level state, for optimizing rewinds and avoiding useless fetch */
@@ -546,6 +550,8 @@ static void merge_fdw_options(PgFdwRelationInfo *fpinfo,
const PgFdwRelationInfo *fpinfo_i);
static int get_batch_size_option(Relation rel);
+static void clean_scan_state_result_store(PgFdwScanState *fsstate);
+
/*
* Foreign-data wrapper handler function: return a struct with pointers
@@ -1605,6 +1611,8 @@ postgresIterateForeignScan(ForeignScanState *node)
{
PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
TupleTableSlot *slot = node->ss.ss_ScanTupleSlot;
+ HeapTuple newtup;
+ MemoryContext oldcontext;
/*
* In sync mode, if this is the first call after Begin or ReScan, we need
@@ -1631,13 +1639,37 @@ postgresIterateForeignScan(ForeignScanState *node)
return ExecClearTuple(slot);
}
+ MemoryContextReset(fsstate->temp_cxt);
+ oldcontext = MemoryContextSwitchTo(fsstate->temp_cxt);
+ if (!tuplestore_gettupleslot(fsstate->result_store, true, false,
+ fsstate->tupstore_slot))
+ {
+ /*
+ * We've checked that next_tuple is less than fsstate->num_tuples, so
+ * there should be some result
+ */
+ MemoryContextSwitchTo(oldcontext);
+ ereport(ERROR,
+ (errcode(ERRCODE_INTERNAL_ERROR),
+ errmsg("unexpected end of store")));
+
+ }
+
+ newtup = ExecFetchSlotHeapTuple(fsstate->tupstore_slot, true, NULL);
+ /* Tuple store doesn't preserve ctid, so restore it separately */
+ newtup->t_self = newtup->t_data->t_ctid = fsstate->ctids[fsstate->next_tuple];
+ ExecClearTuple(fsstate->tupstore_slot);
+
/*
* Return the next tuple.
*/
- ExecStoreHeapTuple(fsstate->tuples[fsstate->next_tuple++],
+ ExecStoreHeapTuple(newtup,
slot,
false);
+ fsstate->next_tuple++;
+ MemoryContextSwitchTo(oldcontext);
+
return slot;
}
@@ -1699,6 +1731,8 @@ postgresReScanForeignScan(ForeignScanState *node)
{
/* Easy: just rescan what we already have in memory, if anything */
fsstate->next_tuple = 0;
+ if (fsstate->result_store)
+ tuplestore_rescan(fsstate->result_store);
return;
}
@@ -1708,7 +1742,7 @@ postgresReScanForeignScan(ForeignScanState *node)
PQclear(res);
/* Now force a fresh FETCH. */
- fsstate->tuples = NULL;
+ clean_scan_state_result_store(fsstate);
fsstate->num_tuples = 0;
fsstate->next_tuple = 0;
fsstate->fetch_ct_2 = 0;
@@ -1737,6 +1771,8 @@ postgresEndForeignScan(ForeignScanState *node)
ReleaseConnection(fsstate->conn);
fsstate->conn = NULL;
+ clean_scan_state_result_store(fsstate);
+
/* MemoryContexts will be deleted automatically. */
}
@@ -3781,7 +3817,8 @@ create_cursor(ForeignScanState *node)
/* Mark the cursor as created, and show no tuples have been retrieved */
fsstate->cursor_exists = true;
- fsstate->tuples = NULL;
+ clean_scan_state_result_store(fsstate);
+
fsstate->num_tuples = 0;
fsstate->next_tuple = 0;
fsstate->fetch_ct_2 = 0;
@@ -3808,7 +3845,9 @@ fetch_more_data(ForeignScanState *node)
* We'll store the tuples in the batch_cxt. First, flush the previous
* batch.
*/
- fsstate->tuples = NULL;
+ clean_scan_state_result_store(fsstate);
+
+ MemoryContextReset(fsstate->temp_cxt);
MemoryContextReset(fsstate->batch_cxt);
oldcontext = MemoryContextSwitchTo(fsstate->batch_cxt);
@@ -3844,21 +3883,29 @@ fetch_more_data(ForeignScanState *node)
/* Convert the data into HeapTuples */
numrows = PQntuples(res);
- fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple));
+ fsstate->result_store = tuplestore_begin_heap(false, false, work_mem);
+ fsstate->tupstore_slot = MakeSingleTupleTableSlot(fsstate->tupdesc, &TTSOpsMinimalTuple);
fsstate->num_tuples = numrows;
fsstate->next_tuple = 0;
+ fsstate->ctids = palloc0(numrows * sizeof(ItemPointerData));
for (i = 0; i < numrows; i++)
{
+ HeapTuple newtup;
+
Assert(IsA(node->ss.ps.plan, ForeignScan));
- fsstate->tuples[i] =
+ newtup =
make_tuple_from_result_row(res, i,
fsstate->rel,
fsstate->attinmeta,
fsstate->retrieved_attrs,
node,
fsstate->temp_cxt);
+
+ tuplestore_puttuple(fsstate->result_store, newtup);
+ ItemPointerCopy(&newtup->t_self, &fsstate->ctids[i]);
+ heap_freetuple(newtup);
}
/* Update fetch_ct_2 */
@@ -7636,6 +7683,25 @@ make_tuple_from_result_row(PGresult *res,
return tuple;
}
+/*
+ * Clean PgFdwScanState result store
+ */
+static void
+clean_scan_state_result_store(PgFdwScanState *fsstate)
+{
+ if (fsstate->result_store)
+ {
+ /*
+ * We partially rely on the fact that batch_cxt is also reset
+ */
+ tuplestore_end(fsstate->result_store);
+ ExecDropSingleTupleTableSlot(fsstate->tupstore_slot);
+ fsstate->result_store = NULL;
+ fsstate->tupstore_slot = NULL;
+ fsstate->ctids = NULL;
+ }
+}
+
/*
* Callback function which is called when error occurs during column value
* conversion. Print names of column and relation.
--
2.43.0
^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Limit memory usage by postgres_fdw batches
@ 2026-01-23 06:18 Alexander Pyhalov <[email protected]>
parent: Alexander Pyhalov <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Alexander Pyhalov @ 2026-01-23 06:18 UTC (permalink / raw)
To: Tomas Vondra <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>
Alexander Pyhalov писал(а) 2026-01-13 13:44:
> For now I start thinking we need some form of FETCH, which stops
> fetching data based on batch size...
Hi.
To limit memory consumption, we actually have to retreive less data. And
we can do it only on the side of the foreign server. I've rewritten the
third patch. We introduce a new parameter - cursor_fetch_limit, which is
set by postgres_fdw. When it is set, fetching limited count of records
from the cursor is also limited by memory consumed by the records. Of
course, record size is some estimation (for example, we don't know what
out function will do).
This works as expected - in my tests with tables of large records,
backends, executing selects, were always restricted by about 2 GB of RAM
overall (without patch memory consumption easily grows up to 8 GB).
However, now when we got less tuples from executor, than expected, we
should recheck, if these are all tuples we can get. I've introduced
es_eof EState field to signal that there's no more tuples. Don't know if
it's the best way.
--
Best regards,
Alexander Pyhalov,
Postgres Professional
Attachments:
[text/x-diff] v4-0001-Limit-batch_size-for-foreign-insert-with-work_mem.patch (2.7K, ../../[email protected]/2-v4-0001-Limit-batch_size-for-foreign-insert-with-work_mem.patch)
download | inline diff:
From 9eca4c6277c63f4f8f28afced07ee13c65bfb36e Mon Sep 17 00:00:00 2001
From: Alexander Pyhalov <[email protected]>
Date: Tue, 16 Dec 2025 18:46:27 +0300
Subject: [PATCH 1/3] Limit batch_size for foreign insert with work_mem
Option batch_size can be set on foreign server level. It can be very optimistic
even for one table with different tuple lengths. To prevent large memory usage
limit effective batch size with work_mem.
---
src/backend/executor/nodeModifyTable.c | 15 ++++++++++++---
src/include/nodes/execnodes.h | 1 +
2 files changed, 13 insertions(+), 3 deletions(-)
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 7d7411a7056..9495d9b04bb 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -943,12 +943,18 @@ ExecInsert(ModifyTableContext *context,
if (resultRelInfo->ri_BatchSize > 1)
{
bool flushed = false;
+ Size tuple_len;
+
+ /* Compute length of the current tuple */
+ slot_getallattrs(slot);
+ tuple_len = heap_compute_data_size(slot->tts_tupleDescriptor, slot->tts_values, slot->tts_isnull);
/*
- * When we've reached the desired batch size, perform the
- * insertion.
+ * When we've reached the desired batch size or exceeded work_mem,
+ * perform the insertion.
*/
- if (resultRelInfo->ri_NumSlots == resultRelInfo->ri_BatchSize)
+ if (resultRelInfo->ri_NumSlots == resultRelInfo->ri_BatchSize ||
+ ((tuple_len + resultRelInfo->ri_BatchMemoryUsed > work_mem * 1024) && (resultRelInfo->ri_NumSlots > 0)))
{
ExecBatchInsert(mtstate, resultRelInfo,
resultRelInfo->ri_Slots,
@@ -1019,6 +1025,8 @@ ExecInsert(ModifyTableContext *context,
resultRelInfo));
resultRelInfo->ri_NumSlots++;
+ /* Batch size can grow up to tuple length even if it exceeds work_mem. */
+ resultRelInfo->ri_BatchMemoryUsed += tuple_len;
MemoryContextSwitchTo(oldContext);
@@ -1418,6 +1426,7 @@ ExecBatchInsert(ModifyTableState *mtstate,
ExecClearTuple(planSlots[i]);
}
resultRelInfo->ri_NumSlots = 0;
+ resultRelInfo->ri_BatchMemoryUsed = 0;
}
/*
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index f8053d9e572..33d3eb06891 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -544,6 +544,7 @@ typedef struct ResultRelInfo
int ri_NumSlots; /* number of slots in the array */
int ri_NumSlotsInitialized; /* number of initialized slots */
int ri_BatchSize; /* max slots inserted in a single batch */
+ int ri_BatchMemoryUsed; /* memory used by batch */
TupleTableSlot **ri_Slots; /* input tuples for batch insert */
TupleTableSlot **ri_PlanSlots;
--
2.43.0
[text/x-diff] v4-0002-Fall-back-to-foreign-insert-when-tuple-is-too-big.patch (10.2K, ../../[email protected]/3-v4-0002-Fall-back-to-foreign-insert-when-tuple-is-too-big.patch)
download | inline diff:
From b4afad03431f376c56238266f401e061c4d31db5 Mon Sep 17 00:00:00 2001
From: Alexander Pyhalov <[email protected]>
Date: Tue, 30 Dec 2025 10:35:22 +0300
Subject: [PATCH 2/3] Fall back to foreign insert when tuple is too big
When tuple is more then work_mem, we don't form batch,
but fall through to usual foreign insert. This means
that executor can form batch, insert it and finish with
a usual insert, so there will be no pending batch inserts
to process. To avoid manipulation with possibly long lists,
we don't clear es_insert_pending_result_relations and
es_insert_pending_modifytables lists, but check in
ExecPendingInserts() if there are any pending tuples.
---
.../postgres_fdw/expected/postgres_fdw.out | 11 ++
contrib/postgres_fdw/sql/postgres_fdw.sql | 9 ++
src/backend/executor/nodeModifyTable.c | 138 ++++++++++--------
src/include/nodes/execnodes.h | 1 +
4 files changed, 96 insertions(+), 63 deletions(-)
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 6066510c7c0..cb3044c769a 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -7640,6 +7640,17 @@ select count(*) from tab_batch_sharded;
45
(1 row)
+delete from tab_batch_sharded;
+-- test batch insert with large tuples and switching from batch to usual insert
+set work_mem to 64;
+insert into tab_batch_sharded select 3, case when i%4 = 0 then lpad('a',65*1024,'a') else 'test'|| i end from generate_series(1, 100) i;
+select count(*) from tab_batch_sharded;
+ count
+-------
+ 100
+(1 row)
+
+reset work_mem;
drop table tab_batch_local;
drop table tab_batch_sharded;
drop table tab_batch_sharded_p1_remote;
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 4f7ab2ed0ac..a084afc8de7 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -1944,10 +1944,19 @@ create foreign table tab_batch_sharded_p1 partition of tab_batch_sharded
server loopback options (table_name 'tab_batch_sharded_p1_remote');
insert into tab_batch_sharded select * from tab_batch_local;
select count(*) from tab_batch_sharded;
+delete from tab_batch_sharded;
+-- test batch insert with large tuples and switching from batch to usual insert
+set work_mem to 64;
+insert into tab_batch_sharded select 3, case when i%4 = 0 then lpad('a',65*1024,'a') else 'test'|| i end from generate_series(1, 100) i;
+select count(*) from tab_batch_sharded;
+reset work_mem;
+
drop table tab_batch_local;
drop table tab_batch_sharded;
drop table tab_batch_sharded_p1_remote;
+
+
alter server loopback options (drop batch_size);
-- ===================================================================
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 9495d9b04bb..ba3d135d6a6 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -942,7 +942,6 @@ ExecInsert(ModifyTableContext *context,
*/
if (resultRelInfo->ri_BatchSize > 1)
{
- bool flushed = false;
Size tuple_len;
/* Compute length of the current tuple */
@@ -961,76 +960,82 @@ ExecInsert(ModifyTableContext *context,
resultRelInfo->ri_PlanSlots,
resultRelInfo->ri_NumSlots,
estate, canSetTag);
- flushed = true;
}
- oldContext = MemoryContextSwitchTo(estate->es_query_cxt);
-
- if (resultRelInfo->ri_Slots == NULL)
+ if (tuple_len < work_mem * 1024)
{
- resultRelInfo->ri_Slots = palloc_array(TupleTableSlot *, resultRelInfo->ri_BatchSize);
- resultRelInfo->ri_PlanSlots = palloc_array(TupleTableSlot *, resultRelInfo->ri_BatchSize);
- }
+ oldContext = MemoryContextSwitchTo(estate->es_query_cxt);
- /*
- * Initialize the batch slots. We don't know how many slots will
- * be needed, so we initialize them as the batch grows, and we
- * keep them across batches. To mitigate an inefficiency in how
- * resource owner handles objects with many references (as with
- * many slots all referencing the same tuple descriptor) we copy
- * the appropriate tuple descriptor for each slot.
- */
- if (resultRelInfo->ri_NumSlots >= resultRelInfo->ri_NumSlotsInitialized)
- {
- TupleDesc tdesc = CreateTupleDescCopy(slot->tts_tupleDescriptor);
- TupleDesc plan_tdesc =
- CreateTupleDescCopy(planSlot->tts_tupleDescriptor);
+ if (resultRelInfo->ri_Slots == NULL)
+ {
+ resultRelInfo->ri_Slots = palloc_array(TupleTableSlot *, resultRelInfo->ri_BatchSize);
+ resultRelInfo->ri_PlanSlots = palloc_array(TupleTableSlot *, resultRelInfo->ri_BatchSize);
+ }
- resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots] =
- MakeSingleTupleTableSlot(tdesc, slot->tts_ops);
+ /*
+ * Initialize the batch slots. We don't know how many slots
+ * will be needed, so we initialize them as the batch grows,
+ * and we keep them across batches. To mitigate an
+ * inefficiency in how resource owner handles objects with
+ * many references (as with many slots all referencing the
+ * same tuple descriptor) we copy the appropriate tuple
+ * descriptor for each slot.
+ */
+ if (resultRelInfo->ri_NumSlots >= resultRelInfo->ri_NumSlotsInitialized)
+ {
+ TupleDesc tdesc = CreateTupleDescCopy(slot->tts_tupleDescriptor);
+ TupleDesc plan_tdesc =
+ CreateTupleDescCopy(planSlot->tts_tupleDescriptor);
- resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots] =
- MakeSingleTupleTableSlot(plan_tdesc, planSlot->tts_ops);
+ resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots] =
+ MakeSingleTupleTableSlot(tdesc, slot->tts_ops);
- /* remember how many batch slots we initialized */
- resultRelInfo->ri_NumSlotsInitialized++;
- }
+ resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots] =
+ MakeSingleTupleTableSlot(plan_tdesc, planSlot->tts_ops);
- ExecCopySlot(resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots],
- slot);
+ /* remember how many batch slots we initialized */
+ resultRelInfo->ri_NumSlotsInitialized++;
+ }
- ExecCopySlot(resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots],
- planSlot);
+ ExecCopySlot(resultRelInfo->ri_Slots[resultRelInfo->ri_NumSlots],
+ slot);
- /*
- * If these are the first tuples stored in the buffers, add the
- * target rel and the mtstate to the
- * es_insert_pending_result_relations and
- * es_insert_pending_modifytables lists respectively, except in
- * the case where flushing was done above, in which case they
- * would already have been added to the lists, so no need to do
- * this.
- */
- if (resultRelInfo->ri_NumSlots == 0 && !flushed)
- {
- Assert(!list_member_ptr(estate->es_insert_pending_result_relations,
- resultRelInfo));
- estate->es_insert_pending_result_relations =
- lappend(estate->es_insert_pending_result_relations,
- resultRelInfo);
- estate->es_insert_pending_modifytables =
- lappend(estate->es_insert_pending_modifytables, mtstate);
- }
- Assert(list_member_ptr(estate->es_insert_pending_result_relations,
- resultRelInfo));
+ ExecCopySlot(resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots],
+ planSlot);
- resultRelInfo->ri_NumSlots++;
- /* Batch size can grow up to tuple length even if it exceeds work_mem. */
- resultRelInfo->ri_BatchMemoryUsed += tuple_len;
+ /*
+ * If these are the first tuples stored in the buffers, add
+ * the target rel and the mtstate to the
+ * es_insert_pending_result_relations and
+ * es_insert_pending_modifytables lists respectively, except
+ * in the case where flushing was done above, in which case
+ * they would already have been added to the lists, so no need
+ * to do this.
+ */
+ if (resultRelInfo->ri_NumSlots == 0 && !resultRelInfo->ri_ExecutorPendingStateModified)
+ {
+ Assert(!list_member_ptr(estate->es_insert_pending_result_relations,
+ resultRelInfo));
+ estate->es_insert_pending_result_relations =
+ lappend(estate->es_insert_pending_result_relations,
+ resultRelInfo);
+ estate->es_insert_pending_modifytables =
+ lappend(estate->es_insert_pending_modifytables, mtstate);
+ resultRelInfo->ri_ExecutorPendingStateModified = true;
+ }
+ Assert(list_member_ptr(estate->es_insert_pending_result_relations,
+ resultRelInfo));
- MemoryContextSwitchTo(oldContext);
+ resultRelInfo->ri_NumSlots++;
+ resultRelInfo->ri_BatchMemoryUsed += tuple_len;
+ /* We've flushed batch if it is too big. */
+ Assert(resultRelInfo->ri_BatchMemoryUsed < work_mem * 1024);
- return NULL;
+ MemoryContextSwitchTo(oldContext);
+
+ return NULL;
+ }
+ /* else do usual foreign insert */
}
/*
@@ -1445,11 +1450,18 @@ ExecPendingInserts(EState *estate)
ModifyTableState *mtstate = (ModifyTableState *) lfirst(l2);
Assert(mtstate);
- ExecBatchInsert(mtstate, resultRelInfo,
- resultRelInfo->ri_Slots,
- resultRelInfo->ri_PlanSlots,
- resultRelInfo->ri_NumSlots,
- estate, mtstate->canSetTag);
+
+ /*
+ * Batch insert could switch to non-batched insert, in this case we
+ * could have no filled slots.
+ */
+ if (resultRelInfo->ri_NumSlots > 0)
+ ExecBatchInsert(mtstate, resultRelInfo,
+ resultRelInfo->ri_Slots,
+ resultRelInfo->ri_PlanSlots,
+ resultRelInfo->ri_NumSlots,
+ estate, mtstate->canSetTag);
+ resultRelInfo->ri_ExecutorPendingStateModified = false;
}
list_free(estate->es_insert_pending_result_relations);
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 33d3eb06891..3220f9b2574 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -545,6 +545,7 @@ typedef struct ResultRelInfo
int ri_NumSlotsInitialized; /* number of initialized slots */
int ri_BatchSize; /* max slots inserted in a single batch */
int ri_BatchMemoryUsed; /* memory used by batch */
+ bool ri_ExecutorPendingStateModified; /* true if member of estate->es_insert_pending_result_relations */
TupleTableSlot **ri_Slots; /* input tuples for batch insert */
TupleTableSlot **ri_PlanSlots;
--
2.43.0
[text/x-diff] v4-0003-Introduce-cursor_fetch_limit-parameter-and-use-it-in.patch (26.9K, ../../[email protected]/4-v4-0003-Introduce-cursor_fetch_limit-parameter-and-use-it-in.patch)
download | inline diff:
From 2e0710dd4db89fbb1f02d3e73165425dee76830b Mon Sep 17 00:00:00 2001
From: Alexander Pyhalov <[email protected]>
Date: Thu, 15 Jan 2026 09:20:51 +0300
Subject: [PATCH 3/3] Introduce cursor_fetch_limit parameter and use it in
postgres_fdw
This parameter is specified in bytes and limits the number of records,
which is retrieved by FETCH statement, based on their size. FETCH ALL
is not limited.
When cursor_fetch_limit >= 0 and records count is passed to
ExecutorRun(), we can't longer rely on fact that when executor
returned less records than expected, there's no more records to fetch,
so we introduce additional EState field which after ExecutorRun()
call says if there are more records.
---
contrib/postgres_fdw/connection.c | 13 +++
contrib/postgres_fdw/postgres_fdw.c | 7 +-
src/backend/access/common/heaptuple.c | 54 +++++++++
src/backend/executor/execMain.c | 37 +++++-
src/backend/executor/execUtils.c | 1 +
src/backend/tcop/pquery.c | 54 +++++++--
src/backend/utils/misc/guc_parameters.dat | 9 ++
src/backend/utils/misc/guc_tables.c | 1 +
src/backend/utils/misc/postgresql.conf.sample | 1 +
src/include/access/htup_details.h | 3 +
src/include/executor/executor.h | 2 +
src/include/nodes/execnodes.h | 1 +
src/test/regress/expected/portals.out | 105 ++++++++++++++++++
src/test/regress/sql/portals.sql | 28 +++++
14 files changed, 304 insertions(+), 12 deletions(-)
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index 487a1a23170..d6c6c26b393 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -802,6 +802,19 @@ configure_remote_session(PGconn *conn)
do_sql_command(conn, "SET extra_float_digits = 3");
else
do_sql_command(conn, "SET extra_float_digits = 2");
+ /*
+ * Limit fetch size with work_mem to avoid uncontrolled memory
+ * consumption.
+ */
+ if (remoteversion >= 190000)
+ {
+ char sql[64];
+ int fetch_limit;
+
+ fetch_limit = (work_mem * 1024 > MaxAllocSize) ? MaxAllocSize : work_mem * 1024;
+ snprintf(sql, sizeof(sql), "SET cursor_fetch_limit = %d", fetch_limit);
+ do_sql_command(conn, sql);
+ }
}
/*
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 3572689e33b..abfbd0e60ac 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -3865,8 +3865,11 @@ fetch_more_data(ForeignScanState *node)
if (fsstate->fetch_ct_2 < 2)
fsstate->fetch_ct_2++;
- /* Must be EOF if we didn't get as many tuples as we asked for. */
- fsstate->eof_reached = (numrows < fsstate->fetch_size);
+ /*
+ * Must be EOF if we got 0 rows. We can get less rows than we asked for
+ * even if there's more data if batch of "fetch_size" exceeds work_mem.
+ */
+ fsstate->eof_reached = (numrows == 0);
PQclear(res);
diff --git a/src/backend/access/common/heaptuple.c b/src/backend/access/common/heaptuple.c
index 11bec20e82e..7308d2ff4f0 100644
--- a/src/backend/access/common/heaptuple.c
+++ b/src/backend/access/common/heaptuple.c
@@ -57,6 +57,7 @@
#include "postgres.h"
+#include "access/detoast.h"
#include "access/heaptoast.h"
#include "access/sysattr.h"
#include "access/tupdesc_details.h"
@@ -266,6 +267,59 @@ heap_compute_data_size(TupleDesc tupleDesc,
return data_length;
}
+/*
+ * Estimate tuple size when it's transferred to the wire. Unlike
+ * heap_compute_data_size() it assumes that all toasts will be
+ * detoasted (but doesn't make assumptions about further
+ * processing).
+ */
+Size
+estimate_tuple_size(TupleDesc tupleDesc,
+ const Datum *values,
+ const bool *isnull)
+{
+ Size data_length = 0;
+ int i;
+ int numberOfAttributes = tupleDesc->natts;
+
+ for (i = 0; i < numberOfAttributes; i++)
+ {
+ Datum val;
+ CompactAttribute *atti;
+ bool is_varlena;
+ bool is_varwidth;
+
+ if (isnull[i])
+ continue;
+
+ val = values[i];
+ atti = TupleDescCompactAttr(tupleDesc, i);
+
+ is_varlena = (!atti->attbyval && atti->attlen == -1);
+ is_varwidth = (!atti->attbyval && atti->attlen == -2);
+
+ if (is_varlena)
+ {
+ data_length = att_nominal_alignby(data_length, atti->attalignby);
+ data_length += toast_raw_datum_size(val);
+ }
+ else if (is_varwidth)
+ {
+ data_length = att_nominal_alignby(data_length, atti->attalignby);
+ data_length += strlen(DatumGetCString(val)) + 1;
+ }
+ else
+ {
+ data_length = att_datum_alignby(data_length, atti->attalignby,
+ atti->attlen, val);
+ data_length = att_addlength_datum(data_length, atti->attlen,
+ val);
+ }
+ }
+
+ return data_length;
+}
+
/*
* Per-attribute helper for heap_fill_tuple and other routines building tuples.
*
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index bfd3ebc601e..7b0fef57a66 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -64,6 +64,9 @@
#include "utils/snapmgr.h"
+/* max batch for fetching from cursor */
+int cursor_fetch_limit = -1;
+
/* Hooks for plugins to get control in ExecutorStart/Run/Finish/End */
ExecutorStart_hook_type ExecutorStart_hook = NULL;
ExecutorRun_hook_type ExecutorRun_hook = NULL;
@@ -343,6 +346,7 @@ standard_ExecutorRun(QueryDesc *queryDesc,
* startup tuple receiver, if we will be emitting tuples
*/
estate->es_processed = 0;
+ estate->es_eof = false;
sendTuples = (operation == CMD_SELECT ||
queryDesc->plannedstmt->hasReturning);
@@ -1669,11 +1673,13 @@ ExecutePlan(QueryDesc *queryDesc,
bool use_parallel_mode;
TupleTableSlot *slot;
uint64 current_tuple_count;
+ Size current_batch_size;
/*
* initialize local variables
*/
current_tuple_count = 0;
+ current_batch_size = 0;
/*
* Set the direction.
@@ -1715,7 +1721,10 @@ ExecutePlan(QueryDesc *queryDesc,
* process so we just end the loop...
*/
if (TupIsNull(slot))
+ {
+ estate->es_eof = true;
break;
+ }
/*
* If we have a junk filter, then project a new tuple with the junk
@@ -1740,7 +1749,10 @@ ExecutePlan(QueryDesc *queryDesc,
* end the loop.
*/
if (!dest->receiveSlot(slot, dest))
+ {
+ estate->es_eof = true;
break;
+ }
}
/*
@@ -1757,8 +1769,29 @@ ExecutePlan(QueryDesc *queryDesc,
* means no limit.
*/
current_tuple_count++;
- if (numberTuples && numberTuples == current_tuple_count)
- break;
+ if (numberTuples)
+ {
+ if (numberTuples == current_tuple_count)
+ break;
+
+ /*
+ * If batch limit in bytes is specified, count current batch size
+ * and exit if limit is reached
+ */
+ if (cursor_fetch_limit >= 0)
+ {
+ Size tuple_len;
+
+ /* Compute length of the current tuple */
+ slot_getallattrs(slot);
+ tuple_len = estimate_tuple_size(slot->tts_tupleDescriptor, slot->tts_values, slot->tts_isnull);
+ current_batch_size += tuple_len;
+
+ if (current_batch_size > cursor_fetch_limit)
+ break;
+ }
+ }
+
}
/*
diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c
index a7955e476f9..7f820388878 100644
--- a/src/backend/executor/execUtils.c
+++ b/src/backend/executor/execUtils.c
@@ -143,6 +143,7 @@ CreateExecutorState(void)
estate->es_processed = 0;
estate->es_total_processed = 0;
+ estate->es_eof = false;
estate->es_top_eflags = 0;
estate->es_instrument = 0;
diff --git a/src/backend/tcop/pquery.c b/src/backend/tcop/pquery.c
index d8fc75d0bb9..78908ae2a8b 100644
--- a/src/backend/tcop/pquery.c
+++ b/src/backend/tcop/pquery.c
@@ -17,6 +17,7 @@
#include <limits.h>
+#include "access/htup_details.h"
#include "access/xact.h"
#include "commands/prepare.h"
#include "executor/executor.h"
@@ -44,7 +45,7 @@ static void ProcessQuery(PlannedStmt *plan,
QueryCompletion *qc);
static void FillPortalStore(Portal portal, bool isTopLevel);
static uint64 RunFromStore(Portal portal, ScanDirection direction, uint64 count,
- DestReceiver *dest);
+ DestReceiver *dest, bool *eof);
static uint64 PortalRunSelect(Portal portal, bool forward, long count,
DestReceiver *dest);
static void PortalRunUtility(Portal portal, PlannedStmt *pstmt,
@@ -864,6 +865,7 @@ PortalRunSelect(Portal portal,
QueryDesc *queryDesc;
ScanDirection direction;
uint64 nprocessed;
+ bool eof;
/*
* NB: queryDesc will be NULL if we are fetching from a held cursor or a
@@ -909,20 +911,23 @@ PortalRunSelect(Portal portal,
count = 0;
if (portal->holdStore)
- nprocessed = RunFromStore(portal, direction, (uint64) count, dest);
+ {
+ nprocessed = RunFromStore(portal, direction, (uint64) count, dest, &eof);
+ }
else
{
PushActiveSnapshot(queryDesc->snapshot);
ExecutorRun(queryDesc, direction, (uint64) count);
nprocessed = queryDesc->estate->es_processed;
PopActiveSnapshot();
+ eof = queryDesc->estate->es_eof;
}
if (!ScanDirectionIsNoMovement(direction))
{
if (nprocessed > 0)
portal->atStart = false; /* OK to go backward now */
- if (count == 0 || nprocessed < (uint64) count)
+ if (count == 0 || eof)
portal->atEnd = true; /* we retrieved 'em all */
portal->portalPos += nprocessed;
}
@@ -948,13 +953,16 @@ PortalRunSelect(Portal portal,
count = 0;
if (portal->holdStore)
- nprocessed = RunFromStore(portal, direction, (uint64) count, dest);
+ {
+ nprocessed = RunFromStore(portal, direction, (uint64) count, dest, &eof);
+ }
else
{
PushActiveSnapshot(queryDesc->snapshot);
ExecutorRun(queryDesc, direction, (uint64) count);
nprocessed = queryDesc->estate->es_processed;
PopActiveSnapshot();
+ eof = queryDesc->estate->es_eof;
}
if (!ScanDirectionIsNoMovement(direction))
@@ -964,7 +972,7 @@ PortalRunSelect(Portal portal,
portal->atEnd = false; /* OK to go forward now */
portal->portalPos++; /* adjust for endpoint case */
}
- if (count == 0 || nprocessed < (uint64) count)
+ if (count == 0 || eof)
{
portal->atStart = true; /* we retrieved 'em all */
portal->portalPos = 0;
@@ -1049,15 +1057,18 @@ FillPortalStore(Portal portal, bool isTopLevel)
*/
static uint64
RunFromStore(Portal portal, ScanDirection direction, uint64 count,
- DestReceiver *dest)
+ DestReceiver *dest, bool *eof)
{
uint64 current_tuple_count = 0;
+ Size current_batch_size = 0;
TupleTableSlot *slot;
slot = MakeSingleTupleTableSlot(portal->tupDesc, &TTSOpsMinimalTuple);
dest->rStartup(dest, CMD_SELECT, portal->tupDesc);
+ *eof = false;
+
if (ScanDirectionIsNoMovement(direction))
{
/* do nothing except start/stop the destination */
@@ -1079,7 +1090,10 @@ RunFromStore(Portal portal, ScanDirection direction, uint64 count,
MemoryContextSwitchTo(oldcontext);
if (!ok)
+ {
+ *eof = true;
break;
+ }
/*
* If we are not able to send the tuple, we assume the destination
@@ -1087,7 +1101,24 @@ RunFromStore(Portal portal, ScanDirection direction, uint64 count,
* end the loop.
*/
if (!dest->receiveSlot(slot, dest))
+ {
+ *eof = true;
break;
+ }
+
+ /*
+ * If batch limit in bytes is specified, count current batch
+ * size.
+ */
+ if (cursor_fetch_limit >= 0 && count)
+ {
+ Size tuple_len;
+
+ /* Compute length of the current tuple. */
+ slot_getallattrs(slot);
+ tuple_len = estimate_tuple_size(slot->tts_tupleDescriptor, slot->tts_values, slot->tts_isnull);
+ current_batch_size += tuple_len;
+ }
ExecClearTuple(slot);
@@ -1097,8 +1128,15 @@ RunFromStore(Portal portal, ScanDirection direction, uint64 count,
* means no limit.
*/
current_tuple_count++;
- if (count && count == current_tuple_count)
- break;
+ if (count)
+ {
+ if (count == current_tuple_count)
+ break;
+
+ /* Exit if batch limit is reached */
+ if (current_batch_size > cursor_fetch_limit)
+ break;
+ }
}
}
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 7c60b125564..e7bc17a8f07 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -522,6 +522,15 @@
assign_hook => 'assign_createrole_self_grant',
},
+{ name => 'cursor_fetch_limit', type => 'int', context => 'PGC_USERSET', group => 'RESOURCES_MEM',
+ short_desc => 'Sets the maximum size in bytes for fetching from cursor',
+ flags => 'GUC_UNIT_BYTE',
+ variable => 'cursor_fetch_limit',
+ boot_val => '-1',
+ min => '-1',
+ max => 'MaxAllocSize',
+},
+
{ name => 'cursor_tuple_fraction', type => 'real', context => 'PGC_USERSET', group => 'QUERY_TUNING_OTHER',
short_desc => 'Sets the planner\'s estimate of the fraction of a cursor\'s rows that will be retrieved.',
flags => 'GUC_EXPLAIN',
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 73ff6ad0a32..bf40d407494 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -50,6 +50,7 @@
#include "commands/vacuum.h"
#include "common/file_utils.h"
#include "common/scram-common.h"
+#include "executor/executor.h"
#include "jit/jit.h"
#include "libpq/auth.h"
#include "libpq/libpq.h"
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index dc9e2255f8a..169e96e20fe 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -468,6 +468,7 @@
#default_statistics_target = 100 # range 1-10000
#constraint_exclusion = partition # on, off, or partition
#cursor_tuple_fraction = 0.1 # range 0.0-1.0
+#cursor_fetch_limit = -1 # maximum size in bytes for fetching from cursor
#from_collapse_limit = 8
#jit = on # allow JIT compilation
#join_collapse_limit = 8 # 1 disables collapsing of explicit
diff --git a/src/include/access/htup_details.h b/src/include/access/htup_details.h
index d406825ff22..cb24e159509 100644
--- a/src/include/access/htup_details.h
+++ b/src/include/access/htup_details.h
@@ -808,6 +808,9 @@ HeapTupleClearHeapOnly(const HeapTupleData *tuple)
/* prototypes for functions in common/heaptuple.c */
extern Size heap_compute_data_size(TupleDesc tupleDesc,
const Datum *values, const bool *isnull);
+extern Size estimate_tuple_size(TupleDesc tupleDesc,
+ const Datum *values, const bool *isnull);
+
extern void heap_fill_tuple(TupleDesc tupleDesc,
const Datum *values, const bool *isnull,
char *data, Size data_size,
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 5929aabc353..80fcb8f6f10 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -96,6 +96,8 @@ typedef bool (*ExecutorCheckPerms_hook_type) (List *rangeTable,
bool ereport_on_violation);
extern PGDLLIMPORT ExecutorCheckPerms_hook_type ExecutorCheckPerms_hook;
+/* GUC variables */
+extern PGDLLIMPORT int cursor_fetch_limit;
/*
* prototypes from functions in execAmi.c
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 3220f9b2574..ecd0edd390c 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -719,6 +719,7 @@ typedef struct EState
* ExecutorRun() call. */
uint64 es_total_processed; /* total # of tuples aggregated across all
* ExecutorRun() calls. */
+ bool es_eof; /* true if no more data to process */
int es_top_eflags; /* eflags passed to ExecutorStart */
int es_instrument; /* OR of InstrumentOption flags */
diff --git a/src/test/regress/expected/portals.out b/src/test/regress/expected/portals.out
index 31f77abc446..2a1d773b374 100644
--- a/src/test/regress/expected/portals.out
+++ b/src/test/regress/expected/portals.out
@@ -1561,3 +1561,108 @@ fetch all in held_portal;
(1 row)
reset default_toast_compression;
+-- Check memory limit for fetching from cursor
+set cursor_fetch_limit = '512B';
+BEGIN;
+DECLARE foo1 SCROLL CURSOR FOR SELECT * FROM tenk1 ORDER BY unique2;
+DECLARE foo2 SCROLL CURSOR FOR SELECT * FROM tenk2;
+DECLARE foo3 SCROLL CURSOR FOR SELECT * FROM tenk1 ORDER BY unique2;
+DECLARE foo4 SCROLL CURSOR FOR SELECT * FROM tenk2;
+FETCH 1 in foo1;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 8800 | 0 | 0 | 0 | 0 | 0 | 0 | 800 | 800 | 3800 | 8800 | 0 | 1 | MAAAAA | AAAAAA | AAAAxx
+(1 row)
+
+FETCH 2 in foo2;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 8800 | 0 | 0 | 0 | 0 | 0 | 0 | 800 | 800 | 3800 | 8800 | 0 | 1 | MAAAAA | AAAAAA | AAAAxx
+ 1891 | 1 | 1 | 3 | 1 | 11 | 91 | 891 | 1891 | 1891 | 1891 | 182 | 183 | TUAAAA | BAAAAA | HHHHxx
+(2 rows)
+
+FETCH 3 in foo3;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 8800 | 0 | 0 | 0 | 0 | 0 | 0 | 800 | 800 | 3800 | 8800 | 0 | 1 | MAAAAA | AAAAAA | AAAAxx
+ 1891 | 1 | 1 | 3 | 1 | 11 | 91 | 891 | 1891 | 1891 | 1891 | 182 | 183 | TUAAAA | BAAAAA | HHHHxx
+ 3420 | 2 | 0 | 0 | 0 | 0 | 20 | 420 | 1420 | 3420 | 3420 | 40 | 41 | OBAAAA | CAAAAA | OOOOxx
+(3 rows)
+
+-- Well, we've reached the limit
+FETCH 4 in foo4;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 8800 | 0 | 0 | 0 | 0 | 0 | 0 | 800 | 800 | 3800 | 8800 | 0 | 1 | MAAAAA | AAAAAA | AAAAxx
+ 1891 | 1 | 1 | 3 | 1 | 11 | 91 | 891 | 1891 | 1891 | 1891 | 182 | 183 | TUAAAA | BAAAAA | HHHHxx
+ 3420 | 2 | 0 | 0 | 0 | 0 | 20 | 420 | 1420 | 3420 | 3420 | 40 | 41 | OBAAAA | CAAAAA | OOOOxx
+(3 rows)
+
+-- Fetch the next batch
+FETCH 4 in foo4;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 9850 | 3 | 0 | 2 | 0 | 10 | 50 | 850 | 1850 | 4850 | 9850 | 100 | 101 | WOAAAA | DAAAAA | VVVVxx
+ 7164 | 4 | 0 | 0 | 4 | 4 | 64 | 164 | 1164 | 2164 | 7164 | 128 | 129 | OPAAAA | EAAAAA | AAAAxx
+ 8009 | 5 | 1 | 1 | 9 | 9 | 9 | 9 | 9 | 3009 | 8009 | 18 | 19 | BWAAAA | FAAAAA | HHHHxx
+(3 rows)
+
+FETCH backward 4 in foo4;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 7164 | 4 | 0 | 0 | 4 | 4 | 64 | 164 | 1164 | 2164 | 7164 | 128 | 129 | OPAAAA | EAAAAA | AAAAxx
+ 9850 | 3 | 0 | 2 | 0 | 10 | 50 | 850 | 1850 | 4850 | 9850 | 100 | 101 | WOAAAA | DAAAAA | VVVVxx
+ 3420 | 2 | 0 | 0 | 0 | 0 | 20 | 420 | 1420 | 3420 | 3420 | 40 | 41 | OBAAAA | CAAAAA | OOOOxx
+(3 rows)
+
+-- Check cursor with hold
+DECLARE foo5 CURSOR WITH HOLD FOR SELECT * FROM tenk2 LIMIT 30;
+commit;
+FETCH 5 FROM foo5;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 8800 | 0 | 0 | 0 | 0 | 0 | 0 | 800 | 800 | 3800 | 8800 | 0 | 1 | MAAAAA | AAAAAA | AAAAxx
+ 1891 | 1 | 1 | 3 | 1 | 11 | 91 | 891 | 1891 | 1891 | 1891 | 182 | 183 | TUAAAA | BAAAAA | HHHHxx
+ 3420 | 2 | 0 | 0 | 0 | 0 | 20 | 420 | 1420 | 3420 | 3420 | 40 | 41 | OBAAAA | CAAAAA | OOOOxx
+(3 rows)
+
+FETCH 5 FROM foo5;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 9850 | 3 | 0 | 2 | 0 | 10 | 50 | 850 | 1850 | 4850 | 9850 | 100 | 101 | WOAAAA | DAAAAA | VVVVxx
+ 7164 | 4 | 0 | 0 | 4 | 4 | 64 | 164 | 1164 | 2164 | 7164 | 128 | 129 | OPAAAA | EAAAAA | AAAAxx
+ 8009 | 5 | 1 | 1 | 9 | 9 | 9 | 9 | 9 | 3009 | 8009 | 18 | 19 | BWAAAA | FAAAAA | HHHHxx
+(3 rows)
+
+-- FETCH ALL is not limited by cursor_fetch_limit
+FETCH ALL FROM foo5;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 5057 | 6 | 1 | 1 | 7 | 17 | 57 | 57 | 1057 | 57 | 5057 | 114 | 115 | NMAAAA | GAAAAA | OOOOxx
+ 6701 | 7 | 1 | 1 | 1 | 1 | 1 | 701 | 701 | 1701 | 6701 | 2 | 3 | TXAAAA | HAAAAA | VVVVxx
+ 4321 | 8 | 1 | 1 | 1 | 1 | 21 | 321 | 321 | 4321 | 4321 | 42 | 43 | FKAAAA | IAAAAA | AAAAxx
+ 3043 | 9 | 1 | 3 | 3 | 3 | 43 | 43 | 1043 | 3043 | 3043 | 86 | 87 | BNAAAA | JAAAAA | HHHHxx
+ 1314 | 10 | 0 | 2 | 4 | 14 | 14 | 314 | 1314 | 1314 | 1314 | 28 | 29 | OYAAAA | KAAAAA | OOOOxx
+ 1504 | 11 | 0 | 0 | 4 | 4 | 4 | 504 | 1504 | 1504 | 1504 | 8 | 9 | WFAAAA | LAAAAA | VVVVxx
+ 5222 | 12 | 0 | 2 | 2 | 2 | 22 | 222 | 1222 | 222 | 5222 | 44 | 45 | WSAAAA | MAAAAA | AAAAxx
+ 6243 | 13 | 1 | 3 | 3 | 3 | 43 | 243 | 243 | 1243 | 6243 | 86 | 87 | DGAAAA | NAAAAA | HHHHxx
+ 5471 | 14 | 1 | 3 | 1 | 11 | 71 | 471 | 1471 | 471 | 5471 | 142 | 143 | LCAAAA | OAAAAA | OOOOxx
+ 5006 | 15 | 0 | 2 | 6 | 6 | 6 | 6 | 1006 | 6 | 5006 | 12 | 13 | OKAAAA | PAAAAA | VVVVxx
+ 5387 | 16 | 1 | 3 | 7 | 7 | 87 | 387 | 1387 | 387 | 5387 | 174 | 175 | FZAAAA | QAAAAA | AAAAxx
+ 5785 | 17 | 1 | 1 | 5 | 5 | 85 | 785 | 1785 | 785 | 5785 | 170 | 171 | NOAAAA | RAAAAA | HHHHxx
+ 6621 | 18 | 1 | 1 | 1 | 1 | 21 | 621 | 621 | 1621 | 6621 | 42 | 43 | RUAAAA | SAAAAA | OOOOxx
+ 6969 | 19 | 1 | 1 | 9 | 9 | 69 | 969 | 969 | 1969 | 6969 | 138 | 139 | BIAAAA | TAAAAA | VVVVxx
+ 9460 | 20 | 0 | 0 | 0 | 0 | 60 | 460 | 1460 | 4460 | 9460 | 120 | 121 | WZAAAA | UAAAAA | AAAAxx
+ 59 | 21 | 1 | 3 | 9 | 19 | 59 | 59 | 59 | 59 | 59 | 118 | 119 | HCAAAA | VAAAAA | HHHHxx
+ 8020 | 22 | 0 | 0 | 0 | 0 | 20 | 20 | 20 | 3020 | 8020 | 40 | 41 | MWAAAA | WAAAAA | OOOOxx
+ 7695 | 23 | 1 | 3 | 5 | 15 | 95 | 695 | 1695 | 2695 | 7695 | 190 | 191 | ZJAAAA | XAAAAA | VVVVxx
+ 3442 | 24 | 0 | 2 | 2 | 2 | 42 | 442 | 1442 | 3442 | 3442 | 84 | 85 | KCAAAA | YAAAAA | AAAAxx
+ 5119 | 25 | 1 | 3 | 9 | 19 | 19 | 119 | 1119 | 119 | 5119 | 38 | 39 | XOAAAA | ZAAAAA | HHHHxx
+ 646 | 26 | 0 | 2 | 6 | 6 | 46 | 646 | 646 | 646 | 646 | 92 | 93 | WYAAAA | ABAAAA | OOOOxx
+ 9605 | 27 | 1 | 1 | 5 | 5 | 5 | 605 | 1605 | 4605 | 9605 | 10 | 11 | LFAAAA | BBAAAA | VVVVxx
+ 263 | 28 | 1 | 3 | 3 | 3 | 63 | 263 | 263 | 263 | 263 | 126 | 127 | DKAAAA | CBAAAA | AAAAxx
+ 3269 | 29 | 1 | 1 | 9 | 9 | 69 | 269 | 1269 | 3269 | 3269 | 138 | 139 | TVAAAA | DBAAAA | HHHHxx
+(24 rows)
+
+CLOSE foo5;
+reset cursor_fetch_limit;
diff --git a/src/test/regress/sql/portals.sql b/src/test/regress/sql/portals.sql
index fc4cccb96c0..17c389b7070 100644
--- a/src/test/regress/sql/portals.sql
+++ b/src/test/regress/sql/portals.sql
@@ -605,3 +605,31 @@ drop table toasted_data;
fetch all in held_portal;
reset default_toast_compression;
+
+-- Check memory limit for fetching from cursor
+set cursor_fetch_limit = '512B';
+
+BEGIN;
+DECLARE foo1 SCROLL CURSOR FOR SELECT * FROM tenk1 ORDER BY unique2;
+DECLARE foo2 SCROLL CURSOR FOR SELECT * FROM tenk2;
+DECLARE foo3 SCROLL CURSOR FOR SELECT * FROM tenk1 ORDER BY unique2;
+DECLARE foo4 SCROLL CURSOR FOR SELECT * FROM tenk2;
+
+FETCH 1 in foo1;
+FETCH 2 in foo2;
+FETCH 3 in foo3;
+-- Well, we've reached the limit
+FETCH 4 in foo4;
+-- Fetch the next batch
+FETCH 4 in foo4;
+FETCH backward 4 in foo4;
+
+-- Check cursor with hold
+DECLARE foo5 CURSOR WITH HOLD FOR SELECT * FROM tenk2 LIMIT 30;
+commit;
+FETCH 5 FROM foo5;
+FETCH 5 FROM foo5;
+-- FETCH ALL is not limited by cursor_fetch_limit
+FETCH ALL FROM foo5;
+CLOSE foo5;
+reset cursor_fetch_limit;
--
2.43.0
^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2026-01-23 06:18 UTC | newest]
Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2024-04-28 11:00 [PATCH v17 5/8] Row pattern recognition patch (executor). Tatsuo Ishii <[email protected]>
2026-01-13 10:44 Re: Limit memory usage by postgres_fdw batches Alexander Pyhalov <[email protected]>
2026-01-23 06:18 ` Re: Limit memory usage by postgres_fdw batches Alexander Pyhalov <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox