public inbox for [email protected]
help / color / mirror / Atom feed[PATCH v9 4/7] Row pattern recognition patch (executor).
6+ messages / 4 participants
[nested] [flat]
* [PATCH v9 4/7] Row pattern recognition patch (executor).
@ 2023-10-04 05:51 Tatsuo Ishii <[email protected]>
0 siblings, 0 replies; 6+ messages in thread
From: Tatsuo Ishii @ 2023-10-04 05:51 UTC (permalink / raw)
---
src/backend/executor/nodeWindowAgg.c | 1021 +++++++++++++++++++++++++-
src/backend/utils/adt/windowfuncs.c | 37 +-
src/include/catalog/pg_proc.dat | 6 +
src/include/nodes/execnodes.h | 26 +
4 files changed, 1077 insertions(+), 13 deletions(-)
diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c
index 77724a6daa..5da1bb5a6f 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,12 @@ typedef struct WindowStatePerAggData
bool restart; /* need to restart this agg in this cycle? */
} WindowStatePerAggData;
+typedef struct StringSet {
+ StringInfo *str_set;
+ Size set_size; /* current array allocation size in number of items */
+ int set_index; /* current used size */
+} StringSet;
+
static void initialize_windowaggregate(WindowAggState *winstate,
WindowStatePerFunc perfuncstate,
WindowStatePerAgg peraggstate);
@@ -182,8 +190,9 @@ static void begin_partition(WindowAggState *winstate);
static void spool_tuples(WindowAggState *winstate, int64 pos);
static void release_partition(WindowAggState *winstate);
-static int row_is_in_frame(WindowAggState *winstate, int64 pos,
+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 +204,37 @@ static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
static bool are_peers(WindowAggState *winstate, TupleTableSlot *slot1,
TupleTableSlot *slot2);
-static bool window_gettupleslot(WindowObject winobj, int64 pos,
- TupleTableSlot *slot);
+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);
+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);
/*
* initialize_windowaggregate
@@ -673,6 +710,7 @@ eval_windowaggregates(WindowAggState *winstate)
WindowObject agg_winobj;
TupleTableSlot *agg_row_slot;
TupleTableSlot *temp_slot;
+ bool agg_result_isnull;
numaggs = winstate->numaggs;
if (numaggs == 0)
@@ -778,6 +816,9 @@ eval_windowaggregates(WindowAggState *winstate)
* 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.
+ *
+ * - if RPR is enabled and skip mode is SKIP TO NEXT ROW,
+ * we restart aggregation too.
*----------
*/
numaggs_restart = 0;
@@ -788,8 +829,11 @@ 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) &&
+ winstate->rpSkipTo == ST_NEXT_ROW))
{
+ elog(DEBUG1, "peraggstate->restart is set");
peraggstate->restart = true;
numaggs_restart++;
}
@@ -861,8 +905,10 @@ eval_windowaggregates(WindowAggState *winstate)
* If we created a mark pointer for aggregates, keep it pushed up to frame
* head, so that tuplestore can discard unnecessary rows.
*/
+#ifdef NOT_USED
if (agg_winobj->markptr >= 0)
WinSetMarkPosition(agg_winobj, winstate->frameheadpos);
+#endif
/*
* Now restart the aggregates that require it.
@@ -917,6 +963,29 @@ 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;
+ }
+
+ agg_result_isnull = false;
+ /* RPR is defined? */
+ if (rpr_is_defined(winstate))
+ {
+ /*
+ * If the skip mode is SKIP TO PAST LAST ROW and we already know that
+ * current row is a skipped row, we don't need to accumulate rows,
+ * just return NULL. Note that for unamtched row, we need to do
+ * aggregation so that count(*) shows 0, rather than NULL.
+ */
+ if (winstate->rpSkipTo == ST_PAST_LAST_ROW &&
+ get_reduced_frame_map(winstate, winstate->currentpos) == RF_SKIPPED)
+ agg_result_isnull = true;
}
/*
@@ -930,6 +999,11 @@ eval_windowaggregates(WindowAggState *winstate)
{
int ret;
+ elog(DEBUG1, "===== loop in frame starts: " INT64_FORMAT, winstate->aggregatedupto);
+
+ if (agg_result_isnull)
+ break;
+
/* Fetch next row if we didn't already */
if (TupIsNull(agg_row_slot))
{
@@ -945,9 +1019,28 @@ 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))
+ {
+ /*
+ * 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;
+ }
+
/* Set tuple context for evaluation of aggregate arguments */
winstate->tmpcontext->ecxt_outertuple = agg_row_slot;
@@ -976,6 +1069,7 @@ next_tuple:
ExecClearTuple(agg_row_slot);
}
+
/* The frame's end is not supposed to move backwards, ever */
Assert(aggregatedupto_nonrestarted <= winstate->aggregatedupto);
@@ -996,6 +1090,16 @@ next_tuple:
peraggstate,
result, isnull);
+ /*
+ * RPR is defined and we just return NULL because skip mode is SKIP
+ * TO PAST LAST ROW and current row is skipped row.
+ */
+ if (agg_result_isnull)
+ {
+ *isnull = true;
+ *result = (Datum) 0;
+ }
+
/*
* save the result in case next row shares the same frame.
*
@@ -1090,6 +1194,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 +2158,8 @@ ExecWindowAgg(PlanState *pstate)
CHECK_FOR_INTERRUPTS();
+ elog(DEBUG1, "ExecWindowAgg called. pos: " INT64_FORMAT , winstate->currentpos);
+
if (winstate->status == WINDOWAGG_DONE)
return NULL;
@@ -2221,6 +2328,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 +2506,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)));
@@ -2483,6 +2604,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 +2798,39 @@ 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;
+
+ elog(DEBUG1, "defineVariable name: %s", name);
+ 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 +2838,57 @@ 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;
+ elog(DEBUG1, "PREV/NEXT's varno is rewritten to: %d", var->varno);
+ }
+ }
+ return expression_tree_walker(node, attno_map_walker, NULL);
+}
+
/* -----------------
* ExecEndWindowAgg
* -----------------
@@ -2723,6 +2938,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 +3300,7 @@ 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 +3620,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 +3734,21 @@ 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 +3815,12 @@ 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 +3839,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 +3876,731 @@ 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;
+ elog(DEBUG1, "row_is_in_reduced_frame returns %d: pos: " INT64_FORMAT, rtn, pos);
+ 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;
+ }
+
+ elog(DEBUG1, "row_is_in_reduced_frame returns %d: pos: " INT64_FORMAT, rtn, pos);
+ return rtn;
+}
+
+#define REDUCED_FRAME_MAP_INIT_SIZE 1024L
+
+/*
+ * 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
+ */
+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 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];
+}
+
+/*
+ * 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 str_set_index = 0;
+
+ /*
+ * Set of pattern variables evaluted 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;
+
+ /*
+ * Loop over until none of pattern matches or encounters end of frame.
+ */
+ for (;;)
+ {
+ int64 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));
+ char *quantifier = strVal(lfirst(lc2));
+
+ elog(DEBUG1, "pos: " INT64_FORMAT " pattern vname: %s quantifier: %s", pos, vname, quantifier);
+
+ expression_result = false;
+
+ /* evaluate row pattern against current row */
+ result_pos = evaluate_pattern(winobj, pos, vname, encoded_str, &expression_result);
+ if (expression_result)
+ {
+ elog(DEBUG1, "expression result is true");
+ 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);
+
+ elog(DEBUG1, "pos: " INT64_FORMAT " str_set_index: %d encoded_str: %s", pos, str_set_index, encoded_str->data);
+
+ /* 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;
+ }
+
+ elog(DEBUG2, "pos: " INT64_FORMAT " encoded_str: %s", pos, encoded_str->data);
+
+ /* build regular expression */
+ pattern_str = makeStringInfo();
+ appendStringInfoChar(pattern_str, '^');
+ 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]);
+ elog(DEBUG1, "vname: %s initial: %c quantifier: %s", vname, initial, quantifier);
+ }
+
+ elog(DEBUG2, "pos: " INT64_FORMAT " pattern: %s", pos, pattern_str->data);
+
+ /* look for matching pattern variable sequence */
+ num_matched_rows = search_str_set(pattern_str->data, 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
+ {
+ int64 i;
+
+ 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;
+}
+
+/*
+ * Perform pattern matching using pattern against str_set. str_set is a set
+ * of StringInfo. Each StringInfo has a string comprising initial characters
+ * of pattern variables being true in a row. Returns the longest number of
+ * the matching rows.
+ */
+static
+int search_str_set(char *pattern, StringSet *str_set)
+{
+ int set_size; /* number of rows in the set */
+ int resultlen = 0;
+ int index;
+ StringSet *old_str_set, *new_str_set;
+ int new_str_size;
+
+ set_size = string_set_get_size(str_set);
+ new_str_set = string_set_init();
+
+ /*
+ * 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.
+ */
+ for (index = 0; index < set_size; index++)
+ {
+ StringInfo str; /* search target row */
+ char *p;
+ int old_set_size;
+ int i;
+
+ 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 in previous result
+ * 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);
+ p++; /* next pattern variable */
+ }
+ }
+ else
+ {
+ 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 old = string_set_get(old_str_set, i);
+
+ p = str->data;
+
+ /*
+ * loop over each pattern variable char in the input set.
+ */
+ while (*p)
+ {
+ StringInfo new = makeStringInfo();
+
+ /* copy source string */
+ appendStringInfoString(new, old->data);
+ /* add pattern variable char */
+ appendStringInfoChar(new, *p);
+ /* add new one to string set */
+ string_set_add(new_str_set, new);
+ p++; /* next pattern variable */
+ }
+ }
+ /* we no long 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);
+
+ for (index = 0; index < new_str_size; index++)
+ {
+ int len;
+ StringInfo s;
+
+ s = string_set_get(new_str_set, index);
+ if (s == NULL)
+ continue; /* no data */
+
+ 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 long 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;
+
+
+ /*
+ * 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(cstring_to_text(encoded_str)),
+ PointerGetDatum(cstring_to_text(pattern)))) > 0)
+ {
+ d = DirectFunctionCall2Coll(textregexsubstr,
+ DEFAULT_COLLATION_OID,
+ PointerGetDatum(cstring_to_text(encoded_str)),
+ PointerGetDatum(cstring_to_text(pattern)));
+ if (d != 0)
+ {
+ res = DatumGetTextPP(d);
+ substr = text_to_cstring(res);
+ len = strlen(substr);
+ }
+ }
+ return len;
+}
+
+
+/*
+ * Evaluate expression associated with PATTERN variable vname.
+ * relpos is relative row position in a frame (starting from 0).
+ * "quantifier" is the quatifier part of the PATTERN regular expression.
+ * Currently only '+' is allowed.
+ * result is out paramater representing the expression evaluation result
+ * is true of false.
+ * Return values are:
+ * >=0: the last match absolute row position
+ * other wise 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;
+
+ forthree (lc1, winstate->defineVariableList, lc2, winstate->defineClauseList, lc3, winstate->defineInitial)
+ {
+ char initial;
+ 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 */
+ elog(DEBUG1, "expression for %s is NULL at row: " INT64_FORMAT, vname, current_pos);
+ *result = false;
+ }
+ else
+ {
+ if (!DatumGetBool(eval_result))
+ {
+ /* expression is false */
+ elog(DEBUG1, "expression for %s is false at row: " INT64_FORMAT, vname, current_pos);
+ *result = false;
+ }
+ else
+ {
+ /* expression is true */
+ elog(DEBUG1, "expression for %s is true at row: " INT64_FORMAT, vname, current_pos);
+ appendStringInfoChar(encoded_str, initial);
+ *result = true;
+ }
+ }
+ break;
+ }
+
+ if (out_of_frame)
+ {
+ *result = false;
+ return -1;
+ }
+ }
+ return current_pos;
+}
+
+/*
+ * 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))
+ {
+ elog(DEBUG1, "current row is out of partition at:" INT64_FORMAT, current_pos);
+ return false;
+ }
+ ret = row_is_in_frame(winstate, current_pos, slot);
+ if (ret <= 0)
+ {
+ elog(DEBUG1, "current row is out of frame at: " INT64_FORMAT, current_pos);
+ return false;
+ }
+ econtext->ecxt_outertuple = slot;
+
+ /* for PREV */
+ if (current_pos > 0)
+ {
+ slot = winstate->prev_slot;
+ if (!window_gettupleslot(winobj, current_pos - 1, slot))
+ {
+ elog(DEBUG1, "previous row is out of partition at: " INT64_FORMAT, current_pos - 1);
+ econtext->ecxt_scantuple = winstate->null_slot;
+ }
+ else
+ {
+ ret = row_is_in_frame(winstate, current_pos - 1, slot);
+ if (ret <= 0)
+ {
+ elog(DEBUG1, "previous row is out of frame at: " INT64_FORMAT, current_pos - 1);
+ 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))
+ {
+ elog(DEBUG1, "next row is out of partiton at: " INT64_FORMAT, current_pos + 1);
+ econtext->ecxt_innertuple = winstate->null_slot;
+ }
+ else
+ {
+ ret = row_is_in_frame(winstate, current_pos + 1, slot);
+ if (ret <= 0)
+ {
+ elog(DEBUG1, "next row is out of frame at: " INT64_FORMAT, current_pos + 1);
+ econtext->ecxt_innertuple = winstate->null_slot;
+ }
+ else
+ econtext->ecxt_innertuple = slot;
+ }
+ return true;
+}
+
+/*
+ * 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;
+}
+
+/*
+ * 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;
+}
+
+/*
+ * 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];
+}
+
+/*
+ * Returns the size of StringSet.
+ */
+static
+int string_set_get_size(StringSet *string_set)
+{
+ return string_set->set_index;
+}
+
+/*
+ * 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];
+ pfree(str->data);
+ pfree(str);
+ }
+ pfree(string_set);
+}
diff --git a/src/backend/utils/adt/windowfuncs.c b/src/backend/utils/adt/windowfuncs.c
index b87a624fb2..9ebcc7b5d2 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/supportnodes.h"
#include "utils/builtins.h"
#include "windowapi.h"
@@ -36,11 +39,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.
*/
@@ -673,7 +684,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();
@@ -713,3 +724,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 f0b7b9cbd8..24d288336a 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -10416,6 +10416,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 108d69ba28..28d098b1cf 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2470,6 +2470,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 */
@@ -2518,6 +2523,15 @@ 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 */
@@ -2554,6 +2568,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(Wed_Oct__4_15_03_28_2023_821)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v9-0005-Row-pattern-recognition-patch-docs.patch"
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Restrict publishing of partitioned table with a foreign table as partition
@ 2025-02-05 06:42 vignesh C <[email protected]>
2025-02-05 08:44 ` Re: Restrict publishing of partitioned table with a foreign table as partition Álvaro Herrera <[email protected]>
0 siblings, 1 reply; 6+ messages in thread
From: vignesh C @ 2025-02-05 06:42 UTC (permalink / raw)
To: Álvaro Herrera <[email protected]>; +Cc: Sergey Tatarintsev <[email protected]>; Shlok Kyal <[email protected]>; [email protected]
On Tue, 4 Feb 2025 at 21:21, Álvaro Herrera <[email protected]> wrote:
>
> On 2025-Feb-04, vignesh C wrote:
>
> > We should throw an error for partitioned tables that contain foreign
> > partitions, as this would include the data from these foreign tables
> > during the initial sync, while incremental changes would not be
> > replicated.
>
> Hmm, I would support the idea of allowing partitioned tables containing
> foreign partitions into publications, where only the tuples in
> non-foreign partitions would be transmitted -- both during initial sync
> and during replicated DML. This way, any partitioned tables with mixed
> local and foreign partitions would work okay for replication. If the
> subscriber wants the data in the foreign partitions, they can create the
> foreign table on the subscription side and all is well. From a users
> POV this is probably the most useful.
>
> Another aspect is the case where you create the publication first and
> the foreign table later. I didn't spot in the proposed patch any code
> that would check whether a publication including this schema exists,
> which we would have to do if we wanted to be watertight about rejecting
> publications with foreign tables. But I think that would be a bad
> direction to go it.
>
> I'd rather have the table-listing code for publications skip any foreign
> tables. For instance, GetPubPartitionOptionRelations() and
> PublicationAddTables() should skip tables with relkind RELKIND_FOREIGN.
> The sync code should also silently ignore all foreign tables.
We can maintain the behavior you suggested when the
PUBLISH_VIA_PARTITION_ROOT option is set to false. However, when
PUBLISH_VIA_PARTITION_ROOT is true, the table data is copied from the
root table (as intended by the user), which will also include the
foreign table data. In this case, wouldn’t it be better to throw an
error?
Regards,
Vignesh
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Restrict publishing of partitioned table with a foreign table as partition
2025-02-05 06:42 Re: Restrict publishing of partitioned table with a foreign table as partition vignesh C <[email protected]>
@ 2025-02-05 08:44 ` Álvaro Herrera <[email protected]>
2025-02-11 11:25 ` Re: Restrict publishing of partitioned table with a foreign table as partition Shlok Kyal <[email protected]>
0 siblings, 1 reply; 6+ messages in thread
From: Álvaro Herrera @ 2025-02-05 08:44 UTC (permalink / raw)
To: vignesh C <[email protected]>; +Cc: Sergey Tatarintsev <[email protected]>; Shlok Kyal <[email protected]>; [email protected]
On 2025-Feb-05, vignesh C wrote:
> We can maintain the behavior you suggested when the
> PUBLISH_VIA_PARTITION_ROOT option is set to false. However, when
> PUBLISH_VIA_PARTITION_ROOT is true, the table data is copied from the
> root table (as intended by the user), which will also include the
> foreign table data. In this case, wouldn’t it be better to throw an
> error?
It sounds to me a reasonable restriction that you can only add a
partitioned table to a publication if publish_via_partition_root=false.
Then the case of hybrid partitioned tables is supported, but it requires
that changes are published directly by partitions, which is sensible
anyway.
In this case, during CREATE FOREIGN TABLE of a partition with this
condition, we must check whether any publications include the schema
that the table is being created on (or attached, for ALTER TABLE ATTACH
PARTITION), and whether there are any publications that are FOR ALL
TABLES that would be affected.
(If we later figure out a way to allow publish_via_partition_root and
skip the tuples in foreign partitions, it's easy to remove the
restriction.)
--
Álvaro Herrera 48°01'N 7°57'E — https://www.EnterpriseDB.com/
"That sort of implies that there are Emacs keystrokes which aren't obscure.
I've been using it daily for 2 years now and have yet to discover any key
sequence which makes any sense." (Paul Thomas)
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Restrict publishing of partitioned table with a foreign table as partition
2025-02-05 06:42 Re: Restrict publishing of partitioned table with a foreign table as partition vignesh C <[email protected]>
2025-02-05 08:44 ` Re: Restrict publishing of partitioned table with a foreign table as partition Álvaro Herrera <[email protected]>
@ 2025-02-11 11:25 ` Shlok Kyal <[email protected]>
2025-02-13 05:54 ` Re: Restrict publishing of partitioned table with a foreign table as partition vignesh C <[email protected]>
0 siblings, 1 reply; 6+ messages in thread
From: Shlok Kyal @ 2025-02-11 11:25 UTC (permalink / raw)
To: Álvaro Herrera <[email protected]>; +Cc: vignesh C <[email protected]>; Sergey Tatarintsev <[email protected]>; [email protected]
On Wed, 5 Feb 2025 at 14:14, Álvaro Herrera <[email protected]> wrote:
>
> On 2025-Feb-05, vignesh C wrote:
>
> > We can maintain the behavior you suggested when the
> > PUBLISH_VIA_PARTITION_ROOT option is set to false. However, when
> > PUBLISH_VIA_PARTITION_ROOT is true, the table data is copied from the
> > root table (as intended by the user), which will also include the
> > foreign table data. In this case, wouldn’t it be better to throw an
> > error?
>
> It sounds to me a reasonable restriction that you can only add a
> partitioned table to a publication if publish_via_partition_root=false.
> Then the case of hybrid partitioned tables is supported, but it requires
> that changes are published directly by partitions, which is sensible
> anyway.
>
I have created a patch with the above approach.
We can create a publication on a partition table with foreign
partition when publish_via_partition_root=false. In this case foreign
partitions will not be published. And if
publish_via_partition_root=true we will throw an error.
Please find the v5 patch.
> In this case, during CREATE FOREIGN TABLE of a partition with this
> condition, we must check whether any publications include the schema
> that the table is being created on (or attached, for ALTER TABLE ATTACH
> PARTITION), and whether there are any publications that are FOR ALL
> TABLES that would be affected.
>
I have handled the above cases and added tests for the same.
> (If we later figure out a way to allow publish_via_partition_root and
> skip the tuples in foreign partitions, it's easy to remove the
> restriction.)
>
Thanks and Regards,
Shlok Kyal
Attachments:
[application/octet-stream] v5-0001-Restrict-publishing-of-partitioned-table-with-for.patch (25.9K, ../../CANhcyEXxjq9U7BXxCba3Njz+eHpNzAsSVY2GzbV=8iy5j=UAUA@mail.gmail.com/2-v5-0001-Restrict-publishing-of-partitioned-table-with-for.patch)
download | inline diff:
From 266366aded45b4c7cdc1525c45286ef198f021c3 Mon Sep 17 00:00:00 2001
From: Shlok Kyal <[email protected]>
Date: Tue, 11 Feb 2025 15:32:55 +0530
Subject: [PATCH v5] Restrict publishing of partitioned table with foreign
partition
Logical replication of foreign table is not supported and we throw an
error in this case. But when create a publication on a partitioned
table that has a foreign partition, the initial sync of such table is
successful and we should avoid such cases.
Current Behaviour, when publication is created:
1. with publish_via_partition_root = true
Root table is published and the initial data of foreign partitions
are replicated.
2. with publish_via_partition_root = false and FOR ALL TABLES
All leaf tables except foreign partitions are published.
3. with publish_via_partition_root = false and
FOR TABLE/ FOR TABLES IN SCHEMA
All leaf tables are published. Initial data of foreign partitions are
replicated.
With this patch we have following behaviour:
1. with publish_via_partition_root = true
We throw an error when we try to publish a foreign partititon. Error
is thrown when we try to create a publication on (or add to existing
publication) a partitioned table with foreign partition, when try to
create a foreign partition and when we try to attach foreign table (or
a table with foreign partition) to existing published tables.
2. with publish_via_partition_root = false
We skip publishing foreign partition which are part of the publication.
This is done by avoid adding foreign partitions in pg_subscription_rel
catalog table.
---
src/backend/catalog/pg_publication.c | 181 ++++++++++++++++++++--
src/backend/commands/foreigncmds.c | 47 ++++++
src/backend/commands/publicationcmds.c | 47 ++++++
src/backend/commands/tablecmds.c | 53 +++++++
src/include/catalog/pg_publication.h | 7 +
src/test/regress/expected/publication.out | 89 +++++++++++
src/test/regress/sql/publication.sql | 81 ++++++++++
7 files changed, 488 insertions(+), 17 deletions(-)
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index d6f94db5d9..d022ff71b5 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -53,7 +53,7 @@ typedef struct
* error if not.
*/
static void
-check_publication_add_relation(Relation targetrel)
+check_publication_add_relation(Relation targetrel, Publication *pub)
{
/* Must be a regular or partitioned table */
if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION &&
@@ -64,6 +64,19 @@ check_publication_add_relation(Relation targetrel)
RelationGetRelationName(targetrel)),
errdetail_relkind_not_supported(RelationGetForm(targetrel)->relkind)));
+ /*
+ * publish_via_root_partition cannot be true if it is a partitioned table
+ * and has any foreign partition
+ */
+ if (pub->pubviaroot &&
+ check_partrel_has_foreign_table(RelationGetForm(targetrel)))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot set parameter \"%s\" to true for publication \"%s\"",
+ "publish_via_partition_root", pub->name),
+ errdetail("partition table \"%s\" in publication contains a foreign partition",
+ RelationGetRelationName(targetrel))));
+
/* Can't be system table */
if (IsCatalogRelation(targetrel))
ereport(ERROR,
@@ -304,7 +317,7 @@ check_and_fetch_column_list(Publication *pub, Oid relid, MemoryContext mcxt,
/*
* Gets the relations based on the publication partition option for a specified
- * relation.
+ * relation. Foreign tables are not included.
*/
List *
GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
@@ -313,25 +326,21 @@ GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE &&
pub_partopt != PUBLICATION_PART_ROOT)
{
- List *all_parts = find_all_inheritors(relid, NoLock,
- NULL);
+ List *all_parts = find_all_inheritors(relid, NoLock, NULL);
- if (pub_partopt == PUBLICATION_PART_ALL)
- result = list_concat(result, all_parts);
- else if (pub_partopt == PUBLICATION_PART_LEAF)
+ foreach_oid(partOid, all_parts)
{
- ListCell *lc;
+ char relkind = get_rel_relkind(partOid);
- foreach(lc, all_parts)
- {
- Oid partOid = lfirst_oid(lc);
+ if (relkind == RELKIND_FOREIGN_TABLE)
+ continue;
- if (get_rel_relkind(partOid) != RELKIND_PARTITIONED_TABLE)
- result = lappend_oid(result, partOid);
- }
+ if (pub_partopt == PUBLICATION_PART_LEAF &&
+ relkind == RELKIND_PARTITIONED_TABLE)
+ continue;
+
+ result = lappend_oid(result, partOid);
}
- else
- Assert(false);
}
else
result = lappend_oid(result, relid);
@@ -463,7 +472,7 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
RelationGetRelationName(targetrel), pub->name)));
}
- check_publication_add_relation(targetrel);
+ check_publication_add_relation(targetrel, pub);
/* Validate and translate column names into a Bitmapset of attnums. */
attnums = pub_collist_validate(pri->relation, pri->columns);
@@ -703,6 +712,13 @@ publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
check_publication_add_schema(schemaid);
+ /*
+ * If publish_via_partition_root is true, check if schema has any foreign
+ * partition
+ */
+ if (pub->pubviaroot)
+ check_foreign_tables_in_schema(schemaid, pub->name);
+
/* Form a tuple */
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
@@ -1332,3 +1348,134 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
SRF_RETURN_DONE(funcctx);
}
+
+/* Check if a partitioned table has a foreign partition */
+bool
+check_partrel_has_foreign_table(Form_pg_class relform)
+{
+ bool has_foreign_tbl = false;
+
+ if (relform->relkind == RELKIND_PARTITIONED_TABLE)
+ {
+ List *relids = NIL;
+
+ relids = find_all_inheritors(relform->oid, NoLock, NULL);
+
+ foreach_oid(relid, relids)
+ {
+ Relation rel = table_open(relid, AccessShareLock);
+
+ if (RelationGetForm(rel)->relkind == RELKIND_FOREIGN_TABLE)
+ has_foreign_tbl = true;
+
+ table_close(rel, AccessShareLock);
+
+ if (has_foreign_tbl)
+ break;
+ }
+ }
+
+ return has_foreign_tbl;
+}
+
+/*
+ * Check if a schema has a partition table which has a foreign partition.
+ * The partition tables in a schema can have partitions in other schema. We also
+ * need to check if such partitions are foreign partition.
+ */
+void
+check_foreign_tables_in_schema(Oid schemaid, char *pubname)
+{
+ Relation classRel;
+ ScanKeyData key[2];
+ TableScanDesc scan;
+ HeapTuple tuple;
+
+ classRel = table_open(RelationRelationId, AccessShareLock);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_class_relnamespace,
+ BTEqualStrategyNumber, F_OIDEQ,
+ schemaid);
+ ScanKeyInit(&key[1],
+ Anum_pg_class_relkind,
+ BTEqualStrategyNumber, F_CHAREQ,
+ CharGetDatum(RELKIND_PARTITIONED_TABLE));
+
+ scan = table_beginscan_catalog(classRel, 2, key);
+ while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ {
+ Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
+
+ if (check_partrel_has_foreign_table(relForm))
+ {
+ List *ancestors = get_partition_ancestors(relForm->oid);
+ Oid parent_oid = relForm->oid;
+ char *parent_name;
+
+ foreach_oid(ancestor, ancestors)
+ {
+ Oid ancestor_schemaid = get_rel_namespace(ancestor);
+
+ if (ancestor_schemaid == schemaid)
+ parent_oid = ancestor;
+ }
+
+ list_free(ancestors);
+ parent_name = get_rel_name(parent_oid);
+
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot set parameter \"%s\" to true for publication \"%s\"",
+ "publish_via_partition_root", pubname),
+ errdetail("partition table \"%s\" in publication contains a foreign partition",
+ parent_name)));
+ }
+ }
+
+ table_endscan(scan);
+ table_close(classRel, AccessShareLock);
+}
+
+/* Check if any foreign table is a partition table */
+void
+check_foreign_tables(char *pubname)
+{
+ Relation classRel;
+ ScanKeyData key[1];
+ TableScanDesc scan;
+ HeapTuple tuple;
+
+ classRel = table_open(RelationRelationId, AccessShareLock);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_class_relkind,
+ BTEqualStrategyNumber, F_CHAREQ,
+ CharGetDatum(RELKIND_FOREIGN_TABLE));
+
+ scan = table_beginscan_catalog(classRel, 1, key);
+ while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ {
+ Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
+
+ if (relForm->relispartition)
+ {
+ Oid parent_oid;
+ char *parent_name;
+ List *ancestors = get_partition_ancestors(relForm->oid);
+
+ parent_oid = llast_oid(ancestors);
+ parent_name = get_rel_name(parent_oid);
+
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot set parameter \"%s\" to true for publication \"%s\"",
+ "publish_via_partition_root", pubname),
+ errdetail("partition table \"%s\" in publication contains a foreign partition",
+ parent_name)));
+ }
+ }
+
+ table_endscan(scan);
+ table_close(classRel, AccessShareLock);
+}
diff --git a/src/backend/commands/foreigncmds.c b/src/backend/commands/foreigncmds.c
index c14e038d54..f41a24aea3 100644
--- a/src/backend/commands/foreigncmds.c
+++ b/src/backend/commands/foreigncmds.c
@@ -21,6 +21,7 @@
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/objectaccess.h"
+#include "catalog/partition.h"
#include "catalog/pg_foreign_data_wrapper.h"
#include "catalog/pg_foreign_server.h"
#include "catalog/pg_foreign_table.h"
@@ -1423,6 +1424,52 @@ CreateForeignTable(CreateForeignTableStmt *stmt, Oid relid)
ftrel = table_open(ForeignTableRelationId, RowExclusiveLock);
+ /*
+ * Check if it is a foreign partition and the partitioned table is not
+ * published or published with publish_via_partition_root option as false.
+ */
+ if (stmt->base.partbound != NULL)
+ {
+ RangeVar *root = castNode(RangeVar, lfirst(list_head(stmt->base.inhRelations)));
+ Relation rootrel = table_openrv(root, AccessShareLock);
+
+ if (RelationGetForm(rootrel)->relkind == RELKIND_PARTITIONED_TABLE)
+ {
+ Oid schemaid = RelationGetNamespace(rootrel);
+ List *puboids = GetRelationPublications(rootrel->rd_id);
+ List *ancestors;
+
+ puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+ puboids = list_concat_unique_oid(puboids, GetSchemaPublications(schemaid));
+ ancestors = get_partition_ancestors(rootrel->rd_id);
+
+ foreach_oid(ancestor, ancestors)
+ {
+ puboids = list_concat_unique_oid(puboids,
+ GetRelationPublications(ancestor));
+ schemaid = get_rel_namespace(ancestor);
+ puboids = list_concat_unique_oid(puboids,
+ GetSchemaPublications(schemaid));
+ }
+ list_free(ancestors);
+
+ foreach_oid(puboid, puboids)
+ {
+ Publication *pub = GetPublication(puboid);
+
+ if (pub->pubviaroot)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot create table foreign partition \"%s\"",
+ get_rel_name(relid)),
+ errdetail("partition table \"%s\" is published with option publish_via_partition_root",
+ RelationGetRelationName(rootrel))));
+ }
+ }
+
+ table_close(rootrel, AccessShareLock);
+ }
+
/*
* For now the owner cannot be specified on create. Use effective user ID.
*/
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 150a768d16..e71d5408c7 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -876,6 +876,10 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
/* Associate objects with the publication. */
if (stmt->for_all_tables)
{
+ /* Check if any foreign table is a part of partitioned table */
+ if (publish_via_partition_root)
+ check_foreign_tables(stmt->pubname);
+
/* Invalidate relcache so that publication info is rebuilt. */
CacheInvalidateRelcacheAll();
}
@@ -1041,6 +1045,49 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
}
}
+ /*
+ * If publish_via_partition_root is set to true, check if the publication
+ * already have any foreign partition
+ */
+ if (publish_via_partition_root_given && publish_via_partition_root)
+ {
+ List *schemaoids = NIL;
+ List *relids = NIL;
+
+ char *pubname = stmt->pubname;
+
+ if (pubform->puballtables)
+ check_foreign_tables(pubname);
+
+ schemaoids = GetPublicationSchemas(pubform->oid);
+
+ foreach_oid(schemaoid, schemaoids)
+ check_foreign_tables_in_schema(schemaoid, pubname);
+
+ relids = GetPublicationRelations(pubform->oid, PUBLICATION_PART_ROOT);
+
+ foreach_oid(relid, relids)
+ {
+ HeapTuple reltup;
+ Form_pg_class relform;
+
+ reltup = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+ relform = (Form_pg_class) GETSTRUCT(reltup);
+
+ ReleaseSysCache(reltup);
+
+ if (check_partrel_has_foreign_table(relform))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot set parameter \"%s\" to true for publication \"%s\"",
+ "publish_via_partition_root", pubname),
+ errdetail("partition table \"%s\" in publication contains a foreign partition",
+ get_rel_name(relid))));
+ }
+ }
+ }
+
/* Everything ok, form a new tuple. */
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 5823fce934..fccc1082f4 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -19440,6 +19440,59 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot attach temporary relation of another session as partition")));
+ /*
+ * Check if attachrel is a foreign table or a partitioned table with
+ * foreign partition and rel is not part of publication with option
+ * publish_via_partition_root as true.
+ */
+ if (attachrel->rd_rel->relkind == RELKIND_FOREIGN_TABLE ||
+ check_partrel_has_foreign_table(RelationGetForm(attachrel)))
+ {
+ Oid schemaid = RelationGetNamespace(rel);
+ List *puboids = GetRelationPublications(rel->rd_id);
+ List *ancestors;
+ char *relname = get_rel_name(rel->rd_id);
+ char *attachrelname = get_rel_name(attachrel->rd_id);
+
+ puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+ puboids = list_concat_unique_oid(puboids, GetSchemaPublications(schemaid));
+ ancestors = get_partition_ancestors(rel->rd_id);
+
+ foreach_oid(ancestor, ancestors)
+ {
+ puboids = list_concat_unique_oid(puboids,
+ GetRelationPublications(ancestor));
+ schemaid = get_rel_namespace(ancestor);
+ puboids = list_concat_unique_oid(puboids,
+ GetSchemaPublications(schemaid));
+ }
+
+ list_free(ancestors);
+
+ foreach_oid(puboid, puboids)
+ {
+ Publication *pub = GetPublication(puboid);
+
+ if (pub->pubviaroot)
+ {
+ if (attachrel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot attach foreign table \"%s\" to partition table \"%s\"",
+ attachrelname, relname),
+ errdetail("partition table \"%s\" is published with option publish_via_partition_root",
+ relname)));
+ else
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot attach table \"%s\" with foreign partition to partition table \"%s\"",
+ attachrelname, relname),
+ errdetail("partition table \"%s\" is published with option publish_via_partition_root",
+ relname)));
+ }
+ }
+ }
+
/*
* Check if attachrel has any identity columns or any columns that aren't
* in the parent.
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 48c7d1a861..b1bb864d2f 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -19,6 +19,7 @@
#include "catalog/genbki.h"
#include "catalog/objectaddress.h"
+#include "catalog/pg_class.h"
#include "catalog/pg_publication_d.h" /* IWYU pragma: export */
/* ----------------
@@ -191,4 +192,10 @@ extern Bitmapset *pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols,
extern Bitmapset *pub_form_cols_map(Relation relation,
PublishGencolsType include_gencols_type);
+extern bool check_partrel_has_foreign_table(Form_pg_class relform);
+
+extern void check_foreign_tables_in_schema(Oid schemaid, char *pubname);
+
+extern void check_foreign_tables(char *pubname);
+
#endif /* PG_PUBLICATION_H */
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 4de96c04f9..6a0c245bd6 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -1924,6 +1924,95 @@ DROP PUBLICATION pub1;
DROP PUBLICATION pub2;
DROP TABLE gencols;
RESET client_min_messages;
+-- ======================================================
+-- Test when foreign table is a partition of a partitioned table on which
+-- publication is created
+SET client_min_messages = 'ERROR';
+CREATE FOREIGN DATA WRAPPER test_fdw;
+CREATE SERVER fdw_server FOREIGN DATA WRAPPER test_fdw;
+CREATE SCHEMA sch3;
+CREATE TABLE sch3.tmain(id int) PARTITION BY RANGE(id);
+CREATE TABLE sch3.part1 PARTITION OF sch3.tmain FOR VALUES FROM (0) TO (5);
+CREATE TABLE sch3.part2(id int) PARTITION BY RANGE(id);
+CREATE FOREIGN TABLE sch3.part2_1 PARTITION OF sch3.part2 FOR VALUES FROM (5) TO (10) SERVER fdw_server;
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part2 FOR VALUES FROM (5) TO (10);
+-- Can't create publications with publish_via_partition_root = true, if table
+-- has a foreign partition
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain WITH (publish_via_partition_root);
+ERROR: cannot set parameter "publish_via_partition_root" to true for publication "pub1"
+DETAIL: partition table "tmain" in publication contains a foreign partition
+CREATE PUBLICATION pub1 FOR TABLES IN SCHEMA sch3 WITH (publish_via_partition_root);
+ERROR: cannot set parameter "publish_via_partition_root" to true for publication "pub1"
+DETAIL: partition table "tmain" in publication contains a foreign partition
+CREATE PUBLICATION pub1 FOR ALL TABLES WITH (publish_via_partition_root);
+ERROR: cannot set parameter "publish_via_partition_root" to true for publication "pub1"
+DETAIL: partition table "tmain" in publication contains a foreign partition
+-- Test when a partitioned table with foreign table as a partition is attached
+-- to partitioned table which is already published
+ALTER TABLE sch3.tmain DETACH PARTITION sch3.part2;
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain WITH (publish_via_partition_root);
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part2 FOR VALUES FROM (5) TO (10);
+ERROR: cannot attach table "part2" with foreign partition to partition table "tmain"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+-- Can't create foreign partition of published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch3.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+ERROR: cannot create table foreign partition "part3_1"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+-- Can't attach foreign partition to published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_2(id int) SERVER fdw_server;
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+ERROR: cannot attach foreign table "part3_2" to partition table "tmain"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+CREATE SCHEMA sch4;
+CREATE TABLE sch4.tmain(id int) PARTITION BY RANGE(id);
+-- publication created with FOR TABLES IN SCHEMA
+DROP PUBLICATION pub1;
+CREATE PUBLICATION pub1 FOR TABLES IN SCHEMA sch4 WITH (publish_via_partition_root);
+-- Can't create foreign partition of published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch4.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+ERROR: cannot create table foreign partition "part3_1"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+-- Can't attach foreign partition to published table with
+-- publish_via_partition_root = true
+ALTER TABLE sch4.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+ERROR: cannot attach foreign table "part3_2" to partition table "tmain"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+DROP PUBLICATION pub1;
+-- Test with publish_via_partition_root = false
+-- Foreign partitions are skipped by default
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain;
+CREATE PUBLICATION pub2 FOR TABLES IN SCHEMA sch3;
+CREATE PUBLICATION pub3 FOR ALL TABLES;
+-- Create foreign partition of published table with
+-- publish_via_partition_root = false
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch3.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+-- Attach foreign partition to published table
+-- publish_via_partition_root = false
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+-- Check the published tables
+SELECT pubname, tablename FROM pg_publication_tables WHERE schemaname in ('sch3', 'sch4') ORDER BY pubname, tablename;
+ pubname | tablename
+---------+-----------
+ pub1 | part1
+ pub2 | part1
+ pub3 | part1
+(3 rows)
+
+-- Can't alter publish_via_partition_root to true, if publication already have
+-- foreign partition
+ALTER PUBLICATION pub1 SET (publish_via_partition_root);
+ERROR: cannot set parameter "publish_via_partition_root" to true for publication "pub1"
+DETAIL: partition table "tmain" in publication contains a foreign partition
+DROP PUBLICATION pub1;
+DROP PUBLICATION pub2;
+DROP PUBLICATION pub3;
+DROP SCHEMA sch3 CASCADE;
+DROP SCHEMA sch4 CASCADE;
+DROP SERVER fdw_server;
+DROP FOREIGN DATA WRAPPER test_fdw;
RESET SESSION AUTHORIZATION;
DROP ROLE regress_publication_user, regress_publication_user2;
DROP ROLE regress_publication_user_dummy;
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index 68001de400..49c9d98b66 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -1223,6 +1223,87 @@ DROP PUBLICATION pub2;
DROP TABLE gencols;
RESET client_min_messages;
+-- ======================================================
+
+-- Test when foreign table is a partition of a partitioned table on which
+-- publication is created
+SET client_min_messages = 'ERROR';
+CREATE FOREIGN DATA WRAPPER test_fdw;
+CREATE SERVER fdw_server FOREIGN DATA WRAPPER test_fdw;
+
+CREATE SCHEMA sch3;
+CREATE TABLE sch3.tmain(id int) PARTITION BY RANGE(id);
+CREATE TABLE sch3.part1 PARTITION OF sch3.tmain FOR VALUES FROM (0) TO (5);
+CREATE TABLE sch3.part2(id int) PARTITION BY RANGE(id);
+CREATE FOREIGN TABLE sch3.part2_1 PARTITION OF sch3.part2 FOR VALUES FROM (5) TO (10) SERVER fdw_server;
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part2 FOR VALUES FROM (5) TO (10);
+
+-- Can't create publications with publish_via_partition_root = true, if table
+-- has a foreign partition
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain WITH (publish_via_partition_root);
+CREATE PUBLICATION pub1 FOR TABLES IN SCHEMA sch3 WITH (publish_via_partition_root);
+CREATE PUBLICATION pub1 FOR ALL TABLES WITH (publish_via_partition_root);
+
+-- Test when a partitioned table with foreign table as a partition is attached
+-- to partitioned table which is already published
+ALTER TABLE sch3.tmain DETACH PARTITION sch3.part2;
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain WITH (publish_via_partition_root);
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part2 FOR VALUES FROM (5) TO (10);
+
+-- Can't create foreign partition of published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch3.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+
+-- Can't attach foreign partition to published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_2(id int) SERVER fdw_server;
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+
+CREATE SCHEMA sch4;
+CREATE TABLE sch4.tmain(id int) PARTITION BY RANGE(id);
+
+-- publication created with FOR TABLES IN SCHEMA
+DROP PUBLICATION pub1;
+CREATE PUBLICATION pub1 FOR TABLES IN SCHEMA sch4 WITH (publish_via_partition_root);
+
+-- Can't create foreign partition of published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch4.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+
+-- Can't attach foreign partition to published table with
+-- publish_via_partition_root = true
+ALTER TABLE sch4.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+DROP PUBLICATION pub1;
+
+-- Test with publish_via_partition_root = false
+-- Foreign partitions are skipped by default
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain;
+CREATE PUBLICATION pub2 FOR TABLES IN SCHEMA sch3;
+CREATE PUBLICATION pub3 FOR ALL TABLES;
+
+-- Create foreign partition of published table with
+-- publish_via_partition_root = false
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch3.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+
+-- Attach foreign partition to published table
+-- publish_via_partition_root = false
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+
+-- Check the published tables
+SELECT pubname, tablename FROM pg_publication_tables WHERE schemaname in ('sch3', 'sch4') ORDER BY pubname, tablename;
+
+-- Can't alter publish_via_partition_root to true, if publication already have
+-- foreign partition
+ALTER PUBLICATION pub1 SET (publish_via_partition_root);
+
+DROP PUBLICATION pub1;
+DROP PUBLICATION pub2;
+DROP PUBLICATION pub3;
+DROP SCHEMA sch3 CASCADE;
+DROP SCHEMA sch4 CASCADE;
+DROP SERVER fdw_server;
+DROP FOREIGN DATA WRAPPER test_fdw;
+
RESET SESSION AUTHORIZATION;
DROP ROLE regress_publication_user, regress_publication_user2;
DROP ROLE regress_publication_user_dummy;
--
2.34.1
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Restrict publishing of partitioned table with a foreign table as partition
2025-02-05 06:42 Re: Restrict publishing of partitioned table with a foreign table as partition vignesh C <[email protected]>
2025-02-05 08:44 ` Re: Restrict publishing of partitioned table with a foreign table as partition Álvaro Herrera <[email protected]>
2025-02-11 11:25 ` Re: Restrict publishing of partitioned table with a foreign table as partition Shlok Kyal <[email protected]>
@ 2025-02-13 05:54 ` vignesh C <[email protected]>
2025-02-13 10:20 ` Re: Restrict publishing of partitioned table with a foreign table as partition Shlok Kyal <[email protected]>
0 siblings, 1 reply; 6+ messages in thread
From: vignesh C @ 2025-02-13 05:54 UTC (permalink / raw)
To: Shlok Kyal <[email protected]>; +Cc: Álvaro Herrera <[email protected]>; Sergey Tatarintsev <[email protected]>; [email protected]
On Tue, 11 Feb 2025 at 16:55, Shlok Kyal <[email protected]> wrote:
> I have handled the above cases and added tests for the same.
There is a concurrency issue with the patch:
+check_partrel_has_foreign_table(Form_pg_class relform)
+{
+ bool has_foreign_tbl = false;
+
+ if (relform->relkind == RELKIND_PARTITIONED_TABLE)
+ {
+ List *relids = NIL;
+
+ relids = find_all_inheritors(relform->oid, NoLock, NULL);
+
+ foreach_oid(relid, relids)
+ {
+ Relation rel = table_open(relid,
AccessShareLock);
+
+ if (RelationGetForm(rel)->relkind ==
RELKIND_FOREIGN_TABLE)
+ has_foreign_tbl = true;
+
+ table_close(rel, AccessShareLock);
+
+ if (has_foreign_tbl)
+ break;
+ }
+ }
+
+ return has_foreign_tbl;
+}
In an ideal scenario, the creation of a foreign table should fail if
there is an associated publication, as demonstrated below:
CREATE TABLE t(id int) PARTITION BY RANGE(id);
CREATE TABLE part1 PARTITION OF t FOR VALUES FROM (0) TO (5);
CREATE TABLE part2 PARTITION OF t FOR VALUES FROM (5) TO (15)
PARTITION BY RANGE(id);
CREATE PUBLICATION pub1 FOR TABLE t with (publish_via_partition_root = true);
postgres=# CREATE FOREIGN TABLE part22 PARTITION OF part2 FOR VALUES
FROM (10) TO (15) SERVER fdw;
ERROR: cannot create table foreign partition "part22"
DETAIL: partition table "part2" is published with option
publish_via_partition_root
Consider a scenario where the publication is being created and after
the check_partrel_has_foreign_table execution is done, concurrently
creation of foreign table is executed, then the creation will be
successful.
postgres=# CREATE FOREIGN TABLE part22 PARTITION OF part2 FOR VALUES
FROM (10) TO (15) SERVER fdw;
CREATE FOREIGN TABLE
I felt the problem here is that you have released the lock:
+ if (RelationGetForm(rel)->relkind ==
RELKIND_FOREIGN_TABLE)
+ has_foreign_tbl = true;
+
+ table_close(rel, AccessShareLock);
We should retain the lock to fix this issue:
+ if (RelationGetForm(rel)->relkind ==
RELKIND_FOREIGN_TABLE)
+ has_foreign_tbl = true;
+
+ table_close(rel, NoLock);
Regards,
Vignesh
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Restrict publishing of partitioned table with a foreign table as partition
2025-02-05 06:42 Re: Restrict publishing of partitioned table with a foreign table as partition vignesh C <[email protected]>
2025-02-05 08:44 ` Re: Restrict publishing of partitioned table with a foreign table as partition Álvaro Herrera <[email protected]>
2025-02-11 11:25 ` Re: Restrict publishing of partitioned table with a foreign table as partition Shlok Kyal <[email protected]>
2025-02-13 05:54 ` Re: Restrict publishing of partitioned table with a foreign table as partition vignesh C <[email protected]>
@ 2025-02-13 10:20 ` Shlok Kyal <[email protected]>
0 siblings, 0 replies; 6+ messages in thread
From: Shlok Kyal @ 2025-02-13 10:20 UTC (permalink / raw)
To: vignesh C <[email protected]>; +Cc: Álvaro Herrera <[email protected]>; Sergey Tatarintsev <[email protected]>; [email protected]
On Thu, 13 Feb 2025 at 11:25, vignesh C <[email protected]> wrote:
>
> On Tue, 11 Feb 2025 at 16:55, Shlok Kyal <[email protected]> wrote:
> > I have handled the above cases and added tests for the same.
>
> There is a concurrency issue with the patch:
> +check_partrel_has_foreign_table(Form_pg_class relform)
> +{
> + bool has_foreign_tbl = false;
> +
> + if (relform->relkind == RELKIND_PARTITIONED_TABLE)
> + {
> + List *relids = NIL;
> +
> + relids = find_all_inheritors(relform->oid, NoLock, NULL);
> +
> + foreach_oid(relid, relids)
> + {
> + Relation rel = table_open(relid,
> AccessShareLock);
> +
> + if (RelationGetForm(rel)->relkind ==
> RELKIND_FOREIGN_TABLE)
> + has_foreign_tbl = true;
> +
> + table_close(rel, AccessShareLock);
> +
> + if (has_foreign_tbl)
> + break;
> + }
> + }
> +
> + return has_foreign_tbl;
> +}
>
> In an ideal scenario, the creation of a foreign table should fail if
> there is an associated publication, as demonstrated below:
> CREATE TABLE t(id int) PARTITION BY RANGE(id);
> CREATE TABLE part1 PARTITION OF t FOR VALUES FROM (0) TO (5);
> CREATE TABLE part2 PARTITION OF t FOR VALUES FROM (5) TO (15)
> PARTITION BY RANGE(id);
> CREATE PUBLICATION pub1 FOR TABLE t with (publish_via_partition_root = true);
>
> postgres=# CREATE FOREIGN TABLE part22 PARTITION OF part2 FOR VALUES
> FROM (10) TO (15) SERVER fdw;
> ERROR: cannot create table foreign partition "part22"
> DETAIL: partition table "part2" is published with option
> publish_via_partition_root
>
> Consider a scenario where the publication is being created and after
> the check_partrel_has_foreign_table execution is done, concurrently
> creation of foreign table is executed, then the creation will be
> successful.
> postgres=# CREATE FOREIGN TABLE part22 PARTITION OF part2 FOR VALUES
> FROM (10) TO (15) SERVER fdw;
> CREATE FOREIGN TABLE
>
> I felt the problem here is that you have released the lock:
> + if (RelationGetForm(rel)->relkind ==
> RELKIND_FOREIGN_TABLE)
> + has_foreign_tbl = true;
> +
> + table_close(rel, AccessShareLock);
>
> We should retain the lock to fix this issue:
> + if (RelationGetForm(rel)->relkind ==
> RELKIND_FOREIGN_TABLE)
> + has_foreign_tbl = true;
> +
> + table_close(rel, NoLock);
>
Hi Vignesh,
I have fixed the issue. Attached the updated v6 patch.
Thanks and Regards,
Shlok Kyal
Attachments:
[application/octet-stream] v6-0001-Restrict-publishing-of-partitioned-table-with-for.patch (25.9K, ../../CANhcyEVbFMWnEkShq37SSgW7+X+xOxx9bvc4Jtj6BgAy9jGnqA@mail.gmail.com/2-v6-0001-Restrict-publishing-of-partitioned-table-with-for.patch)
download | inline diff:
From cd882eb524b192e6260d6459c06200229bdd6a7d Mon Sep 17 00:00:00 2001
From: Shlok Kyal <[email protected]>
Date: Tue, 11 Feb 2025 15:32:55 +0530
Subject: [PATCH v6] Restrict publishing of partitioned table with foreign
partition
Logical replication of foreign table is not supported and we throw an
error in this case. But when create a publication on a partitioned
table that has a foreign partition, the initial sync of such table is
successful and we should avoid such cases.
Current Behaviour, when publication is created:
1. with publish_via_partition_root = true
Root table is published and the initial data of foreign partitions
are replicated.
2. with publish_via_partition_root = false and FOR ALL TABLES
All leaf tables except foreign partitions are published.
3. with publish_via_partition_root = false and
FOR TABLE/ FOR TABLES IN SCHEMA
All leaf tables are published. Initial data of foreign partitions are
replicated.
With this patch we have following behaviour:
1. with publish_via_partition_root = true
We throw an error when we try to publish a foreign partititon. Error
is thrown when we try to create a publication on (or add to existing
publication) a partitioned table with foreign partition, when try to
create a foreign partition and when we try to attach foreign table (or
a table with foreign partition) to existing published tables.
2. with publish_via_partition_root = false
We skip publishing foreign partition which are part of the publication.
This is done by avoid adding foreign partitions in pg_subscription_rel
catalog table.
---
src/backend/catalog/pg_publication.c | 181 ++++++++++++++++++++--
src/backend/commands/foreigncmds.c | 47 ++++++
src/backend/commands/publicationcmds.c | 47 ++++++
src/backend/commands/tablecmds.c | 53 +++++++
src/include/catalog/pg_publication.h | 7 +
src/test/regress/expected/publication.out | 89 +++++++++++
src/test/regress/sql/publication.sql | 81 ++++++++++
7 files changed, 488 insertions(+), 17 deletions(-)
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index d6f94db5d9..520fa2f382 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -53,7 +53,7 @@ typedef struct
* error if not.
*/
static void
-check_publication_add_relation(Relation targetrel)
+check_publication_add_relation(Relation targetrel, Publication *pub)
{
/* Must be a regular or partitioned table */
if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION &&
@@ -64,6 +64,19 @@ check_publication_add_relation(Relation targetrel)
RelationGetRelationName(targetrel)),
errdetail_relkind_not_supported(RelationGetForm(targetrel)->relkind)));
+ /*
+ * publish_via_root_partition cannot be true if it is a partitioned table
+ * and has any foreign partition
+ */
+ if (pub->pubviaroot &&
+ check_partrel_has_foreign_table(RelationGetForm(targetrel)))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot set parameter \"%s\" to true for publication \"%s\"",
+ "publish_via_partition_root", pub->name),
+ errdetail("partition table \"%s\" in publication contains a foreign partition",
+ RelationGetRelationName(targetrel))));
+
/* Can't be system table */
if (IsCatalogRelation(targetrel))
ereport(ERROR,
@@ -304,7 +317,7 @@ check_and_fetch_column_list(Publication *pub, Oid relid, MemoryContext mcxt,
/*
* Gets the relations based on the publication partition option for a specified
- * relation.
+ * relation. Foreign tables are not included.
*/
List *
GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
@@ -313,25 +326,21 @@ GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE &&
pub_partopt != PUBLICATION_PART_ROOT)
{
- List *all_parts = find_all_inheritors(relid, NoLock,
- NULL);
+ List *all_parts = find_all_inheritors(relid, NoLock, NULL);
- if (pub_partopt == PUBLICATION_PART_ALL)
- result = list_concat(result, all_parts);
- else if (pub_partopt == PUBLICATION_PART_LEAF)
+ foreach_oid(partOid, all_parts)
{
- ListCell *lc;
+ char relkind = get_rel_relkind(partOid);
- foreach(lc, all_parts)
- {
- Oid partOid = lfirst_oid(lc);
+ if (relkind == RELKIND_FOREIGN_TABLE)
+ continue;
- if (get_rel_relkind(partOid) != RELKIND_PARTITIONED_TABLE)
- result = lappend_oid(result, partOid);
- }
+ if (pub_partopt == PUBLICATION_PART_LEAF &&
+ relkind == RELKIND_PARTITIONED_TABLE)
+ continue;
+
+ result = lappend_oid(result, partOid);
}
- else
- Assert(false);
}
else
result = lappend_oid(result, relid);
@@ -463,7 +472,7 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
RelationGetRelationName(targetrel), pub->name)));
}
- check_publication_add_relation(targetrel);
+ check_publication_add_relation(targetrel, pub);
/* Validate and translate column names into a Bitmapset of attnums. */
attnums = pub_collist_validate(pri->relation, pri->columns);
@@ -703,6 +712,13 @@ publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
check_publication_add_schema(schemaid);
+ /*
+ * If publish_via_partition_root is true, check if schema has any foreign
+ * partition
+ */
+ if (pub->pubviaroot)
+ check_foreign_tables_in_schema(schemaid, pub->name);
+
/* Form a tuple */
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
@@ -1332,3 +1348,134 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
SRF_RETURN_DONE(funcctx);
}
+
+/* Check if a partitioned table has a foreign partition */
+bool
+check_partrel_has_foreign_table(Form_pg_class relform)
+{
+ bool has_foreign_tbl = false;
+
+ if (relform->relkind == RELKIND_PARTITIONED_TABLE)
+ {
+ List *relids = NIL;
+
+ relids = find_all_inheritors(relform->oid, NoLock, NULL);
+
+ foreach_oid(relid, relids)
+ {
+ Relation rel = table_open(relid, AccessShareLock);
+
+ if (RelationGetForm(rel)->relkind == RELKIND_FOREIGN_TABLE)
+ has_foreign_tbl = true;
+
+ table_close(rel, NoLock);
+
+ if (has_foreign_tbl)
+ break;
+ }
+ }
+
+ return has_foreign_tbl;
+}
+
+/*
+ * Check if a schema has a partition table which has a foreign partition.
+ * The partition tables in a schema can have partitions in other schema. We also
+ * need to check if such partitions are foreign partition.
+ */
+void
+check_foreign_tables_in_schema(Oid schemaid, char *pubname)
+{
+ Relation classRel;
+ ScanKeyData key[2];
+ TableScanDesc scan;
+ HeapTuple tuple;
+
+ classRel = table_open(RelationRelationId, AccessShareLock);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_class_relnamespace,
+ BTEqualStrategyNumber, F_OIDEQ,
+ schemaid);
+ ScanKeyInit(&key[1],
+ Anum_pg_class_relkind,
+ BTEqualStrategyNumber, F_CHAREQ,
+ CharGetDatum(RELKIND_PARTITIONED_TABLE));
+
+ scan = table_beginscan_catalog(classRel, 2, key);
+ while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ {
+ Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
+
+ if (check_partrel_has_foreign_table(relForm))
+ {
+ List *ancestors = get_partition_ancestors(relForm->oid);
+ Oid parent_oid = relForm->oid;
+ char *parent_name;
+
+ foreach_oid(ancestor, ancestors)
+ {
+ Oid ancestor_schemaid = get_rel_namespace(ancestor);
+
+ if (ancestor_schemaid == schemaid)
+ parent_oid = ancestor;
+ }
+
+ list_free(ancestors);
+ parent_name = get_rel_name(parent_oid);
+
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot set parameter \"%s\" to true for publication \"%s\"",
+ "publish_via_partition_root", pubname),
+ errdetail("partition table \"%s\" in publication contains a foreign partition",
+ parent_name)));
+ }
+ }
+
+ table_endscan(scan);
+ table_close(classRel, AccessShareLock);
+}
+
+/* Check if any foreign table is a partition table */
+void
+check_foreign_tables(char *pubname)
+{
+ Relation classRel;
+ ScanKeyData key[1];
+ TableScanDesc scan;
+ HeapTuple tuple;
+
+ classRel = table_open(RelationRelationId, AccessShareLock);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_class_relkind,
+ BTEqualStrategyNumber, F_CHAREQ,
+ CharGetDatum(RELKIND_FOREIGN_TABLE));
+
+ scan = table_beginscan_catalog(classRel, 1, key);
+ while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ {
+ Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
+
+ if (relForm->relispartition)
+ {
+ Oid parent_oid;
+ char *parent_name;
+ List *ancestors = get_partition_ancestors(relForm->oid);
+
+ parent_oid = llast_oid(ancestors);
+ parent_name = get_rel_name(parent_oid);
+
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot set parameter \"%s\" to true for publication \"%s\"",
+ "publish_via_partition_root", pubname),
+ errdetail("partition table \"%s\" in publication contains a foreign partition",
+ parent_name)));
+ }
+ }
+
+ table_endscan(scan);
+ table_close(classRel, AccessShareLock);
+}
diff --git a/src/backend/commands/foreigncmds.c b/src/backend/commands/foreigncmds.c
index c14e038d54..f41a24aea3 100644
--- a/src/backend/commands/foreigncmds.c
+++ b/src/backend/commands/foreigncmds.c
@@ -21,6 +21,7 @@
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/objectaccess.h"
+#include "catalog/partition.h"
#include "catalog/pg_foreign_data_wrapper.h"
#include "catalog/pg_foreign_server.h"
#include "catalog/pg_foreign_table.h"
@@ -1423,6 +1424,52 @@ CreateForeignTable(CreateForeignTableStmt *stmt, Oid relid)
ftrel = table_open(ForeignTableRelationId, RowExclusiveLock);
+ /*
+ * Check if it is a foreign partition and the partitioned table is not
+ * published or published with publish_via_partition_root option as false.
+ */
+ if (stmt->base.partbound != NULL)
+ {
+ RangeVar *root = castNode(RangeVar, lfirst(list_head(stmt->base.inhRelations)));
+ Relation rootrel = table_openrv(root, AccessShareLock);
+
+ if (RelationGetForm(rootrel)->relkind == RELKIND_PARTITIONED_TABLE)
+ {
+ Oid schemaid = RelationGetNamespace(rootrel);
+ List *puboids = GetRelationPublications(rootrel->rd_id);
+ List *ancestors;
+
+ puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+ puboids = list_concat_unique_oid(puboids, GetSchemaPublications(schemaid));
+ ancestors = get_partition_ancestors(rootrel->rd_id);
+
+ foreach_oid(ancestor, ancestors)
+ {
+ puboids = list_concat_unique_oid(puboids,
+ GetRelationPublications(ancestor));
+ schemaid = get_rel_namespace(ancestor);
+ puboids = list_concat_unique_oid(puboids,
+ GetSchemaPublications(schemaid));
+ }
+ list_free(ancestors);
+
+ foreach_oid(puboid, puboids)
+ {
+ Publication *pub = GetPublication(puboid);
+
+ if (pub->pubviaroot)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot create table foreign partition \"%s\"",
+ get_rel_name(relid)),
+ errdetail("partition table \"%s\" is published with option publish_via_partition_root",
+ RelationGetRelationName(rootrel))));
+ }
+ }
+
+ table_close(rootrel, AccessShareLock);
+ }
+
/*
* For now the owner cannot be specified on create. Use effective user ID.
*/
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 150a768d16..e71d5408c7 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -876,6 +876,10 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
/* Associate objects with the publication. */
if (stmt->for_all_tables)
{
+ /* Check if any foreign table is a part of partitioned table */
+ if (publish_via_partition_root)
+ check_foreign_tables(stmt->pubname);
+
/* Invalidate relcache so that publication info is rebuilt. */
CacheInvalidateRelcacheAll();
}
@@ -1041,6 +1045,49 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
}
}
+ /*
+ * If publish_via_partition_root is set to true, check if the publication
+ * already have any foreign partition
+ */
+ if (publish_via_partition_root_given && publish_via_partition_root)
+ {
+ List *schemaoids = NIL;
+ List *relids = NIL;
+
+ char *pubname = stmt->pubname;
+
+ if (pubform->puballtables)
+ check_foreign_tables(pubname);
+
+ schemaoids = GetPublicationSchemas(pubform->oid);
+
+ foreach_oid(schemaoid, schemaoids)
+ check_foreign_tables_in_schema(schemaoid, pubname);
+
+ relids = GetPublicationRelations(pubform->oid, PUBLICATION_PART_ROOT);
+
+ foreach_oid(relid, relids)
+ {
+ HeapTuple reltup;
+ Form_pg_class relform;
+
+ reltup = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+ relform = (Form_pg_class) GETSTRUCT(reltup);
+
+ ReleaseSysCache(reltup);
+
+ if (check_partrel_has_foreign_table(relform))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot set parameter \"%s\" to true for publication \"%s\"",
+ "publish_via_partition_root", pubname),
+ errdetail("partition table \"%s\" in publication contains a foreign partition",
+ get_rel_name(relid))));
+ }
+ }
+ }
+
/* Everything ok, form a new tuple. */
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 5823fce934..fccc1082f4 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -19440,6 +19440,59 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot attach temporary relation of another session as partition")));
+ /*
+ * Check if attachrel is a foreign table or a partitioned table with
+ * foreign partition and rel is not part of publication with option
+ * publish_via_partition_root as true.
+ */
+ if (attachrel->rd_rel->relkind == RELKIND_FOREIGN_TABLE ||
+ check_partrel_has_foreign_table(RelationGetForm(attachrel)))
+ {
+ Oid schemaid = RelationGetNamespace(rel);
+ List *puboids = GetRelationPublications(rel->rd_id);
+ List *ancestors;
+ char *relname = get_rel_name(rel->rd_id);
+ char *attachrelname = get_rel_name(attachrel->rd_id);
+
+ puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+ puboids = list_concat_unique_oid(puboids, GetSchemaPublications(schemaid));
+ ancestors = get_partition_ancestors(rel->rd_id);
+
+ foreach_oid(ancestor, ancestors)
+ {
+ puboids = list_concat_unique_oid(puboids,
+ GetRelationPublications(ancestor));
+ schemaid = get_rel_namespace(ancestor);
+ puboids = list_concat_unique_oid(puboids,
+ GetSchemaPublications(schemaid));
+ }
+
+ list_free(ancestors);
+
+ foreach_oid(puboid, puboids)
+ {
+ Publication *pub = GetPublication(puboid);
+
+ if (pub->pubviaroot)
+ {
+ if (attachrel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot attach foreign table \"%s\" to partition table \"%s\"",
+ attachrelname, relname),
+ errdetail("partition table \"%s\" is published with option publish_via_partition_root",
+ relname)));
+ else
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot attach table \"%s\" with foreign partition to partition table \"%s\"",
+ attachrelname, relname),
+ errdetail("partition table \"%s\" is published with option publish_via_partition_root",
+ relname)));
+ }
+ }
+ }
+
/*
* Check if attachrel has any identity columns or any columns that aren't
* in the parent.
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 48c7d1a861..b1bb864d2f 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -19,6 +19,7 @@
#include "catalog/genbki.h"
#include "catalog/objectaddress.h"
+#include "catalog/pg_class.h"
#include "catalog/pg_publication_d.h" /* IWYU pragma: export */
/* ----------------
@@ -191,4 +192,10 @@ extern Bitmapset *pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols,
extern Bitmapset *pub_form_cols_map(Relation relation,
PublishGencolsType include_gencols_type);
+extern bool check_partrel_has_foreign_table(Form_pg_class relform);
+
+extern void check_foreign_tables_in_schema(Oid schemaid, char *pubname);
+
+extern void check_foreign_tables(char *pubname);
+
#endif /* PG_PUBLICATION_H */
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 4de96c04f9..6a0c245bd6 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -1924,6 +1924,95 @@ DROP PUBLICATION pub1;
DROP PUBLICATION pub2;
DROP TABLE gencols;
RESET client_min_messages;
+-- ======================================================
+-- Test when foreign table is a partition of a partitioned table on which
+-- publication is created
+SET client_min_messages = 'ERROR';
+CREATE FOREIGN DATA WRAPPER test_fdw;
+CREATE SERVER fdw_server FOREIGN DATA WRAPPER test_fdw;
+CREATE SCHEMA sch3;
+CREATE TABLE sch3.tmain(id int) PARTITION BY RANGE(id);
+CREATE TABLE sch3.part1 PARTITION OF sch3.tmain FOR VALUES FROM (0) TO (5);
+CREATE TABLE sch3.part2(id int) PARTITION BY RANGE(id);
+CREATE FOREIGN TABLE sch3.part2_1 PARTITION OF sch3.part2 FOR VALUES FROM (5) TO (10) SERVER fdw_server;
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part2 FOR VALUES FROM (5) TO (10);
+-- Can't create publications with publish_via_partition_root = true, if table
+-- has a foreign partition
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain WITH (publish_via_partition_root);
+ERROR: cannot set parameter "publish_via_partition_root" to true for publication "pub1"
+DETAIL: partition table "tmain" in publication contains a foreign partition
+CREATE PUBLICATION pub1 FOR TABLES IN SCHEMA sch3 WITH (publish_via_partition_root);
+ERROR: cannot set parameter "publish_via_partition_root" to true for publication "pub1"
+DETAIL: partition table "tmain" in publication contains a foreign partition
+CREATE PUBLICATION pub1 FOR ALL TABLES WITH (publish_via_partition_root);
+ERROR: cannot set parameter "publish_via_partition_root" to true for publication "pub1"
+DETAIL: partition table "tmain" in publication contains a foreign partition
+-- Test when a partitioned table with foreign table as a partition is attached
+-- to partitioned table which is already published
+ALTER TABLE sch3.tmain DETACH PARTITION sch3.part2;
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain WITH (publish_via_partition_root);
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part2 FOR VALUES FROM (5) TO (10);
+ERROR: cannot attach table "part2" with foreign partition to partition table "tmain"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+-- Can't create foreign partition of published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch3.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+ERROR: cannot create table foreign partition "part3_1"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+-- Can't attach foreign partition to published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_2(id int) SERVER fdw_server;
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+ERROR: cannot attach foreign table "part3_2" to partition table "tmain"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+CREATE SCHEMA sch4;
+CREATE TABLE sch4.tmain(id int) PARTITION BY RANGE(id);
+-- publication created with FOR TABLES IN SCHEMA
+DROP PUBLICATION pub1;
+CREATE PUBLICATION pub1 FOR TABLES IN SCHEMA sch4 WITH (publish_via_partition_root);
+-- Can't create foreign partition of published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch4.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+ERROR: cannot create table foreign partition "part3_1"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+-- Can't attach foreign partition to published table with
+-- publish_via_partition_root = true
+ALTER TABLE sch4.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+ERROR: cannot attach foreign table "part3_2" to partition table "tmain"
+DETAIL: partition table "tmain" is published with option publish_via_partition_root
+DROP PUBLICATION pub1;
+-- Test with publish_via_partition_root = false
+-- Foreign partitions are skipped by default
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain;
+CREATE PUBLICATION pub2 FOR TABLES IN SCHEMA sch3;
+CREATE PUBLICATION pub3 FOR ALL TABLES;
+-- Create foreign partition of published table with
+-- publish_via_partition_root = false
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch3.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+-- Attach foreign partition to published table
+-- publish_via_partition_root = false
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+-- Check the published tables
+SELECT pubname, tablename FROM pg_publication_tables WHERE schemaname in ('sch3', 'sch4') ORDER BY pubname, tablename;
+ pubname | tablename
+---------+-----------
+ pub1 | part1
+ pub2 | part1
+ pub3 | part1
+(3 rows)
+
+-- Can't alter publish_via_partition_root to true, if publication already have
+-- foreign partition
+ALTER PUBLICATION pub1 SET (publish_via_partition_root);
+ERROR: cannot set parameter "publish_via_partition_root" to true for publication "pub1"
+DETAIL: partition table "tmain" in publication contains a foreign partition
+DROP PUBLICATION pub1;
+DROP PUBLICATION pub2;
+DROP PUBLICATION pub3;
+DROP SCHEMA sch3 CASCADE;
+DROP SCHEMA sch4 CASCADE;
+DROP SERVER fdw_server;
+DROP FOREIGN DATA WRAPPER test_fdw;
RESET SESSION AUTHORIZATION;
DROP ROLE regress_publication_user, regress_publication_user2;
DROP ROLE regress_publication_user_dummy;
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index 68001de400..49c9d98b66 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -1223,6 +1223,87 @@ DROP PUBLICATION pub2;
DROP TABLE gencols;
RESET client_min_messages;
+-- ======================================================
+
+-- Test when foreign table is a partition of a partitioned table on which
+-- publication is created
+SET client_min_messages = 'ERROR';
+CREATE FOREIGN DATA WRAPPER test_fdw;
+CREATE SERVER fdw_server FOREIGN DATA WRAPPER test_fdw;
+
+CREATE SCHEMA sch3;
+CREATE TABLE sch3.tmain(id int) PARTITION BY RANGE(id);
+CREATE TABLE sch3.part1 PARTITION OF sch3.tmain FOR VALUES FROM (0) TO (5);
+CREATE TABLE sch3.part2(id int) PARTITION BY RANGE(id);
+CREATE FOREIGN TABLE sch3.part2_1 PARTITION OF sch3.part2 FOR VALUES FROM (5) TO (10) SERVER fdw_server;
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part2 FOR VALUES FROM (5) TO (10);
+
+-- Can't create publications with publish_via_partition_root = true, if table
+-- has a foreign partition
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain WITH (publish_via_partition_root);
+CREATE PUBLICATION pub1 FOR TABLES IN SCHEMA sch3 WITH (publish_via_partition_root);
+CREATE PUBLICATION pub1 FOR ALL TABLES WITH (publish_via_partition_root);
+
+-- Test when a partitioned table with foreign table as a partition is attached
+-- to partitioned table which is already published
+ALTER TABLE sch3.tmain DETACH PARTITION sch3.part2;
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain WITH (publish_via_partition_root);
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part2 FOR VALUES FROM (5) TO (10);
+
+-- Can't create foreign partition of published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch3.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+
+-- Can't attach foreign partition to published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_2(id int) SERVER fdw_server;
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+
+CREATE SCHEMA sch4;
+CREATE TABLE sch4.tmain(id int) PARTITION BY RANGE(id);
+
+-- publication created with FOR TABLES IN SCHEMA
+DROP PUBLICATION pub1;
+CREATE PUBLICATION pub1 FOR TABLES IN SCHEMA sch4 WITH (publish_via_partition_root);
+
+-- Can't create foreign partition of published table with
+-- publish_via_partition_root = true
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch4.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+
+-- Can't attach foreign partition to published table with
+-- publish_via_partition_root = true
+ALTER TABLE sch4.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+DROP PUBLICATION pub1;
+
+-- Test with publish_via_partition_root = false
+-- Foreign partitions are skipped by default
+CREATE PUBLICATION pub1 FOR TABLE sch3.tmain;
+CREATE PUBLICATION pub2 FOR TABLES IN SCHEMA sch3;
+CREATE PUBLICATION pub3 FOR ALL TABLES;
+
+-- Create foreign partition of published table with
+-- publish_via_partition_root = false
+CREATE FOREIGN TABLE sch3.part3_1 PARTITION OF sch3.tmain FOR VALUES FROM (10) TO (15) SERVER fdw_server;
+
+-- Attach foreign partition to published table
+-- publish_via_partition_root = false
+ALTER TABLE sch3.tmain ATTACH PARTITION sch3.part3_2 FOR VALUES FROM (15) TO (20);
+
+-- Check the published tables
+SELECT pubname, tablename FROM pg_publication_tables WHERE schemaname in ('sch3', 'sch4') ORDER BY pubname, tablename;
+
+-- Can't alter publish_via_partition_root to true, if publication already have
+-- foreign partition
+ALTER PUBLICATION pub1 SET (publish_via_partition_root);
+
+DROP PUBLICATION pub1;
+DROP PUBLICATION pub2;
+DROP PUBLICATION pub3;
+DROP SCHEMA sch3 CASCADE;
+DROP SCHEMA sch4 CASCADE;
+DROP SERVER fdw_server;
+DROP FOREIGN DATA WRAPPER test_fdw;
+
RESET SESSION AUTHORIZATION;
DROP ROLE regress_publication_user, regress_publication_user2;
DROP ROLE regress_publication_user_dummy;
--
2.34.1
^ permalink raw reply [nested|flat] 6+ messages in thread
end of thread, other threads:[~2025-02-13 10:20 UTC | newest]
Thread overview: 6+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2023-10-04 05:51 [PATCH v9 4/7] Row pattern recognition patch (executor). Tatsuo Ishii <[email protected]>
2025-02-05 06:42 Re: Restrict publishing of partitioned table with a foreign table as partition vignesh C <[email protected]>
2025-02-05 08:44 ` Re: Restrict publishing of partitioned table with a foreign table as partition Álvaro Herrera <[email protected]>
2025-02-11 11:25 ` Re: Restrict publishing of partitioned table with a foreign table as partition Shlok Kyal <[email protected]>
2025-02-13 05:54 ` Re: Restrict publishing of partitioned table with a foreign table as partition vignesh C <[email protected]>
2025-02-13 10:20 ` Re: Restrict publishing of partitioned table with a foreign table as partition Shlok Kyal <[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