From 4810cb7a3926ce50ca833b656bf812ea2f51cb01 Mon Sep 17 00:00:00 2001 From: Ajin Cherian Date: Wed, 22 Dec 2021 05:56:03 -0500 Subject: [PATCH v53 2/3] Row filter updates based on old/new tuples. When applying row filter on updates, check both old_tuple and new_tuple to decide how an update needs to be transformed. UPDATE old-row (match) new-row (no match) -> DELETE old-row (no match) new row (match) -> INSERT old-row (match) new row (match) -> UPDATE old-row (no match) new-row (no match) -> (drop change) Also tuples that have been deformed will be cached in slots to avoid multiple deforming of tuples. Author: Ajin Cherian --- src/backend/replication/logical/proto.c | 37 ++-- src/backend/replication/pgoutput/pgoutput.c | 276 +++++++++++++++++++++++----- src/include/replication/logicalproto.h | 7 +- src/include/replication/reorderbuffer.h | 6 +- src/test/subscription/t/027_row_filter.pl | 4 +- src/tools/pgindent/typedefs.list | 1 + 6 files changed, 268 insertions(+), 63 deletions(-) diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 9f5bf4b..4d0c387 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -15,6 +15,7 @@ #include "access/sysattr.h" #include "catalog/pg_namespace.h" #include "catalog/pg_type.h" +#include "executor/executor.h" #include "libpq/pqformat.h" #include "replication/logicalproto.h" #include "utils/lsyscache.h" @@ -31,8 +32,8 @@ static void logicalrep_write_attrs(StringInfo out, Relation rel); static void logicalrep_write_tuple(StringInfo out, Relation rel, - HeapTuple tuple, bool binary); - + HeapTuple tuple, TupleTableSlot *slot, + bool binary); static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel); static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple); @@ -410,7 +411,7 @@ 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, newtuple, binary); + logicalrep_write_tuple(out, rel, newtuple, NULL, binary); } /* @@ -442,7 +443,8 @@ logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup) */ void logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, - HeapTuple oldtuple, HeapTuple newtuple, bool binary) + HeapTuple oldtuple, HeapTuple newtuple, TupleTableSlot *oldslot, + TupleTableSlot *newslot, bool binary) { pq_sendbyte(out, LOGICAL_REP_MSG_UPDATE); @@ -463,11 +465,11 @@ 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, oldtuple, binary); + logicalrep_write_tuple(out, rel, oldtuple, oldslot, binary); } pq_sendbyte(out, 'N'); /* new tuple follows */ - logicalrep_write_tuple(out, rel, newtuple, binary); + logicalrep_write_tuple(out, rel, newtuple, newslot, binary); } /* @@ -536,7 +538,7 @@ logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel, else pq_sendbyte(out, 'K'); /* old key follows */ - logicalrep_write_tuple(out, rel, oldtuple, binary); + logicalrep_write_tuple(out, rel, oldtuple, NULL, binary); } /* @@ -749,13 +751,16 @@ logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp) * Write a tuple to the outputstream, in the most efficient format possible. */ static void -logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binary) +logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, TupleTableSlot *slot, +bool binary) { TupleDesc desc; - Datum values[MaxTupleAttributeNumber]; - bool isnull[MaxTupleAttributeNumber]; + Datum *values; + bool *isnull; int i; uint16 nliveatts = 0; + Datum attr_values[MaxTupleAttributeNumber]; + bool attr_isnull[MaxTupleAttributeNumber]; desc = RelationGetDescr(rel); @@ -771,7 +776,17 @@ logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binar enlargeStringInfo(out, tuple->t_len + nliveatts * (1 + 4)); - heap_deform_tuple(tuple, desc, values, isnull); + if (TupIsNull(slot)) + { + values = attr_values; + isnull = attr_isnull; + heap_deform_tuple(tuple, desc, values, isnull); + } + else + { + values = slot->tts_values; + isnull = slot->tts_isnull; + } /* Write the values */ for (i = 0; i < desc->natts; i++) diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index d9bb0f6..0e4c736 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -13,6 +13,7 @@ #include "postgres.h" #include "access/tupconvert.h" +#include "access/xact.h" #include "catalog/partition.h" #include "catalog/pg_publication.h" #include "catalog/pg_publication_rel.h" @@ -25,6 +26,7 @@ #include "parser/parse_coerce.h" #include "replication/logical.h" #include "replication/logicalproto.h" +#include "replication/logicalrelation.h" #include "replication/origin.h" #include "replication/pgoutput.h" #include "utils/builtins.h" @@ -139,7 +141,9 @@ typedef struct RelationSyncEntry #define NUM_ROWFILTER_PUBACTIONS 3 /* ExprState array for row filter. One per publication action. */ ExprState *exprstate[NUM_ROWFILTER_PUBACTIONS]; - TupleTableSlot *scantuple; /* tuple table slot for row filter */ + TupleTableSlot *scan_slot; /* tuple table slot for row filter */ + TupleTableSlot *new_slot; /* slot for storing deformed new tuple during updates */ + TupleTableSlot *old_slot; /* slot for storing deformed old tuple during updates */ /* * OID of the relation to publish changes as. For a partition, this may @@ -174,11 +178,15 @@ static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry, /* row filter routines */ static EState *create_estate_for_relation(Relation rel); +static void pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry); static ExprState *pgoutput_row_filter_init_expr(Node *rfnode); static bool pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext); -static bool pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data, - Relation relation, HeapTuple oldtuple, - HeapTuple newtuple, RelationSyncEntry *entry); +static bool pgoutput_row_filter(enum ReorderBufferChangeType changetype, EState *relation, Oid relid, + HeapTuple oldtuple, HeapTuple newtuple, + TupleTableSlot *slot, RelationSyncEntry *entry); +static bool pgoutput_row_filter_update_check(enum ReorderBufferChangeType changetype, Relation relation, + HeapTuple oldtuple, HeapTuple newtuple, + RelationSyncEntry *entry, ReorderBufferChangeType *action); /* * Specify output plugin callbacks @@ -742,26 +750,141 @@ pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext) } /* - * Change is checked against the row filter, if any. - * - * If it returns true, the change is replicated, otherwise, it is not. + * Update is checked against the row filter, if any. + * Updates are transformed to inserts and deletes based on the + * old_tuple and new_tuple. The new action is updated in the + * action parameter. If not updated, action remains as update. + * old-row (no match) new-row (no match) -> (drop change) + * old-row (no match) new row (match) -> INSERT + * old-row (match) new-row (no match) -> DELETE + * old-row (match) new row (match) -> UPDATE + * If the change is to be replicated returns true, else false. */ static bool -pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data, - Relation relation, HeapTuple oldtuple, HeapTuple newtuple, - RelationSyncEntry *entry) +pgoutput_row_filter_update_check(enum ReorderBufferChangeType changetype, Relation relation, + HeapTuple oldtuple, HeapTuple newtuple, + RelationSyncEntry *entry, ReorderBufferChangeType *action) { - EState *estate; - ExprContext *ecxt; - ListCell *lc; - bool result = true; - Oid relid = RelationGetRelid(relation); - List *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */ - bool no_filter[] = {false, false, false}; /* One per pubaction */ + TupleDesc desc = RelationGetDescr(relation); + int i; + bool old_matched, new_matched; + TupleTableSlot *tmp_new_slot, *old_slot, *new_slot; + EState *estate = NULL; Assert(changetype == REORDER_BUFFER_CHANGE_INSERT || changetype == REORDER_BUFFER_CHANGE_UPDATE || changetype == REORDER_BUFFER_CHANGE_DELETE); + + /* Bail out if there is no row filter */ + if (!entry->exprstate[changetype]) + return true; + + /* update requires a new tuple */ + Assert(newtuple); + + elog(DEBUG3, "table \"%s.%s\" has row filter", + get_namespace_name(get_rel_namespace(RelationGetRelid(relation))), + get_rel_name(relation->rd_id)); + + /* Clear the tuples */ + ExecClearTuple(entry->old_slot); + ExecClearTuple(entry->new_slot); + + estate = create_estate_for_relation(relation); + + /* + * If no old_tuple, then none of the replica identity columns changed + * and this would reduce to a simple update. + */ + if (!oldtuple) + { + bool res; + + *action = REORDER_BUFFER_CHANGE_UPDATE; + res = pgoutput_row_filter(changetype, estate, + RelationGetRelid(relation), NULL, newtuple, + NULL, entry); + + FreeExecutorState(estate); + return res; + } + + old_slot = entry->old_slot; + new_slot = entry->new_slot; + tmp_new_slot = new_slot; + + heap_deform_tuple(newtuple, desc, new_slot->tts_values, new_slot->tts_isnull); + heap_deform_tuple(oldtuple, desc, old_slot->tts_values, old_slot->tts_isnull); + + ExecStoreVirtualTuple(old_slot); + ExecStoreVirtualTuple(new_slot); + + /* + * For updates, both the newtuple and oldtuple needs to be checked + * against the row-filter. The newtuple might not have all the + * replica identity columns, in which case it needs to be + * copied over from the oldtuple. + */ + for (i = 0; i < desc->natts; i++) + { + Form_pg_attribute att = TupleDescAttr(desc, i); + + /* if the column in the new_tuple is null, nothing to do */ + if (tmp_new_slot->tts_isnull[i]) + continue; + + /* + * Unchanged toasted replica identity columns are + * only detoasted in the old tuple, copy this over to the newtuple. + */ + if ((att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(tmp_new_slot->tts_values[i])) && + (!old_slot->tts_isnull[i] && + !(VARATT_IS_EXTERNAL_ONDISK(old_slot->tts_values[i])))) + { + if (tmp_new_slot == new_slot) + { + tmp_new_slot = MakeSingleTupleTableSlot(desc, &TTSOpsVirtual);; + ExecClearTuple(tmp_new_slot); + ExecCopySlot(tmp_new_slot, new_slot); + } + + tmp_new_slot->tts_values[i] = old_slot->tts_values[i]; + } + + } + + old_matched = pgoutput_row_filter(changetype, estate, + RelationGetRelid(relation), NULL, NULL, + old_slot, entry); + new_matched = pgoutput_row_filter(changetype, estate, + RelationGetRelid(relation), NULL, NULL, + tmp_new_slot, entry); + + FreeExecutorState(estate); + + if (!old_matched && !new_matched) + return false; + + if (!old_matched && new_matched) + *action = REORDER_BUFFER_CHANGE_INSERT; + else if (old_matched && !new_matched) + *action = REORDER_BUFFER_CHANGE_DELETE; + else if (new_matched && old_matched) + *action = REORDER_BUFFER_CHANGE_UPDATE; + + return true; +} + +/* + * Initialize the row filter, the first time. + */ +static void +pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry) +{ + ListCell *lc; + List *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */ + bool no_filter[] = {false, false, false}; /* One per pubaction */ + /* * If the row filter caching is currently flagged "invalid" then it means we * don't know yet if there is/isn't any row filters for this relation. @@ -971,16 +1094,32 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data, */ oldctx = MemoryContextSwitchTo(CacheMemoryContext); tupdesc = CreateTupleDescCopy(tupdesc); - entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple); + entry->scan_slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple); + entry->old_slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual); + entry->new_slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual); MemoryContextSwitchTo(oldctx); } entry->exprstate_valid = true; } +} - /* Bail out if there is no row filter */ - if (!entry->exprstate[changetype]) - return true; +/* + * Change is checked against the row filter, if any. + * + * If it returns true, the change is replicated, otherwise, it is not. + */ +static bool +pgoutput_row_filter(enum ReorderBufferChangeType changetype, EState *estate, Oid relid, + HeapTuple oldtuple, HeapTuple newtuple, TupleTableSlot *slot, + RelationSyncEntry *entry) +{ + ExprContext *ecxt; + bool result = true; + + Assert(changetype == REORDER_BUFFER_CHANGE_INSERT || + changetype == REORDER_BUFFER_CHANGE_UPDATE || + changetype == REORDER_BUFFER_CHANGE_DELETE); if (message_level_is_interesting(DEBUG3)) elog(DEBUG3, "table \"%s.%s\" has row filter", @@ -989,13 +1128,16 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data, PushActiveSnapshot(GetTransactionSnapshot()); - estate = create_estate_for_relation(relation); - /* Prepare context per tuple */ ecxt = GetPerTupleExprContext(estate); - ecxt->ecxt_scantuple = entry->scantuple; + ecxt->ecxt_scantuple = entry->scan_slot; - ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false); + if (newtuple || oldtuple) + ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false); + else + { + ecxt->ecxt_scantuple = slot; + } /* * NOTE: Multiple publication row filters have already been combined to a @@ -1007,9 +1149,6 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data, result = pgoutput_row_filter_exec_expr(entry->exprstate[changetype], ecxt); } - /* Cleanup allocated resources */ - ResetExprContext(ecxt); - FreeExecutorState(estate); PopActiveSnapshot(); return result; @@ -1029,6 +1168,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, RelationSyncEntry *relentry; TransactionId xid = InvalidTransactionId; Relation ancestor = NULL; + EState *estate = NULL; if (!is_publishable_relation(relation)) return; @@ -1066,6 +1206,9 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, /* Avoid leaking memory by using and resetting our own context */ old = MemoryContextSwitchTo(data->context); + /* Initialize the row_filter */ + pgoutput_row_filter_init(data, relation, relentry); + /* Send the data */ switch (change->action) { @@ -1073,10 +1216,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, { HeapTuple tuple = &change->data.tp.newtuple->tuple; - /* Check row filter. */ - if (!pgoutput_row_filter(change->action, data, relation, NULL, tuple, relentry)) - break; - /* * Schema should be sent before the logic that replaces the * relation because it also sends the ancestor's relation. @@ -1094,6 +1233,17 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, tuple = execute_attr_map_tuple(tuple, relentry->map); } + if (relentry->exprstate[change->action]) + { + estate = create_estate_for_relation(relation); + + /* Check row filter. */ + if (!pgoutput_row_filter(change->action, estate, + RelationGetRelid(relation), NULL, + tuple, NULL, relentry)) + break; + } + OutputPluginPrepareWrite(ctx, true); logicalrep_write_insert(ctx->out, xid, relation, tuple, data->binary); @@ -1105,10 +1255,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, HeapTuple oldtuple = change->data.tp.oldtuple ? &change->data.tp.oldtuple->tuple : NULL; HeapTuple newtuple = &change->data.tp.newtuple->tuple; - - /* Check row filter. */ - if (!pgoutput_row_filter(change->action, data, relation, oldtuple, newtuple, relentry)) - break; + ReorderBufferChangeType modified_action = REORDER_BUFFER_CHANGE_UPDATE; maybe_send_schema(ctx, change, relation, relentry); @@ -1129,9 +1276,34 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } } + /* Check row filter */ + if (!pgoutput_row_filter_update_check(change->action, relation, + oldtuple, newtuple, relentry, + &modified_action)) + break; + OutputPluginPrepareWrite(ctx, true); - logicalrep_write_update(ctx->out, xid, relation, oldtuple, - newtuple, data->binary); + + switch (modified_action) + { + case REORDER_BUFFER_CHANGE_INSERT: + logicalrep_write_insert(ctx->out, xid, relation, newtuple, + data->binary); + break; + case REORDER_BUFFER_CHANGE_UPDATE: + logicalrep_write_update(ctx->out, xid, relation, + oldtuple, newtuple, relentry->old_slot, + relentry->new_slot, + data->binary); + break; + case REORDER_BUFFER_CHANGE_DELETE: + logicalrep_write_delete(ctx->out, xid, relation, oldtuple, + data->binary); + break; + default: + Assert(false); + } + OutputPluginWrite(ctx, true); break; } @@ -1140,10 +1312,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, { HeapTuple oldtuple = &change->data.tp.oldtuple->tuple; - /* Check row filter. */ - if (!pgoutput_row_filter(change->action, data, relation, oldtuple, NULL, relentry)) - break; - maybe_send_schema(ctx, change, relation, relentry); /* Switch relation if publishing via root. */ @@ -1157,6 +1325,17 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, oldtuple = execute_attr_map_tuple(oldtuple, relentry->map); } + if (relentry->exprstate[change->action]) + { + estate = create_estate_for_relation(relation); + + /* Check row filter. */ + if (!pgoutput_row_filter(change->action, estate, + RelationGetRelid(relation), oldtuple, + NULL, NULL, relentry)) + break; + } + OutputPluginPrepareWrite(ctx, true); logicalrep_write_delete(ctx->out, xid, relation, oldtuple, data->binary); @@ -1175,6 +1354,9 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, ancestor = NULL; } + if (estate) + FreeExecutorState(estate); + /* Cleanup */ MemoryContextSwitchTo(old); MemoryContextReset(data->context); @@ -1558,7 +1740,9 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) entry->exprstate_valid = false; entry->pubactions.pubinsert = entry->pubactions.pubupdate = entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false; - entry->scantuple = NULL; + entry->scan_slot = NULL; + entry->new_slot = NULL; + entry->old_slot = NULL; entry->exprstate[REORDER_BUFFER_CHANGE_INSERT] = NULL; entry->exprstate[REORDER_BUFFER_CHANGE_UPDATE] = NULL; entry->exprstate[REORDER_BUFFER_CHANGE_DELETE] = NULL; @@ -1769,10 +1953,10 @@ rel_sync_cache_relation_cb(Datum arg, Oid relid) * Row filter cache cleanups. (Will be rebuilt later if needed). */ entry->exprstate_valid = false; - if (entry->scantuple != NULL) + if (entry->scan_slot != NULL) { - ExecDropSingleTupleTableSlot(entry->scantuple); - entry->scantuple = NULL; + ExecDropSingleTupleTableSlot(entry->scan_slot); + entry->scan_slot = NULL; } /* Cleanup the ExprState for each of the pubactions. */ for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++) diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 83741dc..427c40a 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -16,6 +16,7 @@ #include "access/xact.h" #include "replication/reorderbuffer.h" #include "utils/rel.h" +#include "executor/executor.h" /* * Protocol capabilities @@ -211,7 +212,11 @@ extern void logicalrep_write_insert(StringInfo out, TransactionId xid, extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup); extern void logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, HeapTuple oldtuple, - HeapTuple newtuple, bool binary); + HeapTuple newtuple, TupleTableSlot *oldslot, + TupleTableSlot *newslot, bool binary); +extern void logicalrep_write_update_cached(StringInfo out, TransactionId xid, Relation rel, + TupleTableSlot *oldtuple, TupleTableSlot *newtuple, + bool binary); extern LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup); diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 5b40ff7..aec0059 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -51,7 +51,7 @@ typedef struct ReorderBufferTupleBuf * respectively. They're used by INSERT .. ON CONFLICT .. UPDATE. Users of * logical decoding don't have to care about these. */ -enum ReorderBufferChangeType +typedef enum ReorderBufferChangeType { REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_UPDATE, @@ -65,7 +65,7 @@ enum ReorderBufferChangeType REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, REORDER_BUFFER_CHANGE_TRUNCATE -}; +} ReorderBufferChangeType; /* forward declaration */ struct ReorderBufferTXN; @@ -82,7 +82,7 @@ typedef struct ReorderBufferChange XLogRecPtr lsn; /* The type of change. */ - enum ReorderBufferChangeType action; + ReorderBufferChangeType action; /* Transaction this change belongs to. */ struct ReorderBufferTXN *txn; diff --git a/src/test/subscription/t/027_row_filter.pl b/src/test/subscription/t/027_row_filter.pl index abeaf76..73add45 100644 --- a/src/test/subscription/t/027_row_filter.pl +++ b/src/test/subscription/t/027_row_filter.pl @@ -383,7 +383,8 @@ is($result, qq(13|0|12), 'check replicated rows to tab_rowfilter_4'); # # - 1001, 1002, 1980 already exist from initial data copy # - INSERT (800, 'test 800') NO, because 800 is not > 1000 -# - INSERT (1600, 'test 1600') YES, because 1600 > 1000 and 'test 1600' <> 'filtered' +# - INSERT (1600, 'test 1600') YES, because 1600 > 1000 and 'test 1600' <> 'filtered', +# but row deleted after the update below. # - INSERT (1601, 'test 1601') YES, because 1601 > 1000 and 'test 1601' <> 'filtered' # - INSERT (1700, 'test 1700') YES, because 1700 > 1000 and 'test 1700' <> 'filtered' # - UPDATE (1600, NULL) NO, row filter evaluates to false because NULL is not <> 'filtered' @@ -395,7 +396,6 @@ $result = "SELECT a, b FROM tab_rowfilter_1 ORDER BY 1, 2"); is($result, qq(1001|test 1001 1002|test 1002 -1600|test 1600 1601|test 1601 updated 1980|not filtered), 'check replicated rows to table tab_rowfilter_1'); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 915386e..982df27 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2199,6 +2199,7 @@ ReorderBufferApplyChangeCB ReorderBufferApplyTruncateCB ReorderBufferBeginCB ReorderBufferChange +ReorderBufferChangeType ReorderBufferCommitCB ReorderBufferCommitPreparedCB ReorderBufferDiskChange -- 1.8.3.1