public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v3 1/2] In-place table persistence change 3+ messages / 3 participants [nested] [flat]
* [PATCH v3 1/2] In-place table persistence change @ 2020-11-11 12:51 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Kyotaro Horiguchi @ 2020-11-11 12:51 UTC (permalink / raw) Even though ALTER TABLE SET LOGGED/UNLOGGED does not require data rewriting, currently it runs heap rewrite which causes large amount of file I/O. This patch makes the command run without heap rewrite. Addition to that, SET LOGGED while wal_level > minimal emits WAL using XLOG_FPI instead of massive number of HEAP_INSERT's, which should be smaller. Also this allows for the cleanup of files left behind in the crash of the transaction that created it. --- src/backend/access/rmgrdesc/smgrdesc.c | 23 ++ src/backend/access/transam/README | 10 + src/backend/catalog/storage.c | 420 +++++++++++++++++++++++-- src/backend/commands/tablecmds.c | 246 ++++++++++++--- src/backend/storage/buffer/bufmgr.c | 88 ++++++ src/backend/storage/file/reinit.c | 162 ++++++---- src/backend/storage/smgr/md.c | 4 +- src/backend/storage/smgr/smgr.c | 6 + src/common/relpath.c | 3 +- src/include/catalog/storage.h | 2 + src/include/catalog/storage_xlog.h | 22 +- src/include/common/relpath.h | 5 +- src/include/storage/bufmgr.h | 2 + src/include/storage/smgr.h | 1 + 14 files changed, 854 insertions(+), 140 deletions(-) diff --git a/src/backend/access/rmgrdesc/smgrdesc.c b/src/backend/access/rmgrdesc/smgrdesc.c index 7755553d57..2c109b8ca4 100644 --- a/src/backend/access/rmgrdesc/smgrdesc.c +++ b/src/backend/access/rmgrdesc/smgrdesc.c @@ -40,6 +40,23 @@ smgr_desc(StringInfo buf, XLogReaderState *record) xlrec->blkno, xlrec->flags); pfree(path); } + else if (info == XLOG_SMGR_UNLINK) + { + xl_smgr_unlink *xlrec = (xl_smgr_unlink *) rec; + char *path = relpathperm(xlrec->rnode, xlrec->forkNum); + + appendStringInfoString(buf, path); + pfree(path); + } + else if (info == XLOG_SMGR_BUFPERSISTENCE) + { + xl_smgr_bufpersistence *xlrec = (xl_smgr_bufpersistence *) rec; + char *path = relpathperm(xlrec->rnode, MAIN_FORKNUM); + + appendStringInfoString(buf, path); + appendStringInfo(buf, " persistence %d", xlrec->persistence); + pfree(path); + } } const char * @@ -55,6 +72,12 @@ smgr_identify(uint8 info) case XLOG_SMGR_TRUNCATE: id = "TRUNCATE"; break; + case XLOG_SMGR_UNLINK: + id = "UNLINK"; + break; + case XLOG_SMGR_BUFPERSISTENCE: + id = "BUFPERSISTENCE"; + break; } return id; diff --git a/src/backend/access/transam/README b/src/backend/access/transam/README index 1edc8180c1..51616b2458 100644 --- a/src/backend/access/transam/README +++ b/src/backend/access/transam/README @@ -724,6 +724,16 @@ we must panic and abort recovery. The DBA will have to manually clean up then restart recovery. This is part of the reason for not writing a WAL entry until we've successfully done the original action. +The INITTMP fork file +-------------------------------- + +An INITTMP fork is created when new relation file is created to mark +the relfilenode needs to be cleaned up at recovery time. The file is +removed at transaction end but is left when the process crashes before +the transaction ends. In contrast to 4 above, failure to remove an +INITTMP file will lead to data loss, in which case the server will +shut down. + Skipping WAL for New RelFileNode -------------------------------- diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c index cba7a9ada0..bd9680583b 100644 --- a/src/backend/catalog/storage.c +++ b/src/backend/catalog/storage.c @@ -19,6 +19,7 @@ #include "postgres.h" +#include "access/amapi.h" #include "access/parallel.h" #include "access/visibilitymap.h" #include "access/xact.h" @@ -27,6 +28,7 @@ #include "access/xlogutils.h" #include "catalog/storage.h" #include "catalog/storage_xlog.h" +#include "common/hashfn.h" #include "miscadmin.h" #include "storage/freespace.h" #include "storage/smgr.h" @@ -57,9 +59,16 @@ int wal_skip_threshold = 2048; /* in kilobytes */ * but I'm being paranoid. */ +#define PDOP_DELETE (0) +#define PDOP_UNLINK_FORK (1 << 0) +#define PDOP_SET_PERSISTENCE (1 << 1) + typedef struct PendingRelDelete { RelFileNode relnode; /* relation that may need to be deleted */ + int op; /* operation mask */ + bool bufpersistence; /* buffer persistence to set */ + int unlink_forknum; /* forknum to unlink */ BackendId backend; /* InvalidBackendId if not a temp rel */ bool atCommit; /* T=delete at commit; F=delete at abort */ int nestLevel; /* xact nesting level of request */ @@ -75,6 +84,24 @@ typedef struct PendingRelSync static PendingRelDelete *pendingDeletes = NULL; /* head of linked list */ HTAB *pendingSyncHash = NULL; +typedef struct SRelHashEntry +{ + SMgrRelation srel; + char status; /* for simplehash use */ +} SRelHashEntry; + +/* define hashtable for workarea for pending deletes */ +#define SH_PREFIX srelhash +#define SH_ELEMENT_TYPE SRelHashEntry +#define SH_KEY_TYPE SMgrRelation +#define SH_KEY srel +#define SH_HASH_KEY(tb, key) \ + hash_bytes((unsigned char *)&key, sizeof(SMgrRelation)) +#define SH_EQUAL(tb, a, b) (memcmp(&a, &b, sizeof(SMgrRelation)) == 0) +#define SH_SCOPE static inline +#define SH_DEFINE +#define SH_DECLARE +#include "lib/simplehash.h" /* * AddPendingSync @@ -143,7 +170,17 @@ RelationCreateStorage(RelFileNode rnode, char relpersistence) return NULL; /* placate compiler */ } + /* + * We are going to create a new storage file. If server crashes before the + * current transaction ends the file needs to be cleaned up but there's no + * clue to the orphan files. The inittmp fork works as the sentinel to + * identify that situation. + */ srel = smgropen(rnode, backend); + smgrcreate(srel, INITTMP_FORKNUM, false); + log_smgrcreate(&rnode, INITTMP_FORKNUM); + smgrimmedsync(srel, INITTMP_FORKNUM); + smgrcreate(srel, MAIN_FORKNUM, false); if (needs_wal) @@ -153,12 +190,25 @@ RelationCreateStorage(RelFileNode rnode, char relpersistence) pending = (PendingRelDelete *) MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); pending->relnode = rnode; + pending->op = PDOP_DELETE; pending->backend = backend; pending->atCommit = false; /* delete if abort */ pending->nestLevel = GetCurrentTransactionNestLevel(); pending->next = pendingDeletes; pendingDeletes = pending; + /* drop inittmp fork at commit */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_UNLINK_FORK; + pending->unlink_forknum = INITTMP_FORKNUM; + pending->backend = backend; + pending->atCommit = true; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; + if (relpersistence == RELPERSISTENCE_PERMANENT && !XLogIsNeeded()) { Assert(backend == InvalidBackendId); @@ -168,6 +218,215 @@ RelationCreateStorage(RelFileNode rnode, char relpersistence) return srel; } +/* + * RelationCreateInitFork + * Create physical storage for the init fork of a relation. + * + * Create the init fork for the relation. + * + * This function is transactional. The creation is WAL-logged, and if the + * transaction aborts later on, the init fork will be removed. + */ +void +RelationCreateInitFork(Relation rel) +{ + RelFileNode rnode = rel->rd_node; + PendingRelDelete *pending; + SMgrRelation srel; + PendingRelDelete *prev; + PendingRelDelete *next; + bool create = true; + + /* switch buffer persistence */ + SetRelationBuffersPersistence(rel->rd_smgr, false, false); + + /* + * If we have entries for init-fork operation of this relation, that means + * that we have already registered pending sync entries to drop preexisting + * init fork since before the current transaction started. This function + * reverts that change just by removing the entries. + */ + prev = NULL; + for (pending = pendingDeletes; pending != NULL; pending = next) + { + next = pending->next; + if (RelFileNodeEquals(rnode, pending->relnode) && + pending->op != PDOP_DELETE) + { + if (prev) + prev->next = next; + else + pendingDeletes = next; + pfree(pending); + + create = false; + } + else + { + /* unrelated entry, don't touch it */ + prev = pending; + } + } + + if (!create) + return; + + /* + * We are going to create the init fork. If server crashes before the + * current transaction ends the init fork left alone corrupts data while + * recovery. The inittmp fork works as the sentinel to identify that + * situation. + */ + srel = smgropen(rnode, InvalidBackendId); + smgrcreate(srel, INITTMP_FORKNUM, false); + log_smgrcreate(&rnode, INITTMP_FORKNUM); + smgrimmedsync(srel, INITTMP_FORKNUM); + + /* We don't have existing init fork, create it. */ + smgrcreate(srel, INIT_FORKNUM, false); + + /* + * index-init fork needs further initialization. ambuildempty shoud do + * WAL-log and file sync by itself but otherwise we do that by myself. + */ + if (rel->rd_rel->relkind == RELKIND_INDEX) + rel->rd_indam->ambuildempty(rel); + else + { + log_smgrcreate(&rnode, INIT_FORKNUM); + smgrimmedsync(srel, INIT_FORKNUM); + } + + /* drop this init fork file at abort and revert persistence */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_UNLINK_FORK | PDOP_SET_PERSISTENCE; + pending->unlink_forknum = INIT_FORKNUM; + pending->bufpersistence = true; + pending->backend = InvalidBackendId; + pending->atCommit = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; + + /* drop inittmp fork at abort */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_UNLINK_FORK; + pending->unlink_forknum = INITTMP_FORKNUM; + pending->backend = InvalidBackendId; + pending->atCommit = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; + + /* drop inittmp fork at commit */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_UNLINK_FORK; + pending->unlink_forknum = INITTMP_FORKNUM; + pending->backend = InvalidBackendId; + pending->atCommit = true; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; +} + +/* + * RelationDropInitFork + * Delete physical storage for the init fork of a relation. + * + * Register pending-delete of the init fork. The real deletion is performed by + * smgrDoPendingDeletes at commit. + * + * This function is transactional. If the transaction aborts later on, the + * deletion doesn't happen. + */ +void +RelationDropInitFork(Relation rel) +{ + RelFileNode rnode = rel->rd_node; + PendingRelDelete *pending; + PendingRelDelete *prev; + PendingRelDelete *next; + bool inxact_created = false; + + /* switch buffer persistence */ + SetRelationBuffersPersistence(rel->rd_smgr, true, false); + + /* + * If we have entries for init-fork operation of this relation, that means + * that we have created the init fork in the current transaction. We + * immediately remove the init and inittmp forks immediately in that case. + * Otherwise just reister pending-delete for the existing init fork. + */ + prev = NULL; + for (pending = pendingDeletes; pending != NULL; pending = next) + { + next = pending->next; + if (RelFileNodeEquals(rnode, pending->relnode) && + pending->op != PDOP_DELETE) + { + /* unlink list entry */ + if (prev) + prev->next = next; + else + pendingDeletes = next; + pfree(pending); + + inxact_created = true; + } + else + { + /* unrelated entry, don't touch it */ + prev = pending; + } + } + + if (inxact_created) + { + SMgrRelation srel = smgropen(rnode, InvalidBackendId); + + /* + * INIT/INITTMP forks never be loaded to shared buffer so no point in + * dropping buffers for these files. + */ + log_smgrunlink(&rnode, INIT_FORKNUM); + smgrunlink(srel, INIT_FORKNUM, false); + log_smgrunlink(&rnode, INITTMP_FORKNUM); + smgrunlink(srel, INITTMP_FORKNUM, false); + smgrclose(srel); + return; + } + + /* register drop of this init fork file at commit */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_UNLINK_FORK; + pending->unlink_forknum = INIT_FORKNUM; + pending->backend = InvalidBackendId; + pending->atCommit = true; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; + + /* revert buffer-persistence changes at abort */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_SET_PERSISTENCE; + pending->bufpersistence = false; + pending->backend = InvalidBackendId; + pending->atCommit = true; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; +} + /* * Perform XLogInsert of an XLOG_SMGR_CREATE record to WAL. */ @@ -187,6 +446,44 @@ log_smgrcreate(const RelFileNode *rnode, ForkNumber forkNum) XLogInsert(RM_SMGR_ID, XLOG_SMGR_CREATE | XLR_SPECIAL_REL_UPDATE); } +/* + * Perform XLogInsert of an XLOG_SMGR_UNLINK record to WAL. + */ +void +log_smgrunlink(const RelFileNode *rnode, ForkNumber forkNum) +{ + xl_smgr_unlink xlrec; + + /* + * Make an XLOG entry reporting the file unlink. + */ + xlrec.rnode = *rnode; + xlrec.forkNum = forkNum; + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, sizeof(xlrec)); + XLogInsert(RM_SMGR_ID, XLOG_SMGR_UNLINK | XLR_SPECIAL_REL_UPDATE); +} + +/* + * Perform XLogInsert of an XLOG_SMGR_BUFPERSISTENCE record to WAL. + */ +void +log_smgrbufpersistence(const RelFileNode *rnode, bool persistence) +{ + xl_smgr_bufpersistence xlrec; + + /* + * Make an XLOG entry reporting the change of buffer persistence. + */ + xlrec.rnode = *rnode; + xlrec.persistence = persistence; + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, sizeof(xlrec)); + XLogInsert(RM_SMGR_ID, XLOG_SMGR_BUFPERSISTENCE | XLR_SPECIAL_REL_UPDATE); +} + /* * RelationDropStorage * Schedule unlinking of physical storage at transaction commit. @@ -200,6 +497,7 @@ RelationDropStorage(Relation rel) pending = (PendingRelDelete *) MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); pending->relnode = rel->rd_node; + pending->op = PDOP_DELETE; pending->backend = rel->rd_backend; pending->atCommit = true; /* delete if commit */ pending->nestLevel = GetCurrentTransactionNestLevel(); @@ -602,59 +900,97 @@ smgrDoPendingDeletes(bool isCommit) int nrels = 0, maxrels = 0; SMgrRelation *srels = NULL; + srelhash_hash *close_srels = NULL; + bool found; + prev = NULL; for (pending = pendingDeletes; pending != NULL; pending = next) { + SMgrRelation srel; + next = pending->next; if (pending->nestLevel < nestLevel) { /* outer-level entries should not be processed yet */ prev = pending; + continue; } + + /* unlink list entry first, so we don't retry on failure */ + if (prev) + prev->next = next; else + pendingDeletes = next; + + if (pending->atCommit != isCommit) { - /* unlink list entry first, so we don't retry on failure */ - if (prev) - prev->next = next; - else - pendingDeletes = next; - /* do deletion if called for */ - if (pending->atCommit == isCommit) - { - SMgrRelation srel; - - srel = smgropen(pending->relnode, pending->backend); - - /* allocate the initial array, or extend it, if needed */ - if (maxrels == 0) - { - maxrels = 8; - srels = palloc(sizeof(SMgrRelation) * maxrels); - } - else if (maxrels <= nrels) - { - maxrels *= 2; - srels = repalloc(srels, sizeof(SMgrRelation) * maxrels); - } - - srels[nrels++] = srel; - } /* must explicitly free the list entry */ pfree(pending); /* prev does not change */ + continue; + } + + if (close_srels == NULL) + close_srels = srelhash_create(CurrentMemoryContext, 32, NULL); + + srel = smgropen(pending->relnode, pending->backend); + + /* Uniquify the smgr relations */ + srelhash_insert(close_srels, srel, &found); + + if (pending->op != PDOP_DELETE) + { + if (pending->op & PDOP_UNLINK_FORK) + { + /* other forks needs to drop buffers */ + Assert(pending->unlink_forknum == INIT_FORKNUM || + pending->unlink_forknum == INITTMP_FORKNUM); + + log_smgrunlink(&pending->relnode, pending->unlink_forknum); + smgrunlink(srel, pending->unlink_forknum, false); + + } + + if (pending->op & PDOP_SET_PERSISTENCE) + SetRelationBuffersPersistence(srel, pending->bufpersistence, + false); + } + else + { + /* allocate the initial array, or extend it, if needed */ + if (maxrels == 0) + { + maxrels = 8; + srels = palloc(sizeof(SMgrRelation) * maxrels); + } + else if (maxrels <= nrels) + { + maxrels *= 2; + srels = repalloc(srels, sizeof(SMgrRelation) * maxrels); + } + + srels[nrels++] = srel; } } if (nrels > 0) { smgrdounlinkall(srels, nrels, false); - - for (int i = 0; i < nrels; i++) - smgrclose(srels[i]); - pfree(srels); } + + if (close_srels) + { + srelhash_iterator i; + SRelHashEntry *ent; + + /* close smgr relatoins */ + srelhash_start_iterate(close_srels, &i); + while ((ent = srelhash_iterate(close_srels, &i)) != NULL) + smgrclose(ent->srel); + srelhash_destroy(close_srels); + } } /* @@ -824,7 +1160,8 @@ smgrGetPendingDeletes(bool forCommit, RelFileNode **ptr) for (pending = pendingDeletes; pending != NULL; pending = pending->next) { if (pending->nestLevel >= nestLevel && pending->atCommit == forCommit - && pending->backend == InvalidBackendId) + && pending->backend == InvalidBackendId + && pending->op == PDOP_DELETE) nrels++; } if (nrels == 0) @@ -837,7 +1174,8 @@ smgrGetPendingDeletes(bool forCommit, RelFileNode **ptr) for (pending = pendingDeletes; pending != NULL; pending = pending->next) { if (pending->nestLevel >= nestLevel && pending->atCommit == forCommit - && pending->backend == InvalidBackendId) + && pending->backend == InvalidBackendId && + pending->op == PDOP_DELETE) { *rptr = pending->relnode; rptr++; @@ -917,6 +1255,15 @@ smgr_redo(XLogReaderState *record) reln = smgropen(xlrec->rnode, InvalidBackendId); smgrcreate(reln, xlrec->forkNum, true); } + else if (info == XLOG_SMGR_UNLINK) + { + xl_smgr_unlink *xlrec = (xl_smgr_unlink *) XLogRecGetData(record); + SMgrRelation reln; + + reln = smgropen(xlrec->rnode, InvalidBackendId); + smgrunlink(reln, xlrec->forkNum, true); + smgrclose(reln); + } else if (info == XLOG_SMGR_TRUNCATE) { xl_smgr_truncate *xlrec = (xl_smgr_truncate *) XLogRecGetData(record); @@ -1005,6 +1352,15 @@ smgr_redo(XLogReaderState *record) FreeFakeRelcacheEntry(rel); } + else if (info == XLOG_SMGR_BUFPERSISTENCE) + { + xl_smgr_bufpersistence *xlrec = + (xl_smgr_bufpersistence *) XLogRecGetData(record); + SMgrRelation reln; + + reln = smgropen(xlrec->rnode, InvalidBackendId); + SetRelationBuffersPersistence(reln, xlrec->persistence, true); + } else elog(PANIC, "smgr_redo: unknown op code %u", info); } diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 993da56d43..37a15d31ee 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -50,6 +50,7 @@ #include "commands/defrem.h" #include "commands/event_trigger.h" #include "commands/policy.h" +#include "commands/progress.h" #include "commands/sequence.h" #include "commands/tablecmds.h" #include "commands/tablespace.h" @@ -4917,6 +4918,170 @@ ATParseTransformCmd(List **wqueue, AlteredTableInfo *tab, Relation rel, return newcmd; } +/* + * RelationChangePersistence: do in-place persistence change of a relation + */ +static void +RelationChangePersistence(AlteredTableInfo *tab, char persistence, + LOCKMODE lockmode) +{ + Relation rel; + Relation classRel; + HeapTuple tuple, + newtuple; + Datum new_val[Natts_pg_class]; + bool new_null[Natts_pg_class], + new_repl[Natts_pg_class]; + int i; + List *relids; + ListCell *lc_oid; + + Assert(tab->rewrite == AT_REWRITE_ALTER_PERSISTENCE); + Assert(lockmode == AccessExclusiveLock); + + /* + * Under the following condition, we need to call ATRewriteTable, which + * cannot be false in the AT_REWRITE_ALTER_PERSISTENCE case. + */ + Assert(tab->constraints == NULL && tab->partition_constraint == NULL && + tab->newvals == NULL && !tab->verify_new_notnull); + + rel = table_open(tab->relid, lockmode); + + Assert(rel->rd_rel->relpersistence != persistence); + + elog(DEBUG1, "perform im-place persistnce change"); + + RelationOpenSmgr(rel); + + /* + * First we collect all relations that we need to change persistence. + */ + + /* Collect OIDs of indexes and toast relations */ + relids = RelationGetIndexList(rel); + relids = lcons_oid(rel->rd_id, relids); + + /* Add toast relation if any */ + if (OidIsValid(rel->rd_rel->reltoastrelid)) + { + List *toastidx; + Relation toastrel = table_open(rel->rd_rel->reltoastrelid, lockmode); + + RelationOpenSmgr(toastrel); + relids = lappend_oid(relids, rel->rd_rel->reltoastrelid); + toastidx = RelationGetIndexList(toastrel); + relids = list_concat(relids, toastidx); + pfree(toastidx); + table_close(toastrel, NoLock); + } + + table_close(rel, NoLock); + + /* Make changes in storage */ + classRel = table_open(RelationRelationId, RowExclusiveLock); + + foreach (lc_oid, relids) + { + Oid reloid = lfirst_oid(lc_oid); + Relation r = relation_open(reloid, lockmode); + + /* + * Some access methods do not accept in-place persistence change. For + * example, GiST uses page LSNs to figure out whether a block has + * changed, where UNLOGGED GiST indexes use fake LSNs that are + * incompatible with real LSNs used for LOGGED ones. + * + * XXXX: We don't bother allowing in-place persistence change for index + * methods other than btree for now. + */ + if (r->rd_rel->relkind == RELKIND_INDEX && + r->rd_rel->relam != BTREE_AM_OID) + { + int reindex_flags; + + /* reindex doesn't allow concurrent use of the index */ + table_close(r, NoLock); + + reindex_flags = + REINDEX_REL_SUPPRESS_INDEX_USE | + REINDEX_REL_CHECK_CONSTRAINTS; + + /* Set the same persistence with the parent relation. */ + if (persistence == RELPERSISTENCE_UNLOGGED) + reindex_flags |= REINDEX_REL_FORCE_INDEXES_UNLOGGED; + else + reindex_flags |= REINDEX_REL_FORCE_INDEXES_PERMANENT; + + reindex_index(reloid, reindex_flags, persistence, 0); + + continue; + } + + RelationOpenSmgr(r); + + /* Create or drop init fork */ + if (persistence == RELPERSISTENCE_UNLOGGED) + RelationCreateInitFork(r); + else + RelationDropInitFork(r); + + /* + * When this relation gets WAL-logged, immediately sync all files but + * initfork to establish the initial state on storage. Buffers have + * already flushed out by RelationCreate(Drop)InitFork called just + * above. Initfork should have been synced as needed. + */ + if (persistence == RELPERSISTENCE_PERMANENT) + { + for (i = 0 ; i < INIT_FORKNUM ; i++) + { + if (smgrexists(r->rd_smgr, i)) + smgrimmedsync(r->rd_smgr, i); + } + } + + /* Update catalog */ + tuple = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(reloid)); + if (!HeapTupleIsValid(tuple)) + elog(ERROR, "cache lookup failed for relation %u", reloid); + + memset(new_val, 0, sizeof(new_val)); + memset(new_null, false, sizeof(new_null)); + memset(new_repl, false, sizeof(new_repl)); + + new_val[Anum_pg_class_relpersistence - 1] = CharGetDatum(persistence); + new_null[Anum_pg_class_relpersistence - 1] = false; + new_repl[Anum_pg_class_relpersistence - 1] = true; + + newtuple = heap_modify_tuple(tuple, RelationGetDescr(classRel), + new_val, new_null, new_repl); + + CatalogTupleUpdate(classRel, &newtuple->t_self, newtuple); + heap_freetuple(newtuple); + + /* + * While wal_level >= replica, switching to LOGGED requires the + * relation content to be WAL-logged to recover the table. + */ + if (persistence == RELPERSISTENCE_PERMANENT && XLogIsNeeded()) + { + ForkNumber fork; + + for (fork = 0; fork < INIT_FORKNUM ; fork++) + { + if (smgrexists(r->rd_smgr, fork)) + log_newpage_range(r, fork, + 0, smgrnblocks(r->rd_smgr, fork), false); + } + } + + table_close(r, NoLock); + } + + table_close(classRel, NoLock); +} + /* * ATRewriteTables: ALTER TABLE phase 3 */ @@ -5037,45 +5202,52 @@ ATRewriteTables(AlterTableStmt *parsetree, List **wqueue, LOCKMODE lockmode, tab->relid, tab->rewrite); - /* - * Create transient table that will receive the modified data. - * - * Ensure it is marked correctly as logged or unlogged. We have - * to do this here so that buffers for the new relfilenode will - * have the right persistence set, and at the same time ensure - * that the original filenode's buffers will get read in with the - * correct setting (i.e. the original one). Otherwise a rollback - * after the rewrite would possibly result with buffers for the - * original filenode having the wrong persistence setting. - * - * NB: This relies on swap_relation_files() also swapping the - * persistence. That wouldn't work for pg_class, but that can't be - * unlogged anyway. - */ - OIDNewHeap = make_new_heap(tab->relid, NewTableSpace, persistence, - lockmode); + if (tab->rewrite == AT_REWRITE_ALTER_PERSISTENCE) + RelationChangePersistence(tab, persistence, lockmode); + else + { + /* + * Create transient table that will receive the modified data. + * + * Ensure it is marked correctly as logged or unlogged. We + * have to do this here so that buffers for the new relfilenode + * will have the right persistence set, and at the same time + * ensure that the original filenode's buffers will get read in + * with the correct setting (i.e. the original one). Otherwise + * a rollback after the rewrite would possibly result with + * buffers for the original filenode having the wrong + * persistence setting. + * + * NB: This relies on swap_relation_files() also swapping the + * persistence. That wouldn't work for pg_class, but that can't + * be unlogged anyway. + */ + OIDNewHeap = make_new_heap(tab->relid, NewTableSpace, persistence, + lockmode); - /* - * Copy the heap data into the new table with the desired - * modifications, and test the current data within the table - * against new constraints generated by ALTER TABLE commands. - */ - ATRewriteTable(tab, OIDNewHeap, lockmode); + /* + * Copy the heap data into the new table with the desired + * modifications, and test the current data within the table + * against new constraints generated by ALTER TABLE commands. + */ + ATRewriteTable(tab, OIDNewHeap, lockmode); - /* - * Swap the physical files of the old and new heaps, then rebuild - * indexes and discard the old heap. We can use RecentXmin for - * the table's new relfrozenxid because we rewrote all the tuples - * in ATRewriteTable, so no older Xid remains in the table. Also, - * we never try to swap toast tables by content, since we have no - * interest in letting this code work on system catalogs. - */ - finish_heap_swap(tab->relid, OIDNewHeap, - false, false, true, - !OidIsValid(tab->newTableSpace), - RecentXmin, - ReadNextMultiXactId(), - persistence); + /* + * Swap the physical files of the old and new heaps, then + * rebuild indexes and discard the old heap. We can use + * RecentXmin for the table's new relfrozenxid because we + * rewrote all the tuples in ATRewriteTable, so no older Xid + * remains in the table. Also, we never try to swap toast + * tables by content, since we have no interest in letting this + * code work on system catalogs. + */ + finish_heap_swap(tab->relid, OIDNewHeap, + false, false, true, + !OidIsValid(tab->newTableSpace), + RecentXmin, + ReadNextMultiXactId(), + persistence); + } } else { diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index 71b5852224..b730b4417c 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -37,6 +37,7 @@ #include "access/xlog.h" #include "catalog/catalog.h" #include "catalog/storage.h" +#include "catalog/storage_xlog.h" #include "executor/instrument.h" #include "lib/binaryheap.h" #include "miscadmin.h" @@ -3032,6 +3033,93 @@ DropRelFileNodeBuffers(RelFileNodeBackend rnode, ForkNumber *forkNum, } } +/* --------------------------------------------------------------------- + * SetRelFileNodeBuffersPersistence + * + * This function changes the persistence of all buffer pages of a relation + * then writes all dirty pages of the relation out to disk when switching + * to PERMANENT. (or more accurately, out to kernel disk buffers), + * ensuring that the kernel has an up-to-date view of the relation. + * + * Generally, the caller should be holding AccessExclusiveLock on the + * target relation to ensure that no other backend is busy dirtying + * more blocks of the relation; the effects can't be expected to last + * after the lock is released. + * + * XXX currently it sequentially searches the buffer pool, should be + * changed to more clever ways of searching. This routine is not + * used in any performance-critical code paths, so it's not worth + * adding additional overhead to normal paths to make it go faster; + * but see also DropRelFileNodeBuffers. + * -------------------------------------------------------------------- + */ +void +SetRelationBuffersPersistence(SMgrRelation srel, bool permanent, bool isRedo) +{ + int i; + RelFileNodeBackend rnode = srel->smgr_rnode; + + Assert (!RelFileNodeBackendIsTemp(rnode)); + + if (!isRedo) + log_smgrbufpersistence(&srel->smgr_rnode.node, permanent); + + ResourceOwnerEnlargeBuffers(CurrentResourceOwner); + + for (i = 0; i < NBuffers; i++) + { + BufferDesc *bufHdr = GetBufferDescriptor(i); + uint32 buf_state; + + if (!RelFileNodeEquals(bufHdr->tag.rnode, rnode.node)) + continue; + + ReservePrivateRefCountEntry(); + + buf_state = LockBufHdr(bufHdr); + + if (!RelFileNodeEquals(bufHdr->tag.rnode, rnode.node)) + { + UnlockBufHdr(bufHdr, buf_state); + continue; + } + + if (permanent) + { + /* Init fork is being dropped, drop buffers for it. */ + if (bufHdr->tag.forkNum == INIT_FORKNUM) + { + InvalidateBuffer(bufHdr); + continue; + } + + buf_state |= BM_PERMANENT; + pg_atomic_write_u32(&bufHdr->state, buf_state); + + /* we flush this buffer when switching to PERMANENT */ + if ((buf_state & (BM_VALID | BM_DIRTY)) == (BM_VALID | BM_DIRTY)) + { + PinBuffer_Locked(bufHdr); + LWLockAcquire(BufferDescriptorGetContentLock(bufHdr), + LW_SHARED); + FlushBuffer(bufHdr, srel); + LWLockRelease(BufferDescriptorGetContentLock(bufHdr)); + UnpinBuffer(bufHdr, true); + } + else + UnlockBufHdr(bufHdr, buf_state); + } + else + { + /* init fork is always BM_PERMANENT. See BufferAlloc */ + if (bufHdr->tag.forkNum != INIT_FORKNUM) + buf_state &= ~BM_PERMANENT; + + UnlockBufHdr(bufHdr, buf_state); + } + } +} + /* --------------------------------------------------------------------- * DropRelFileNodesAllBuffers * diff --git a/src/backend/storage/file/reinit.c b/src/backend/storage/file/reinit.c index 40c758d789..adcb54b0fa 100644 --- a/src/backend/storage/file/reinit.c +++ b/src/backend/storage/file/reinit.c @@ -31,7 +31,8 @@ static void ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, typedef struct { Oid reloid; /* hash key */ -} unlogged_relation_entry; + bool dirty; /* to be removed */ +} relfile_entry; /* * Reset unlogged relations from before the last restart. @@ -151,6 +152,8 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) DIR *dbspace_dir; struct dirent *de; char rm_path[MAXPGPATH * 2]; + HTAB *hash; + HASHCTL ctl; /* Caller must specify at least one operation. */ Assert((op & (UNLOGGED_RELATION_CLEANUP | UNLOGGED_RELATION_INIT)) != 0); @@ -160,88 +163,86 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) * the files with init forks. Then, we go through again and nuke * everything with the same OID except the init fork. */ + + /* + * It's possible that someone could create a ton of unlogged relations + * in the same database & tablespace, so we'd better use a hash table + * rather than an array or linked list to keep track of which files + * need to be reset. Otherwise, this cleanup operation would be + * O(n^2). + */ + memset(&ctl, 0, sizeof(ctl)); + ctl.keysize = sizeof(Oid); + ctl.entrysize = sizeof(relfile_entry); + hash = hash_create("inittmp hash", 32, &ctl, HASH_ELEM | HASH_BLOBS); + + /* Collect inttmp forks in the directory. */ + dbspace_dir = AllocateDir(dbspacedirname); + while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL) + { + int oidchars; + ForkNumber forkNum; + + /* Skip anything that doesn't look like a relation data file. */ + if (!parse_filename_for_nontemp_relation(de->d_name, &oidchars, + &forkNum)) + continue; + + /* Record init and inittmp forks */ + if (forkNum == INIT_FORKNUM || forkNum == INITTMP_FORKNUM) + { + Oid key; + relfile_entry *ent; + bool found; + + /* + * Record the relfilenode. If it has INITTMP fork, the all files + * needs to be cleaned up. Otherwise the relfilenode is cleaned up + * according to the unloggedness. + */ + key = atooid(de->d_name); + ent = hash_search(hash, &key, HASH_ENTER, &found); + + if (!found) + ent->dirty = false; + + if (forkNum == INITTMP_FORKNUM) + ent->dirty = true; + } + } + + /* Done with the first pass. */ + FreeDir(dbspace_dir); + if ((op & UNLOGGED_RELATION_CLEANUP) != 0) { - HTAB *hash; - HASHCTL ctl; - - /* - * It's possible that someone could create a ton of unlogged relations - * in the same database & tablespace, so we'd better use a hash table - * rather than an array or linked list to keep track of which files - * need to be reset. Otherwise, this cleanup operation would be - * O(n^2). - */ - ctl.keysize = sizeof(Oid); - ctl.entrysize = sizeof(unlogged_relation_entry); - ctl.hcxt = CurrentMemoryContext; - hash = hash_create("unlogged relation OIDs", 32, &ctl, - HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); - - /* Scan the directory. */ - dbspace_dir = AllocateDir(dbspacedirname); - while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL) - { - ForkNumber forkNum; - int oidchars; - unlogged_relation_entry ent; - - /* Skip anything that doesn't look like a relation data file. */ - if (!parse_filename_for_nontemp_relation(de->d_name, &oidchars, - &forkNum)) - continue; - - /* Also skip it unless this is the init fork. */ - if (forkNum != INIT_FORKNUM) - continue; - - /* - * Put the OID portion of the name into the hash table, if it - * isn't already. - */ - ent.reloid = atooid(de->d_name); - (void) hash_search(hash, &ent, HASH_ENTER, NULL); - } - - /* Done with the first pass. */ - FreeDir(dbspace_dir); - - /* - * If we didn't find any init forks, there's no point in continuing; - * we can bail out now. - */ - if (hash_get_num_entries(hash) == 0) - { - hash_destroy(hash); - return; - } - /* * Now, make a second pass and remove anything that matches. */ dbspace_dir = AllocateDir(dbspacedirname); while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL) { - ForkNumber forkNum; - int oidchars; - unlogged_relation_entry ent; + ForkNumber forkNum; + int oidchars; + Oid key; + relfile_entry *ent; /* Skip anything that doesn't look like a relation data file. */ if (!parse_filename_for_nontemp_relation(de->d_name, &oidchars, &forkNum)) continue; - /* We never remove the init fork. */ - if (forkNum == INIT_FORKNUM) - continue; - /* * See whether the OID portion of the name shows up in the hash * table. If so, nuke it! */ - ent.reloid = atooid(de->d_name); - if (hash_search(hash, &ent, HASH_FIND, NULL)) + key = atooid(de->d_name); + ent = hash_search(hash, &key, HASH_FIND, NULL); + + /* we don't remove clean init file */ + if (ent && (ent->dirty || forkNum != INIT_FORKNUM)) { + /* so, nuke it! */ snprintf(rm_path, sizeof(rm_path), "%s/%s", dbspacedirname, de->d_name); if (unlink(rm_path) < 0) @@ -256,7 +257,6 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) /* Cleanup is complete. */ FreeDir(dbspace_dir); - hash_destroy(hash); } /* @@ -277,12 +277,42 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) char oidbuf[OIDCHARS + 1]; char srcpath[MAXPGPATH * 2]; char dstpath[MAXPGPATH]; + Oid key; + relfile_entry *ent; /* Skip anything that doesn't look like a relation data file. */ if (!parse_filename_for_nontemp_relation(de->d_name, &oidchars, &forkNum)) continue; + /* + * See whether the OID portion of the name shows up in the hash + * table. + */ + key = atooid(de->d_name); + ent = hash_search(hash, &key, HASH_FIND, NULL); + + /* we don't remove clean init file */ + if (ent && (ent->dirty || forkNum != INIT_FORKNUM)) + { + /* + * The file is dirty. It shoudl have been removed once at + * cleanup time but recovery can create them again. Remove + * them. + */ + snprintf(rm_path, sizeof(rm_path), "%s/%s", + dbspacedirname, de->d_name); + if (unlink(rm_path) < 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not remove file \"%s\": %m", + rm_path))); + else + elog(DEBUG2, "unlinked file \"%s\"", rm_path); + + continue; + } + /* Also skip it unless this is the init fork. */ if (forkNum != INIT_FORKNUM) continue; @@ -351,6 +381,8 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) */ fsync_fname(dbspacedirname, true); } + + hash_destroy(hash); } /* diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c index 0643d714fb..416fd859e6 100644 --- a/src/backend/storage/smgr/md.c +++ b/src/backend/storage/smgr/md.c @@ -338,8 +338,10 @@ mdunlinkfork(RelFileNodeBackend rnode, ForkNumber forkNum, bool isRedo) if (ret == 0 || errno != ENOENT) { ret = unlink(path); + + /* failure of removing inittmp fork leads to a data loss. */ if (ret < 0 && errno != ENOENT) - ereport(WARNING, + ereport((forkNum != INITTMP_FORKNUM ? WARNING : ERROR), (errcode_for_file_access(), errmsg("could not remove file \"%s\": %m", path))); } diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c index 0f31ff3822..4102d3d59c 100644 --- a/src/backend/storage/smgr/smgr.c +++ b/src/backend/storage/smgr/smgr.c @@ -644,6 +644,12 @@ smgrimmedsync(SMgrRelation reln, ForkNumber forknum) smgrsw[reln->smgr_which].smgr_immedsync(reln, forknum); } +void +smgrunlink(SMgrRelation reln, ForkNumber forknum, bool isRedo) +{ + smgrsw[reln->smgr_which].smgr_unlink(reln->smgr_rnode, forknum, isRedo); +} + /* * AtEOXact_SMgr * diff --git a/src/common/relpath.c b/src/common/relpath.c index 1f5c426ec0..2954cd9c24 100644 --- a/src/common/relpath.c +++ b/src/common/relpath.c @@ -34,7 +34,8 @@ const char *const forkNames[] = { "main", /* MAIN_FORKNUM */ "fsm", /* FSM_FORKNUM */ "vm", /* VISIBILITYMAP_FORKNUM */ - "init" /* INIT_FORKNUM */ + "init", /* INIT_FORKNUM */ + "itmp" /* INITTMP_FORKNUM */ }; StaticAssertDecl(lengthof(forkNames) == (MAX_FORKNUM + 1), diff --git a/src/include/catalog/storage.h b/src/include/catalog/storage.h index 0ab32b44e9..382623159c 100644 --- a/src/include/catalog/storage.h +++ b/src/include/catalog/storage.h @@ -23,6 +23,8 @@ extern int wal_skip_threshold; extern SMgrRelation RelationCreateStorage(RelFileNode rnode, char relpersistence); +extern void RelationCreateInitFork(Relation rel); +extern void RelationDropInitFork(Relation rel); extern void RelationDropStorage(Relation rel); extern void RelationPreserveStorage(RelFileNode rnode, bool atCommit); extern void RelationPreTruncate(Relation rel); diff --git a/src/include/catalog/storage_xlog.h b/src/include/catalog/storage_xlog.h index f0814f1458..0fd0832a8b 100644 --- a/src/include/catalog/storage_xlog.h +++ b/src/include/catalog/storage_xlog.h @@ -22,13 +22,17 @@ /* * Declarations for smgr-related XLOG records * - * Note: we log file creation and truncation here, but logging of deletion - * actions is handled by xact.c, because it is part of transaction commit. + * Note: we log file creation, truncation, deletion and persistence change + * here. logging of deletion actions is mainly handled by xact.c, because it is + * part of transaction commit, but we log deletions happens outside of a + * transaction. */ /* XLOG gives us high 4 bits */ #define XLOG_SMGR_CREATE 0x10 #define XLOG_SMGR_TRUNCATE 0x20 +#define XLOG_SMGR_UNLINK 0x30 +#define XLOG_SMGR_BUFPERSISTENCE 0x40 typedef struct xl_smgr_create { @@ -36,6 +40,18 @@ typedef struct xl_smgr_create ForkNumber forkNum; } xl_smgr_create; +typedef struct xl_smgr_unlink +{ + RelFileNode rnode; + ForkNumber forkNum; +} xl_smgr_unlink; + +typedef struct xl_smgr_bufpersistence +{ + RelFileNode rnode; + bool persistence; +} xl_smgr_bufpersistence; + /* flags for xl_smgr_truncate */ #define SMGR_TRUNCATE_HEAP 0x0001 #define SMGR_TRUNCATE_VM 0x0002 @@ -51,6 +67,8 @@ typedef struct xl_smgr_truncate } xl_smgr_truncate; extern void log_smgrcreate(const RelFileNode *rnode, ForkNumber forkNum); +extern void log_smgrunlink(const RelFileNode *rnode, ForkNumber forkNum); +extern void log_smgrbufpersistence(const RelFileNode *rnode, bool persistence); extern void smgr_redo(XLogReaderState *record); extern void smgr_desc(StringInfo buf, XLogReaderState *record); diff --git a/src/include/common/relpath.h b/src/include/common/relpath.h index a44be11ca0..4305bdbe96 100644 --- a/src/include/common/relpath.h +++ b/src/include/common/relpath.h @@ -43,7 +43,8 @@ typedef enum ForkNumber MAIN_FORKNUM = 0, FSM_FORKNUM, VISIBILITYMAP_FORKNUM, - INIT_FORKNUM + INIT_FORKNUM, + INITTMP_FORKNUM /* * NOTE: if you add a new fork, change MAX_FORKNUM and possibly @@ -52,7 +53,7 @@ typedef enum ForkNumber */ } ForkNumber; -#define MAX_FORKNUM INIT_FORKNUM +#define MAX_FORKNUM INITTMP_FORKNUM #define FORKNAMECHARS 4 /* max chars for a fork name */ diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h index ff6cd0fc54..d9752a8317 100644 --- a/src/include/storage/bufmgr.h +++ b/src/include/storage/bufmgr.h @@ -205,6 +205,8 @@ extern void FlushRelationsAllBuffers(struct SMgrRelationData **smgrs, int nrels) extern void FlushDatabaseBuffers(Oid dbid); extern void DropRelFileNodeBuffers(RelFileNodeBackend rnode, ForkNumber *forkNum, int nforks, BlockNumber *firstDelBlock); +extern void SetRelationBuffersPersistence(struct SMgrRelationData *srel, + bool permanent, bool isRedo); extern void DropRelFileNodesAllBuffers(RelFileNodeBackend *rnodes, int nnodes); extern void DropDatabaseBuffers(Oid dbid); diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h index ebf4a199dc..8be17d9afc 100644 --- a/src/include/storage/smgr.h +++ b/src/include/storage/smgr.h @@ -86,6 +86,7 @@ extern void smgrclose(SMgrRelation reln); extern void smgrcloseall(void); extern void smgrclosenode(RelFileNodeBackend rnode); extern void smgrcreate(SMgrRelation reln, ForkNumber forknum, bool isRedo); +extern void smgrunlink(SMgrRelation reln, ForkNumber forknum, bool isRedo); extern void smgrdosyncall(SMgrRelation *rels, int nrels); extern void smgrdounlinkall(SMgrRelation *rels, int nrels, bool isRedo); extern void smgrextend(SMgrRelation reln, ForkNumber forknum, -- 2.27.0 ----Next_Part(Fri_Jan__8_14_47_05_2021_579)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v3-0002-New-command-ALTER-TABLE-ALL-IN-TABLESPACE-SET-LOG.patch" ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: bogus: logical replication rows/cols combinations @ 2022-04-28 15:35 Tomas Vondra <[email protected]> 0 siblings, 1 reply; 3+ messages in thread From: Tomas Vondra @ 2022-04-28 15:35 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Alvaro Herrera <[email protected]>; Pg Hackers <[email protected]> On 4/28/22 05:17, Amit Kapila wrote: > On Thu, Apr 28, 2022 at 3:26 AM Tomas Vondra > <[email protected]> wrote: >> >> so I've been looking at tweaking the code so that the behavior matches >> Alvaro's expectations. It passes check-world but I'm not claiming it's >> nowhere near commitable - the purpose is mostly to give better idea of >> how invasive the change is etc. >> > > I was just skimming through the patch and didn't find anything related > to initial sync handling. I feel the behavior should be same for > initial sync and replication. > Yeah, sorry for not mentioning that - my goal was to explore and try getting the behavior in regular replication right first, before attempting to do the same thing in tablesync. Attached is a patch doing the same thing in tablesync. The overall idea is to generate copy statement with CASE expressions, applying filters to individual columns. For Alvaro's example, this generates something like SELECT (CASE WHEN (a < 0) OR (a > 0) THEN a ELSE NULL END) AS a, (CASE WHEN (a > 0) THEN b ELSE NULL END) AS b, (CASE WHEN (a < 0) THEN c ELSE NULL END) AS c FROM uno WHERE (a < 0) OR (a > 0) And that seems to work fine. Similarly to regular replication we have to use both the "total" column list (union of per-publication lists) and per-publication (row filter + column list), but that's expected. There's a couple options how we might optimize this for common cases. For example if there's just a single publication, there's no need to generate the CASE expressions - the WHERE filter will do the trick. regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company Attachments: [text/x-patch] rework-column-row-filtering-v2.patch (38.2K, ../../[email protected]/2-rework-column-row-filtering-v2.patch) download | inline diff: 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); ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: bogus: logical replication rows/cols combinations @ 2022-05-02 16:30 Alvaro Herrera <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Alvaro Herrera @ 2022-05-02 16:30 UTC (permalink / raw) To: Tomas Vondra <[email protected]>; +Cc: Amit Kapila <[email protected]>; Pg Hackers <[email protected]> On 2022-Apr-28, Tomas Vondra wrote: > SELECT > (CASE WHEN (a < 0) OR (a > 0) THEN a ELSE NULL END) AS a, > (CASE WHEN (a > 0) THEN b ELSE NULL END) AS b, > (CASE WHEN (a < 0) THEN c ELSE NULL END) AS c > FROM uno WHERE (a < 0) OR (a > 0) BTW, looking at the new COPY commands, the idea of "COPY table_foo (PUBLICATION pub1, pub2)" is looking more and more attractive, as a replacement for having the replica cons up an ad-hoc subquery to COPY from. Something to think about for pg16, maybe. -- Álvaro Herrera Breisgau, Deutschland — https://www.EnterpriseDB.com/ "You're _really_ hosed if the person doing the hiring doesn't understand relational systems: you end up with a whole raft of programmers, none of whom has had a Date with the clue stick." (Andrew Sullivan) ^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2022-05-02 16:30 UTC | newest] Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-11-11 12:51 [PATCH v3 1/2] In-place table persistence change Kyotaro Horiguchi <[email protected]> 2022-04-28 15:35 Re: bogus: logical replication rows/cols combinations Tomas Vondra <[email protected]> 2022-05-02 16:30 ` Re: bogus: logical replication rows/cols combinations Alvaro Herrera <[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