rework-column-row-filtering-v2.patch
text/x-patch
Filename: rework-column-row-filtering-v2.patch
Type: text/x-patch
Part: 0
Patch
Format: unified
Series: patch v2
| File | + | − |
|---|---|---|
| src/backend/replication/logical/proto.c | 31 | 11 |
| src/backend/replication/logical/tablesync.c | 108 | 15 |
| src/backend/replication/pgoutput/pgoutput.c | 288 | 287 |
| src/include/replication/logicalproto.h | 5 | 2 |
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index ff8513e2d29..41c5e3413f6 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -33,7 +33,9 @@ static void logicalrep_write_attrs(StringInfo out, Relation rel,
Bitmapset *columns);
static void logicalrep_write_tuple(StringInfo out, Relation rel,
TupleTableSlot *slot,
- bool binary, Bitmapset *columns);
+ bool binary,
+ Bitmapset *schema_columns,
+ Bitmapset *columns);
static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel);
static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple);
@@ -412,7 +414,8 @@ logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn)
*/
void
logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel,
- TupleTableSlot *newslot, bool binary, Bitmapset *columns)
+ TupleTableSlot *newslot, bool binary,
+ Bitmapset *schema_columns, Bitmapset *columns)
{
pq_sendbyte(out, LOGICAL_REP_MSG_INSERT);
@@ -424,7 +427,8 @@ logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel,
pq_sendint32(out, RelationGetRelid(rel));
pq_sendbyte(out, 'N'); /* new tuple follows */
- logicalrep_write_tuple(out, rel, newslot, binary, columns);
+ logicalrep_write_tuple(out, rel, newslot, binary,
+ schema_columns, columns);
}
/*
@@ -457,7 +461,8 @@ logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup)
void
logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel,
TupleTableSlot *oldslot, TupleTableSlot *newslot,
- bool binary, Bitmapset *columns)
+ bool binary, Bitmapset *schema_columns,
+ Bitmapset *columns)
{
pq_sendbyte(out, LOGICAL_REP_MSG_UPDATE);
@@ -478,11 +483,12 @@ logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel,
pq_sendbyte(out, 'O'); /* old tuple follows */
else
pq_sendbyte(out, 'K'); /* old key follows */
- logicalrep_write_tuple(out, rel, oldslot, binary, NULL);
+ logicalrep_write_tuple(out, rel, oldslot, binary, NULL, NULL);
}
pq_sendbyte(out, 'N'); /* new tuple follows */
- logicalrep_write_tuple(out, rel, newslot, binary, columns);
+ logicalrep_write_tuple(out, rel, newslot, binary,
+ schema_columns, columns);
}
/*
@@ -551,7 +557,7 @@ logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel,
else
pq_sendbyte(out, 'K'); /* old key follows */
- logicalrep_write_tuple(out, rel, oldslot, binary, NULL);
+ logicalrep_write_tuple(out, rel, oldslot, binary, NULL, NULL);
}
/*
@@ -766,7 +772,8 @@ logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp)
*/
static void
logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
- bool binary, Bitmapset *columns)
+ bool binary,
+ Bitmapset *schema_columns, Bitmapset *columns)
{
TupleDesc desc;
Datum *values;
@@ -783,7 +790,7 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
if (att->attisdropped || att->attgenerated)
continue;
- if (!column_in_column_list(att->attnum, columns))
+ if (!column_in_column_list(att->attnum, schema_columns))
continue;
nliveatts++;
@@ -804,10 +811,23 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
if (att->attisdropped || att->attgenerated)
continue;
- if (!column_in_column_list(att->attnum, columns))
+ /*
+ * Columns that are not in schema (union of column lists) should
+ * be skipped entirely.
+ */
+ if (!column_in_column_list(att->attnum, schema_columns))
continue;
- if (isnull[i])
+ /*
+ * Columns not in the column list (derived consindering row filters)
+ * we just send NULL.
+ *
+ * XXX Not sure this is quite correct, though. Imagine you replicate
+ * values for columns (A,B), but it changes the row filter. Can we
+ * send NULL that would overwrite "proper" value replicated earlier?
+ */
+ if (isnull[i] ||
+ !column_in_column_list(att->attnum, columns))
{
pq_sendbyte(out, LOGICALREP_COLUMN_NULL);
continue;
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 49ceec3bdc8..fd547e16f4a 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -126,6 +126,20 @@ static bool FetchTableStates(bool *started_tx);
StringInfo copybuf = NULL;
+/*
+ * Info use to track and evaluate row filters for each publication the relation
+ * is included in, and calculat ethe column list.
+ */
+typedef struct PublicationInfo {
+
+ /* row filter (expression state) */
+ Node *rowfilter;
+
+ /* column list */
+ Bitmapset *columns;
+
+} PublicationInfo;
+
/*
* Exit routine for synchronization worker.
*/
@@ -696,14 +710,14 @@ copy_read_data(void *outbuf, int minread, int maxread)
*/
static void
fetch_remote_table_info(char *nspname, char *relname,
- LogicalRepRelation *lrel, List **qual)
+ LogicalRepRelation *lrel, List **pubinfos)
{
WalRcvExecResult *res;
StringInfoData cmd;
TupleTableSlot *slot;
Oid tableRow[] = {OIDOID, CHAROID, CHAROID};
Oid attrRow[] = {INT2OID, TEXTOID, OIDOID, BOOLOID};
- Oid qualRow[] = {TEXTOID};
+ Oid qualRow[] = {TEXTOID, INT2VECTOROID};
bool isnull;
int natt;
ListCell *lc;
@@ -878,6 +892,7 @@ fetch_remote_table_info(char *nspname, char *relname,
/* We don't know the number of rows coming, so allocate enough space. */
lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
+ lrel->attnums = palloc0(MaxTupleAttributeNumber * sizeof(int16));
lrel->attkeys = NULL;
/*
@@ -905,6 +920,7 @@ fetch_remote_table_info(char *nspname, char *relname,
Assert(!isnull);
lrel->attnames[natt] = rel_colname;
+ lrel->attnums[natt] = attnum;
lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 3, &isnull));
Assert(!isnull);
@@ -943,6 +959,7 @@ fetch_remote_table_info(char *nspname, char *relname,
* 3) one of the subscribed publications is declared as ALL TABLES IN
* SCHEMA that includes this relation
*/
+ *pubinfos = NIL;
if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
{
StringInfoData pub_names;
@@ -965,7 +982,7 @@ fetch_remote_table_info(char *nspname, char *relname,
/* Check for row filters. */
resetStringInfo(&cmd);
appendStringInfo(&cmd,
- "SELECT DISTINCT pg_get_expr(pr.prqual, pr.prrelid)"
+ "SELECT DISTINCT pg_get_expr(pr.prqual, pr.prrelid), pr.prattrs"
" FROM pg_publication p"
" LEFT OUTER JOIN pg_publication_rel pr"
" ON (p.oid = pr.prpubid AND pr.prrelid = %u),"
@@ -976,7 +993,7 @@ fetch_remote_table_info(char *nspname, char *relname,
lrel->remoteid,
pub_names.data);
- res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
+ res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, lengthof(qualRow), qualRow);
if (res->status != WALRCV_OK_TUPLES)
ereport(ERROR,
@@ -993,21 +1010,31 @@ fetch_remote_table_info(char *nspname, char *relname,
slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
{
+ PublicationInfo *pubinfo;
Datum rf = slot_getattr(slot, 1, &isnull);
+ Datum cl;
+
+ pubinfo = (PublicationInfo *) palloc0(sizeof(PublicationInfo));
+
+ if (!isnull)
+ pubinfo->rowfilter = (Node *) makeString(TextDatumGetCString(rf));
+
+ cl = slot_getattr(slot, 2, &isnull);
if (!isnull)
- *qual = lappend(*qual, makeString(TextDatumGetCString(rf)));
- else
{
- /* Ignore filters and cleanup as necessary. */
- if (*qual)
+ int i;
+ int2vector *prattrs = (int2vector *) cl;
+
+ for (i = 0; i < prattrs->dim1; i++)
{
- list_free_deep(*qual);
- *qual = NIL;
+ pubinfo->columns = bms_add_member(pubinfo->columns,
+ prattrs->values[i]);
}
- break;
}
+ *pubinfos = lappend(*pubinfos, pubinfo);
+
ExecClearTuple(slot);
}
ExecDropSingleTupleTableSlot(slot);
@@ -1028,7 +1055,7 @@ copy_table(Relation rel)
{
LogicalRepRelMapEntry *relmapentry;
LogicalRepRelation lrel;
- List *qual = NIL;
+ List *pubinfos = NIL;
WalRcvExecResult *res;
StringInfoData cmd;
CopyFromState cstate;
@@ -1037,7 +1064,7 @@ copy_table(Relation rel)
/* Get the publisher relation info. */
fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
- RelationGetRelationName(rel), &lrel, &qual);
+ RelationGetRelationName(rel), &lrel, &pubinfos);
/* Put the relation into relmap. */
logicalrep_relmap_update(&lrel);
@@ -1050,7 +1077,9 @@ copy_table(Relation rel)
initStringInfo(&cmd);
/* Regular table with no row filter */
- if (lrel.relkind == RELKIND_RELATION && qual == NIL)
+ /* FIXME pubinfos is never NULL now, need to detect absence of row filters
+ * in a different way */
+ if (lrel.relkind == RELKIND_RELATION && pubinfos == NIL)
{
appendStringInfo(&cmd, "COPY %s (",
quote_qualified_identifier(lrel.nspname, lrel.relname));
@@ -1076,11 +1105,55 @@ copy_table(Relation rel)
* (SELECT ...), but we can't just do SELECT * because we need to not
* copy generated columns. For tables with any row filters, build a
* SELECT query with OR'ed row filters for COPY.
+ *
+ * FIXME can be simplified if all subscriptions have the same column
+ * list (or no column list), in which case we don't need the CASE
+ * expressions at all.
*/
appendStringInfoString(&cmd, "COPY (SELECT ");
for (int i = 0; i < lrel.natts; i++)
{
- appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
+ ListCell *lc;
+ StringInfoData qual;
+ bool no_filter = false;
+ bool is_first = true;
+
+ initStringInfo(&qual);
+
+ /* find all row filters for the column, combine them using OR */
+ foreach (lc, pubinfos)
+ {
+ PublicationInfo *pubinfo = (PublicationInfo *) lfirst(lc);
+
+ /* not included in this publication column list */
+ if (pubinfo->columns != NULL &&
+ !bms_is_member(lrel.attnums[i], pubinfo->columns))
+ continue;
+
+ /* covered by this publication, is there an expression? */
+ if (pubinfo->rowfilter == NULL)
+ {
+ no_filter = true;
+ break;
+ }
+
+ if (is_first)
+ {
+ appendStringInfo(&qual, "%s", strVal(pubinfo->rowfilter));
+ is_first = false;
+ }
+ else
+ appendStringInfo(&qual, " OR %s", strVal(pubinfo->rowfilter));
+ }
+
+ if (no_filter)
+ appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
+ else
+ appendStringInfo(&cmd, "(CASE WHEN (%s) THEN %s ELSE NULL END) AS %s",
+ qual.data,
+ quote_identifier(lrel.attnames[i]),
+ quote_identifier(lrel.attnames[i]));
+
if (i < lrel.natts - 1)
appendStringInfoString(&cmd, ", ");
}
@@ -1095,6 +1168,24 @@ copy_table(Relation rel)
appendStringInfoString(&cmd, "ONLY ");
appendStringInfoString(&cmd, quote_qualified_identifier(lrel.nspname, lrel.relname));
+
+ {
+ List *qual = NIL;
+ ListCell *lc;
+
+ foreach (lc, pubinfos)
+ {
+ PublicationInfo *pubinfo = (PublicationInfo *) lfirst(lc);
+
+ if (pubinfo->rowfilter == NULL)
+ {
+ qual = NIL;
+ break;
+ }
+
+ qual = lappend(qual, pubinfo->rowfilter);
+ }
+
/* list of OR'ed filters */
if (qual != NIL)
{
@@ -1110,6 +1201,8 @@ copy_table(Relation rel)
list_free_deep(qual);
}
+ }
+
appendStringInfoString(&cmd, ") TO STDOUT");
}
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index b197bfd565d..85456e6d9f5 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -137,13 +137,13 @@ typedef struct RelationSyncEntry
PublicationActions pubactions;
/*
- * ExprState array for row filter. Different publication actions don't
- * allow multiple expressions to always be combined into one, because
- * updates or deletes restrict the column in expression to be part of the
- * replica identity index whereas inserts do not have this restriction, so
- * there is one ExprState per publication action.
+ * Info about row filters and column lists for each publication this
+ * relation is included in. We keep a list with per-publication info in
+ * order to calculate appropriate column list depending on which row
+ * filter(s) match. Each element contains an ExprState for the filter
+ * and column list.
*/
- ExprState *exprstate[NUM_ROWFILTER_PUBACTIONS];
+ dlist_head pubinfos; /* one per publication */
EState *estate; /* executor state used for row filter */
TupleTableSlot *new_slot; /* slot for storing new tuple */
TupleTableSlot *old_slot; /* slot for storing old tuple */
@@ -208,6 +208,29 @@ typedef struct PGOutputTxnData
* been sent */
} PGOutputTxnData;
+/*
+ * Info use to track and evaluate row filters for each publication the relation
+ * is included in, and calculat ethe column list.
+ */
+typedef struct PublicationInfo {
+
+ /* doubly-linked list */
+ dlist_node node;
+
+ /* publication OID (XXX not really needed) */
+ Oid oid;
+
+ /* row filter (expression state) */
+ ExprState *rowfilter;
+
+ /* column list */
+ Bitmapset *columns;
+
+ /* actions published by the publication */
+ PublicationActions pubactions;
+
+} PublicationInfo;
+
/* Map used to remember which relation schemas we sent. */
static HTAB *RelationSyncCache = NULL;
@@ -235,12 +258,8 @@ static bool pgoutput_row_filter_exec_expr(ExprState *state,
static bool pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
TupleTableSlot **new_slot_ptr,
RelationSyncEntry *entry,
- ReorderBufferChangeType *action);
-
-/* column list routines */
-static void pgoutput_column_list_init(PGOutputData *data,
- List *publications,
- RelationSyncEntry *entry);
+ ReorderBufferChangeType *action,
+ Bitmapset **column_list);
/*
* Specify output plugin callbacks
@@ -822,18 +841,24 @@ pgoutput_ensure_entry_cxt(PGOutputData *data, RelationSyncEntry *entry)
}
/*
- * Initialize the row filter.
+ * Initialize the row filter and column list.
+ *
+ * Prepare information (ExprState, etc) used to evaluate per-publication row
+ * filters, and column lists.
+ *
+ * We also calculate a "total" column list as a union of all per-publication
+ * column lists, irrespectedly of row filters. This is used to send schema
+ * for the relsync entry, etc.
*/
static void
pgoutput_row_filter_init(PGOutputData *data, List *publications,
RelationSyncEntry *entry)
{
ListCell *lc;
- List *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */
- bool no_filter[] = {false, false, false}; /* One per pubaction */
MemoryContext oldctx;
- int idx;
- bool has_filter = true;
+ bool all_columns = false;
+
+ dlist_init(&entry->pubinfos);
/*
* Find if there are any row filters for this relation. If there are, then
@@ -855,7 +880,12 @@ pgoutput_row_filter_init(PGOutputData *data, List *publications,
Publication *pub = lfirst(lc);
HeapTuple rftuple = NULL;
Datum rfdatum = 0;
+ Datum cfdatum = 0;
bool pub_no_filter = false;
+ bool pub_no_list = false;
+ Relation relation = RelationIdGetRelation(entry->publish_as_relid);
+
+ PublicationInfo *pubinfo = NULL;
if (pub->alltables)
{
@@ -865,6 +895,7 @@ pgoutput_row_filter_init(PGOutputData *data, List *publications,
* publications it does).
*/
pub_no_filter = true;
+ pub_no_list = true;
}
else
{
@@ -881,191 +912,75 @@ pgoutput_row_filter_init(PGOutputData *data, List *publications,
rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
Anum_pg_publication_rel_prqual,
&pub_no_filter);
+
+ /*
+ * Lookup the column list attribute.
+ *
+ * Null indicates no list.
+ */
+ cfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
+ Anum_pg_publication_rel_prattrs,
+ &pub_no_list);
}
else
{
pub_no_filter = true;
+ pub_no_list = true;
}
}
- if (pub_no_filter)
- {
- if (rftuple)
- ReleaseSysCache(rftuple);
-
- no_filter[PUBACTION_INSERT] |= pub->pubactions.pubinsert;
- no_filter[PUBACTION_UPDATE] |= pub->pubactions.pubupdate;
- no_filter[PUBACTION_DELETE] |= pub->pubactions.pubdelete;
-
- /*
- * Quick exit if all the DML actions are publicized via this
- * publication.
- */
- if (no_filter[PUBACTION_INSERT] &&
- no_filter[PUBACTION_UPDATE] &&
- no_filter[PUBACTION_DELETE])
- {
- has_filter = false;
- break;
- }
-
- /* No additional work for this publication. Next one. */
- continue;
- }
-
- /* Form the per pubaction row filter lists. */
- if (pub->pubactions.pubinsert && !no_filter[PUBACTION_INSERT])
- rfnodes[PUBACTION_INSERT] = lappend(rfnodes[PUBACTION_INSERT],
- TextDatumGetCString(rfdatum));
- if (pub->pubactions.pubupdate && !no_filter[PUBACTION_UPDATE])
- rfnodes[PUBACTION_UPDATE] = lappend(rfnodes[PUBACTION_UPDATE],
- TextDatumGetCString(rfdatum));
- if (pub->pubactions.pubdelete && !no_filter[PUBACTION_DELETE])
- rfnodes[PUBACTION_DELETE] = lappend(rfnodes[PUBACTION_DELETE],
- TextDatumGetCString(rfdatum));
-
- ReleaseSysCache(rftuple);
- } /* loop all subscribed publications */
-
- /* Clean the row filter */
- for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
- {
- if (no_filter[idx])
- {
- list_free_deep(rfnodes[idx]);
- rfnodes[idx] = NIL;
- }
- }
-
- if (has_filter)
- {
- Relation relation = RelationIdGetRelation(entry->publish_as_relid);
-
pgoutput_ensure_entry_cxt(data, entry);
- /*
- * Now all the filters for all pubactions are known. Combine them when
- * their pubactions are the same.
- */
oldctx = MemoryContextSwitchTo(entry->entry_cxt);
- entry->estate = create_estate_for_relation(relation);
- for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
- {
- List *filters = NIL;
- Expr *rfnode;
- if (rfnodes[idx] == NIL)
- continue;
-
- foreach(lc, rfnodes[idx])
- filters = lappend(filters, stringToNode((char *) lfirst(lc)));
-
- /* combine the row filter and cache the ExprState */
- rfnode = make_orclause(filters);
- entry->exprstate[idx] = ExecPrepareExpr(rfnode, entry->estate);
- } /* for each pubaction */
- MemoryContextSwitchTo(oldctx);
-
- RelationClose(relation);
- }
-}
+ pubinfo = (PublicationInfo *) palloc0(sizeof(PublicationInfo));
-/*
- * Initialize the column list.
- */
-static void
-pgoutput_column_list_init(PGOutputData *data, List *publications,
- RelationSyncEntry *entry)
-{
- ListCell *lc;
+ pubinfo->oid = pub->oid;
+ pubinfo->pubactions = pub->pubactions;
- /*
- * Find if there are any column lists for this relation. If there are,
- * build a bitmap merging all the column lists.
- *
- * All the given publication-table mappings must be checked.
- *
- * Multiple publications might have multiple column lists for this relation.
- *
- * FOR ALL TABLES and FOR ALL TABLES IN SCHEMA implies "don't use column
- * list" so it takes precedence.
- */
- foreach(lc, publications)
- {
- Publication *pub = lfirst(lc);
- HeapTuple cftuple = NULL;
- Datum cfdatum = 0;
+ if (!pub_no_filter)
+ {
+ entry->estate = create_estate_for_relation(relation);
- /*
- * Assume there's no column list. Only if we find pg_publication_rel
- * entry with a column list we'll switch it to false.
- */
- bool pub_no_list = true;
+ pubinfo->rowfilter
+ = ExecPrepareExpr(stringToNode(TextDatumGetCString(rfdatum)),
+ entry->estate);
+ }
/*
- * If the publication is FOR ALL TABLES then it is treated the same as if
- * there are no column lists (even if other publications have a list).
+ * Build the column list bitmap in the per-entry context.
+ *
+ * We need to merge column lists from all publications, so we
+ * update the same bitmapset. If the column list is null, we
+ * interpret it as replicating all columns.
*/
- if (!pub->alltables)
+ pubinfo->columns = NULL;
+ if (!pub_no_list) /* when not null */
{
- /*
- * Check for the presence of a column list in this publication.
- *
- * Note: If we find no pg_publication_rel row, it's a publication
- * defined for a whole schema, so it can't have a column list, just
- * like a FOR ALL TABLES publication.
- */
- cftuple = SearchSysCache2(PUBLICATIONRELMAP,
- ObjectIdGetDatum(entry->publish_as_relid),
- ObjectIdGetDatum(pub->oid));
-
- if (HeapTupleIsValid(cftuple))
- {
- /*
- * Lookup the column list attribute.
- *
- * Note: We update the pub_no_list value directly, because if
- * the value is NULL, we have no list (and vice versa).
- */
- cfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, cftuple,
- Anum_pg_publication_rel_prattrs,
- &pub_no_list);
-
- /*
- * Build the column list bitmap in the per-entry context.
- *
- * We need to merge column lists from all publications, so we
- * update the same bitmapset. If the column list is null, we
- * interpret it as replicating all columns.
- */
- if (!pub_no_list) /* when not null */
- {
- pgoutput_ensure_entry_cxt(data, entry);
+ pubinfo->columns = pub_collist_to_bitmapset(pubinfo->columns,
+ cfdatum,
+ entry->entry_cxt);
- entry->columns = pub_collist_to_bitmapset(entry->columns,
- cfdatum,
- entry->entry_cxt);
- }
- }
+ entry->columns = pub_collist_to_bitmapset(entry->columns,
+ cfdatum,
+ entry->entry_cxt);
}
+ else
+ all_columns = true;
- /*
- * Found a publication with no column list, so we're done. But first
- * discard column list we might have from preceding publications.
- */
- if (pub_no_list)
- {
- if (cftuple)
- ReleaseSysCache(cftuple);
+ if (HeapTupleIsValid(rftuple))
+ ReleaseSysCache(rftuple);
- bms_free(entry->columns);
- entry->columns = NULL;
+ MemoryContextSwitchTo(oldctx);
+ RelationClose(relation);
- break;
- }
+ dlist_push_tail(&entry->pubinfos, &pubinfo->node);
- ReleaseSysCache(cftuple);
} /* loop all subscribed publications */
+
+ /* any of the publications replicates all columns */
+ if (all_columns)
+ entry->columns = NULL;
}
/*
@@ -1115,7 +1030,8 @@ init_tuple_slot(PGOutputData *data, Relation relation,
}
/*
- * Change is checked against the row filter if any.
+ * Change is checked against the row filter if any, and calculate the column
+ * list applicable to the operation (with respect to matching row filters).
*
* Returns true if the change is to be replicated, else false.
*
@@ -1136,6 +1052,8 @@ init_tuple_slot(PGOutputData *data, Relation relation,
*
* The new action is updated in the action parameter.
*
+ * The calculated column list is returned in the column_list parameter.
+ *
* The new slot could be updated when transforming the UPDATE into INSERT,
* because the original new tuple might not have column values from the replica
* identity.
@@ -1167,17 +1085,21 @@ init_tuple_slot(PGOutputData *data, Relation relation,
static bool
pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
TupleTableSlot **new_slot_ptr, RelationSyncEntry *entry,
- ReorderBufferChangeType *action)
+ ReorderBufferChangeType *action, Bitmapset **column_list)
{
TupleDesc desc;
int i;
- bool old_matched,
- new_matched,
+ bool old_matched_any = false,
+ new_matched_any = false,
result;
- TupleTableSlot *tmp_new_slot;
+ TupleTableSlot *tmp_new_slot = NULL;
TupleTableSlot *new_slot = *new_slot_ptr;
- ExprContext *ecxt;
- ExprState *filter_exprstate;
+ dlist_iter iter;
+ bool matching = false;
+
+ /* Column list calculated from publications matching the row filter. */
+ Bitmapset *columns = NULL;
+ bool all_columns = false;
/*
* We need this map to avoid relying on ReorderBufferChangeType enums
@@ -1195,115 +1117,191 @@ pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
Assert(new_slot || old_slot);
- /* Get the corresponding row filter */
- filter_exprstate = entry->exprstate[map_changetype_pubaction[*action]];
+ dlist_foreach(iter, &entry->pubinfos)
+ {
+ ExprContext *ecxt;
+ bool old_matched,
+ new_matched;
- /* Bail out if there is no row filter */
- if (!filter_exprstate)
- return true;
+ PublicationInfo *pubinfo
+ = dlist_container(PublicationInfo, node, iter.cur);
- elog(DEBUG3, "table \"%s.%s\" has row filter",
- get_namespace_name(RelationGetNamespace(relation)),
- RelationGetRelationName(relation));
+ /* ignore publications not replicating this action */
+ if ((*action == REORDER_BUFFER_CHANGE_INSERT) &&
+ (!pubinfo->pubactions.pubinsert))
+ continue;
+ else if ((*action == REORDER_BUFFER_CHANGE_UPDATE) &&
+ (!pubinfo->pubactions.pubupdate))
+ continue;
+ else if ((*action == REORDER_BUFFER_CHANGE_DELETE) &&
+ (!pubinfo->pubactions.pubdelete))
+ continue;
- ResetPerTupleExprContext(entry->estate);
+ if (!pubinfo->rowfilter)
+ {
+ matching = true;
- ecxt = GetPerTupleExprContext(entry->estate);
+ /*
+ * Update/merge the column list.
+ *
+ * If the publication has no column list, we interpret it as a list
+ * with all columns. Otherwise we just add it to the bitmap.
+ *
+ * FIXME This is repeated in three places. Maybe refactor?
+ */
+ if (!pubinfo->columns)
+ {
+ all_columns = true;
+ bms_free(columns);
+ columns = NULL;
+ }
+ else if (!all_columns)
+ columns = bms_union(columns, pubinfo->columns);
- /*
- * For the following occasions where there is only one tuple, we can
- * evaluate the row filter for that tuple and return.
- *
- * For inserts, we only have the new tuple.
- *
- * For updates, we can have only a new tuple when none of the replica
- * identity columns changed and none of those columns have external data
- * but we still need to evaluate the row filter for the new tuple as the
- * existing values of those columns might not match the filter. Also, users
- * can use constant expressions in the row filter, so we anyway need to
- * evaluate it for the new tuple.
- *
- * For deletes, we only have the old tuple.
- */
- if (!new_slot || !old_slot)
- {
- ecxt->ecxt_scantuple = new_slot ? new_slot : old_slot;
- result = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
+ continue;
+ }
- return result;
- }
+ elog(DEBUG3, "table \"%s.%s\" has row filter",
+ get_namespace_name(RelationGetNamespace(relation)),
+ RelationGetRelationName(relation));
- /*
- * Both the old and new tuples must be valid only for updates and need to
- * be checked against the row filter.
- */
- Assert(map_changetype_pubaction[*action] == PUBACTION_UPDATE);
+ ResetPerTupleExprContext(entry->estate);
+
+ ecxt = GetPerTupleExprContext(entry->estate);
- slot_getallattrs(new_slot);
- slot_getallattrs(old_slot);
+ /*
+ * For the following occasions where there is only one tuple, we can
+ * evaluate the row filter for that tuple and return.
+ *
+ * For inserts, we only have the new tuple.
+ *
+ * For updates, we can have only a new tuple when none of the replica
+ * identity columns changed and none of those columns have external data
+ * but we still need to evaluate the row filter for the new tuple as the
+ * existing values of those columns might not match the filter. Also, users
+ * can use constant expressions in the row filter, so we anyway need to
+ * evaluate it for the new tuple.
+ *
+ * For deletes, we only have the old tuple.
+ */
+ if (!new_slot || !old_slot)
+ {
+ ecxt->ecxt_scantuple = new_slot ? new_slot : old_slot;
+ result = pgoutput_row_filter_exec_expr(pubinfo->rowfilter, ecxt);
- tmp_new_slot = NULL;
- desc = RelationGetDescr(relation);
+ matching |= result;
- /*
- * The new tuple might not have all the replica identity columns, in which
- * case it needs to be copied over from the old tuple.
- */
- for (i = 0; i < desc->natts; i++)
- {
- Form_pg_attribute att = TupleDescAttr(desc, i);
+ /*
+ * FIXME refactor to reuse this code in multiple places
+ *
+ * XXX Should this only update column list when using new slot?
+ * If evaluationg old slot, that's delete, no?
+ */
+ if (result)
+ {
+ if (!pubinfo->columns)
+ {
+ all_columns = true;
+ bms_free(columns);
+ columns = NULL;
+ }
+ else if (!all_columns)
+ {
+ columns = bms_union(columns, pubinfo->columns);
+ }
+ }
+
+ continue;
+ }
/*
- * if the column in the new tuple or old tuple is null, nothing to do
+ * Both the old and new tuples must be valid only for updates and need to
+ * be checked against the row filter.
*/
- if (new_slot->tts_isnull[i] || old_slot->tts_isnull[i])
- continue;
+ Assert(map_changetype_pubaction[*action] == PUBACTION_UPDATE);
+
+ slot_getallattrs(new_slot);
+ slot_getallattrs(old_slot);
+
+ tmp_new_slot = NULL;
+ desc = RelationGetDescr(relation);
/*
- * Unchanged toasted replica identity columns are only logged in the
- * old tuple. Copy this over to the new tuple. The changed (or WAL
- * Logged) toast values are always assembled in memory and set as
- * VARTAG_INDIRECT. See ReorderBufferToastReplace.
+ * The new tuple might not have all the replica identity columns, in which
+ * case it needs to be copied over from the old tuple.
*/
- if (att->attlen == -1 &&
- VARATT_IS_EXTERNAL_ONDISK(new_slot->tts_values[i]) &&
- !VARATT_IS_EXTERNAL_ONDISK(old_slot->tts_values[i]))
+ for (i = 0; i < desc->natts; i++)
{
- if (!tmp_new_slot)
+ Form_pg_attribute att = TupleDescAttr(desc, i);
+
+ /*
+ * if the column in the new tuple or old tuple is null, nothing to do
+ */
+ if (new_slot->tts_isnull[i] || old_slot->tts_isnull[i])
+ continue;
+
+ /*
+ * Unchanged toasted replica identity columns are only logged in the
+ * old tuple. Copy this over to the new tuple. The changed (or WAL
+ * Logged) toast values are always assembled in memory and set as
+ * VARTAG_INDIRECT. See ReorderBufferToastReplace.
+ */
+ if (att->attlen == -1 &&
+ VARATT_IS_EXTERNAL_ONDISK(new_slot->tts_values[i]) &&
+ !VARATT_IS_EXTERNAL_ONDISK(old_slot->tts_values[i]))
{
- tmp_new_slot = MakeSingleTupleTableSlot(desc, &TTSOpsVirtual);
- ExecClearTuple(tmp_new_slot);
+ if (!tmp_new_slot)
+ {
+ tmp_new_slot = MakeSingleTupleTableSlot(desc, &TTSOpsVirtual);
+ ExecClearTuple(tmp_new_slot);
- memcpy(tmp_new_slot->tts_values, new_slot->tts_values,
- desc->natts * sizeof(Datum));
- memcpy(tmp_new_slot->tts_isnull, new_slot->tts_isnull,
- desc->natts * sizeof(bool));
- }
+ memcpy(tmp_new_slot->tts_values, new_slot->tts_values,
+ desc->natts * sizeof(Datum));
+ memcpy(tmp_new_slot->tts_isnull, new_slot->tts_isnull,
+ desc->natts * sizeof(bool));
+ }
- tmp_new_slot->tts_values[i] = old_slot->tts_values[i];
- tmp_new_slot->tts_isnull[i] = old_slot->tts_isnull[i];
+ tmp_new_slot->tts_values[i] = old_slot->tts_values[i];
+ tmp_new_slot->tts_isnull[i] = old_slot->tts_isnull[i];
+ }
}
- }
- ecxt->ecxt_scantuple = old_slot;
- old_matched = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
+ ecxt->ecxt_scantuple = old_slot;
- if (tmp_new_slot)
- {
- ExecStoreVirtualTuple(tmp_new_slot);
- ecxt->ecxt_scantuple = tmp_new_slot;
- }
- else
- ecxt->ecxt_scantuple = new_slot;
+ old_matched = pgoutput_row_filter_exec_expr(pubinfo->rowfilter, ecxt);
+ old_matched_any |= old_matched;
+
+ if (tmp_new_slot)
+ {
+ ExecStoreVirtualTuple(tmp_new_slot);
+ ecxt->ecxt_scantuple = tmp_new_slot;
+ }
+ else
+ ecxt->ecxt_scantuple = new_slot;
- new_matched = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
+ new_matched = pgoutput_row_filter_exec_expr(pubinfo->rowfilter, ecxt);
+ new_matched_any |= new_matched;
- /*
- * Case 1: if both tuples don't match the row filter, bailout. Send
- * nothing.
- */
- if (!old_matched && !new_matched)
- return false;
+ /*
+ * Case 1: if both tuples don't match the row filter, bailout. Send
+ * nothing.
+ */
+ if (!old_matched && !new_matched)
+ continue; /* continue with the next row filter */
+
+ /*
+ * Case 4: if both tuples match the row filter, transformation isn't
+ * required. (*action is default UPDATE).
+ */
+ if (!pubinfo->columns)
+ {
+ all_columns = true;
+ bms_free(columns);
+ columns = NULL;
+ }
+ else if (!all_columns && new_matched)
+ columns = bms_union(columns, pubinfo->columns);
+ }
/*
* Case 2: if the old tuple doesn't satisfy the row filter but the new
@@ -1314,9 +1312,10 @@ pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
* while inserting the tuple in the downstream node, we have all the
* required column values.
*/
- if (!old_matched && new_matched)
+ if (!old_matched_any && new_matched_any)
{
*action = REORDER_BUFFER_CHANGE_INSERT;
+ matching = true;
if (tmp_new_slot)
*new_slot_ptr = tmp_new_slot;
@@ -1329,15 +1328,18 @@ pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
* This transformation does not require another tuple. The Old tuple will
* be used for DELETE.
*/
- else if (old_matched && !new_matched)
+ else if (old_matched_any && !new_matched_any)
+ {
*action = REORDER_BUFFER_CHANGE_DELETE;
+ matching = true;
+ }
+ else if (old_matched_any && new_matched_any)
+ matching = true;
- /*
- * Case 4: if both tuples match the row filter, transformation isn't
- * required. (*action is default UPDATE).
- */
+ if (column_list)
+ *column_list = columns;
- return true;
+ return matching;
}
/*
@@ -1359,6 +1361,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
ReorderBufferChangeType action = change->action;
TupleTableSlot *old_slot = NULL;
TupleTableSlot *new_slot = NULL;
+ Bitmapset *columns = NULL;
if (!is_publishable_relation(relation))
return;
@@ -1423,7 +1426,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* Check row filter */
if (!pgoutput_row_filter(targetrel, NULL, &new_slot, relentry,
- &action))
+ &action, &columns))
break;
/*
@@ -1444,7 +1447,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_insert(ctx->out, xid, targetrel, new_slot,
- data->binary, relentry->columns);
+ data->binary,
+ relentry->columns, columns);
OutputPluginWrite(ctx, true);
break;
case REORDER_BUFFER_CHANGE_UPDATE:
@@ -1483,7 +1487,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* Check row filter */
if (!pgoutput_row_filter(targetrel, old_slot, &new_slot,
- relentry, &action))
+ relentry, &action, &columns))
break;
/* Send BEGIN if we haven't yet */
@@ -1503,12 +1507,12 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
case REORDER_BUFFER_CHANGE_INSERT:
logicalrep_write_insert(ctx->out, xid, targetrel,
new_slot, data->binary,
- relentry->columns);
+ relentry->columns, columns);
break;
case REORDER_BUFFER_CHANGE_UPDATE:
logicalrep_write_update(ctx->out, xid, targetrel,
old_slot, new_slot, data->binary,
- relentry->columns);
+ relentry->columns, columns);
break;
case REORDER_BUFFER_CHANGE_DELETE:
logicalrep_write_delete(ctx->out, xid, targetrel,
@@ -1547,7 +1551,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* Check row filter */
if (!pgoutput_row_filter(targetrel, old_slot, &new_slot,
- relentry, &action))
+ relentry, &action, NULL))
break;
/* Send BEGIN if we haven't yet */
@@ -1977,7 +1981,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
entry->new_slot = NULL;
entry->old_slot = NULL;
- memset(entry->exprstate, 0, sizeof(entry->exprstate));
+ dlist_init(&entry->pubinfos);
entry->entry_cxt = NULL;
entry->publish_as_relid = InvalidOid;
entry->columns = NULL;
@@ -2056,7 +2060,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->entry_cxt = NULL;
entry->estate = NULL;
- memset(entry->exprstate, 0, sizeof(entry->exprstate));
+ dlist_init(&entry->pubinfos);
/*
* Build publication cache. We can't use one provided by relcache as
@@ -2192,11 +2196,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
/* Initialize the tuple slot and map */
init_tuple_slot(data, relation, entry);
- /* Initialize the row filter */
+ /* Initialize the row filter and column list info */
pgoutput_row_filter_init(data, rel_publications, entry);
-
- /* Initialize the column list */
- pgoutput_column_list_init(data, rel_publications, entry);
}
list_free(pubids);
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index a771ab8ff33..1a3f4f34bf4 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -104,6 +104,7 @@ typedef struct LogicalRepRelation
char *relname; /* relation name */
int natts; /* number of columns */
char **attnames; /* column names */
+ int16 *attnums; /* column attnums */
Oid *atttyps; /* column types */
char replident; /* replica identity */
char relkind; /* remote relation kind */
@@ -209,12 +210,14 @@ extern char *logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn);
extern void logicalrep_write_insert(StringInfo out, TransactionId xid,
Relation rel,
TupleTableSlot *newslot,
- bool binary, Bitmapset *columns);
+ bool binary, Bitmapset *schema_columns,
+ Bitmapset *columns);
extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup);
extern void logicalrep_write_update(StringInfo out, TransactionId xid,
Relation rel,
TupleTableSlot *oldslot,
- TupleTableSlot *newslot, bool binary, Bitmapset *columns);
+ TupleTableSlot *newslot, bool binary,
+ Bitmapset *schema_columns, Bitmapset *columns);
extern LogicalRepRelId logicalrep_read_update(StringInfo in,
bool *has_oldtuple, LogicalRepTupleData *oldtup,
LogicalRepTupleData *newtup);