public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v5 1/2] In-place table persistence change 3+ messages / 3 participants [nested] [flat]
* [PATCH v5 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 | 8 + src/backend/access/transam/xlog.c | 17 + src/backend/catalog/storage.c | 436 +++++++++++++++++++++++-- src/backend/commands/tablecmds.c | 246 +++++++++++--- src/backend/storage/buffer/bufmgr.c | 88 +++++ src/backend/storage/file/reinit.c | 316 ++++++++++++------ src/backend/storage/smgr/md.c | 13 +- src/backend/storage/smgr/smgr.c | 6 + src/common/relpath.c | 4 +- src/include/catalog/storage.h | 2 + src/include/catalog/storage_xlog.h | 22 +- src/include/common/relpath.h | 6 +- src/include/storage/bufmgr.h | 2 + src/include/storage/md.h | 2 + src/include/storage/reinit.h | 3 +- src/include/storage/smgr.h | 1 + 17 files changed, 1028 insertions(+), 167 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..547107a771 100644 --- a/src/backend/access/transam/README +++ b/src/backend/access/transam/README @@ -724,6 +724,14 @@ 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 CLEANUP fork file +-------------------------------- + +An CLEANUP fork is created when a new relation file is created to mark +the relfilenode needs to be cleaned up at recovery time. In contrast +to 4 above, failure to remove an CLEANUP fork file will lead to data +loss, in which case the server will shut down. + Skipping WAL for New RelFileNode -------------------------------- diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index b18257c198..6dcbcbe387 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -40,6 +40,7 @@ #include "catalog/catversion.h" #include "catalog/pg_control.h" #include "catalog/pg_database.h" +#include "catalog/storage.h" #include "commands/progress.h" #include "commands/tablespace.h" #include "common/controldata_utils.h" @@ -4442,6 +4443,14 @@ ReadRecord(XLogReaderState *xlogreader, int emode, { ereport(DEBUG1, (errmsg_internal("reached end of WAL in pg_wal, entering archive recovery"))); + + /* cleanup garbage files left during crash recovery */ + ResetUnloggedRelations(UNLOGGED_RELATION_DROP_BUFFER | + UNLOGGED_RELATION_CLEANUP); + + /* run rollback cleanup if any */ + smgrDoPendingDeletes(false); + InArchiveRecovery = true; if (StandbyModeRequested) StandbyMode = true; @@ -7455,6 +7464,14 @@ StartupXLOG(void) } } + /* cleanup garbage files left during crash recovery */ + if (!InArchiveRecovery) + ResetUnloggedRelations(UNLOGGED_RELATION_DROP_BUFFER | + UNLOGGED_RELATION_CLEANUP); + + /* run rollback cleanup if any */ + smgrDoPendingDeletes(false); + /* Allow resource managers to do any required cleanup. */ for (rmid = 0; rmid <= RM_MAX_ID; rmid++) { diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c index cba7a9ada0..c54d70747f 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 cleanup fork works as the sentinel to + * identify that situation. + */ srel = smgropen(rnode, backend); + smgrcreate(srel, CLEANUP2_FORKNUM, false); + log_smgrcreate(&rnode, CLEANUP2_FORKNUM); + smgrimmedsync(srel, CLEANUP2_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 cleanup fork at commit */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_UNLINK_FORK; + pending->unlink_forknum = CLEANUP2_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,218 @@ 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 delete 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 && + ((pending->op & PDOP_UNLINK_FORK) != 0 && + pending->unlink_forknum == CLEANUP_FORKNUM)) + { + 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 cleanup fork works as the sentinel to identify that + * situation. + */ + srel = smgropen(rnode, InvalidBackendId); + smgrcreate(srel, CLEANUP_FORKNUM, false); + log_smgrcreate(&rnode, CLEANUP_FORKNUM); + smgrimmedsync(srel, CLEANUP_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 cleanup fork at abort */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_UNLINK_FORK; + pending->unlink_forknum = CLEANUP_FORKNUM; + pending->backend = InvalidBackendId; + pending->atCommit = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; + + /* drop cleanup fork at commit */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = rnode; + pending->op = PDOP_UNLINK_FORK; + pending->unlink_forknum = CLEANUP_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 cleanup 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 && + ((pending->op & PDOP_UNLINK_FORK) != 0 && + pending->unlink_forknum == CLEANUP_FORKNUM)) + { + /* 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/CLEANUP 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, CLEANUP_FORKNUM); + smgrunlink(srel, CLEANUP_FORKNUM, false); + 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 = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; +} + /* * Perform XLogInsert of an XLOG_SMGR_CREATE record to WAL. */ @@ -187,6 +449,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 +500,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 +903,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 == CLEANUP_FORKNUM || + pending->unlink_forknum == CLEANUP2_FORKNUM); + + log_smgrunlink(&pending->relnode, pending->unlink_forknum); + smgrunlink(srel, pending->unlink_forknum, false); + + } + + if (pending->op & PDOP_SET_PERSISTENCE) + SetRelationBuffersPersistence(srel, pending->bufpersistence, + InRecovery); + } + 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 +1163,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 +1177,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 +1258,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 +1355,28 @@ smgr_redo(XLogReaderState *record) FreeFakeRelcacheEntry(rel); } + else if (info == XLOG_SMGR_BUFPERSISTENCE) + { + xl_smgr_bufpersistence *xlrec = + (xl_smgr_bufpersistence *) XLogRecGetData(record); + SMgrRelation reln; + PendingRelDelete *pending; + + reln = smgropen(xlrec->rnode, InvalidBackendId); + SetRelationBuffersPersistence(reln, xlrec->persistence, true); + + /* revert buffer-persistence changes at abort */ + pending = (PendingRelDelete *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); + pending->relnode = xlrec->rnode; + pending->op = PDOP_SET_PERSISTENCE; + pending->bufpersistence = !xlrec->persistence; + pending->backend = InvalidBackendId; + pending->atCommit = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingDeletes; + pendingDeletes = pending; + } 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 561c212092..eacbdc6447 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" @@ -3094,6 +3095,93 @@ DropRelFileNodeBuffers(SMgrRelation smgr_reln, 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..0eac1956cc 100644 --- a/src/backend/storage/file/reinit.c +++ b/src/backend/storage/file/reinit.c @@ -16,29 +16,50 @@ #include <unistd.h> +#include "access/xlog.h" +#include "catalog/pg_tablespace_d.h" #include "common/relpath.h" +#include "storage/bufmgr.h" #include "storage/copydir.h" #include "storage/fd.h" +#include "storage/md.h" #include "storage/reinit.h" +#include "storage/smgr.h" #include "utils/hsearch.h" #include "utils/memutils.h" static void ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, - int op); + Oid tspid, int op); static void ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, - int op); + Oid tspid, Oid dbid, int op); typedef struct { Oid reloid; /* hash key */ -} unlogged_relation_entry; + bool has_init; /* has INIT fork */ + bool dirty_init; /* needs to remove INIT fork */ + bool dirty_all; /* needs to remove all forks */ +} relfile_entry; /* - * Reset unlogged relations from before the last restart. + * Clean up and reset relation files from before the last restart. * - * If op includes UNLOGGED_RELATION_CLEANUP, we remove all forks of any - * relation with an "init" fork, except for the "init" fork itself. + * If op includes UNLOGGED_RELATION_CLEANUP, we perform different operations + * depending on the existence of the "cleanup" forks. * + * If CLEANUP_FORKNUM (clup) is present, we remove the init fork of the same + * relation along with the clup fork. + * + * If CLEANUP2_FORKNUM (cln2) is present we remove the whole relation along + * with the cln2 fork. + * + * Otherwise, if the "init" fork is found. we remove all forks of any relation + * with the "init" fork, except for the "init" fork itself. + * + * + * If op includes UNLOGGED_RELATION_DROP_BUFFER, we drop all buffers for all + * relations that have the "cleanup" and/or the "init" forks. + * * If op includes UNLOGGED_RELATION_INIT, we copy the "init" fork to the main * fork. */ @@ -68,7 +89,7 @@ ResetUnloggedRelations(int op) /* * First process unlogged files in pg_default ($PGDATA/base) */ - ResetUnloggedRelationsInTablespaceDir("base", op); + ResetUnloggedRelationsInTablespaceDir("base", DEFAULTTABLESPACE_OID, op); /* * Cycle through directories for all non-default tablespaces. @@ -77,13 +98,19 @@ ResetUnloggedRelations(int op) while ((spc_de = ReadDir(spc_dir, "pg_tblspc")) != NULL) { + Oid tspid; + if (strcmp(spc_de->d_name, ".") == 0 || strcmp(spc_de->d_name, "..") == 0) continue; snprintf(temp_path, sizeof(temp_path), "pg_tblspc/%s/%s", spc_de->d_name, TABLESPACE_VERSION_DIRECTORY); - ResetUnloggedRelationsInTablespaceDir(temp_path, op); + + tspid = atooid(spc_de->d_name); + Assert(tspid != 0); + + ResetUnloggedRelationsInTablespaceDir(temp_path, tspid, op); } FreeDir(spc_dir); @@ -99,7 +126,8 @@ ResetUnloggedRelations(int op) * Process one tablespace directory for ResetUnloggedRelations */ static void -ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op) +ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, + Oid tspid, int op) { DIR *ts_dir; struct dirent *de; @@ -126,6 +154,8 @@ ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op) while ((de = ReadDir(ts_dir, tsdirname)) != NULL) { + Oid dbid; + /* * We're only interested in the per-database directories, which have * numeric names. Note that this code will also (properly) ignore "." @@ -136,7 +166,10 @@ ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op) snprintf(dbspace_path, sizeof(dbspace_path), "%s/%s", tsdirname, de->d_name); - ResetUnloggedRelationsInDbspaceDir(dbspace_path, op); + dbid = atooid(de->d_name); + Assert(dbid != 0); + + ResetUnloggedRelationsInDbspaceDir(dbspace_path, tspid, dbid, op); } FreeDir(ts_dir); @@ -146,125 +179,226 @@ ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op) * Process one per-dbspace directory for ResetUnloggedRelations */ static void -ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) +ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, + Oid tspid, Oid dbid, 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); + Assert((op & (UNLOGGED_RELATION_CLEANUP | + UNLOGGED_RELATION_DROP_BUFFER | + UNLOGGED_RELATION_INIT)) != 0); /* * Cleanup is a two-pass operation. First, we go through and identify all * 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("relfilenode cleanup hash", + 32, &ctl, HASH_ELEM | HASH_BLOBS); + + /* Collect INIT and CLEANUP 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; + + if (forkNum == INIT_FORKNUM || + forkNum == CLEANUP_FORKNUM || forkNum == CLEANUP2_FORKNUM) + { + Oid key; + relfile_entry *ent; + bool found; + + /* + * Record the relfilenode information. If it has the CLEANUP fork, + * the relfilenode is in dirty state, where clean up is needed. + */ + key = atooid(de->d_name); + ent = hash_search(hash, &key, HASH_ENTER, &found); + + if (!found) + { + ent->has_init = false; + ent->dirty_init = false; + ent->dirty_all = false; + } + + if (forkNum == CLEANUP_FORKNUM) + ent->dirty_init = true; + else if (forkNum == CLEANUP2_FORKNUM) + ent->dirty_all = true; + else + { + Assert(forkNum == INIT_FORKNUM); + ent->has_init = true; + } + } + } + + /* Done with the first pass. */ + FreeDir(dbspace_dir); + + /* nothing to do if we don't have init nor cleanup forks */ + if (hash_get_num_entries(hash) < 1) + { + hash_destroy(hash); + return; + } + + if ((op & UNLOGGED_RELATION_DROP_BUFFER) != 0) + { + /* + * When we come here after recovery, smgr object for this file might + * have been created. In that case we need to drop all buffers then the + * smgr object before initializing the unlogged relation. This is safe + * as far as no other backends have accessed the relation before + * starting archive recovery. + */ + HASH_SEQ_STATUS status; + relfile_entry *ent; + SMgrRelation *srels = palloc(sizeof(SMgrRelation) * 8); + int maxrels = 8; + int nrels = 0; + int i; + + Assert(!HotStandbyActive()); + + hash_seq_init(&status, hash); + while((ent = (relfile_entry *) hash_seq_search(&status)) != NULL) + { + RelFileNodeBackend rel; + + /* + * The relation is persistent and stays remain persistent. Don't + * drop the buffers for this relation. + */ + if (ent->has_init && ent->dirty_init) + continue; + + if (maxrels <= nrels) + { + maxrels *= 2; + srels = repalloc(srels, sizeof(SMgrRelation) * maxrels); + } + + rel.backend = InvalidBackendId; + rel.node.spcNode = tspid; + rel.node.dbNode = dbid; + rel.node.relNode = ent->reloid; + + srels[nrels++] = smgropen(rel.node, InvalidBackendId); + } + + DropRelFileNodesAllBuffers(srels, nrels); + + for (i = 0 ; i < nrels ; i++) + smgrclose(srels[i]); + } + + /* + * Now, make a second pass and remove anything that matches. + */ 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; + RelFileNodeBackend rel; /* 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); + + if (!ent) + continue; + + if (!ent->dirty_all) { - 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))); + /* clean permanent relations don't need cleanup */ + if (!ent->has_init) + continue; + + if (ent->dirty_init) + { + /* + * The crashed trasaction did SET UNLOGGED. This relation + * is restored to a LOGGED relation. + */ + if (forkNum != INIT_FORKNUM && forkNum != CLEANUP_FORKNUM) + continue; + } else - elog(DEBUG2, "unlinked file \"%s\"", rm_path); + { + /* + * we don't remove the INIT fork of a non-dirty + * relfilenode + */ + if (forkNum == INIT_FORKNUM) + continue; + } } + + /* so, nuke it! */ + 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))); + + rel.backend = InvalidBackendId; + rel.node.spcNode = tspid; + rel.node.dbNode = dbid; + rel.node.relNode = atooid(de->d_name); + + ForgetRelationForkSyncRequests(rel, forkNum); } /* Cleanup is complete. */ FreeDir(dbspace_dir); - hash_destroy(hash); } + hash_destroy(hash); + hash = NULL; + /* * Initialization happens after cleanup is complete: we copy each init - * fork file to the corresponding main fork file. Note that if we are - * asked to do both cleanup and init, we may never get here: if the - * cleanup code determines that there are no init forks in this dbspace, - * it will return before we get to this point. + * fork file to the corresponding main fork file. */ if ((op & UNLOGGED_RELATION_INIT) != 0) { diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c index 0643d714fb..6b37195c52 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 cleanup fork leads to a data loss. */ if (ret < 0 && errno != ENOENT) - ereport(WARNING, + ereport((forkNum != CLEANUP_FORKNUM ? WARNING : ERROR), (errcode_for_file_access(), errmsg("could not remove file \"%s\": %m", path))); } @@ -1024,6 +1026,15 @@ register_forget_request(RelFileNodeBackend rnode, ForkNumber forknum, RegisterSyncRequest(&tag, SYNC_FORGET_REQUEST, true /* retryOnError */ ); } +/* + * ForgetRelationForkSyncRequests -- forget any fsyncs and unlinks for a fork + */ +void +ForgetRelationForkSyncRequests(RelFileNodeBackend rnode, ForkNumber forknum) +{ + register_forget_request(rnode, forknum, 0); +} + /* * ForgetDatabaseSyncRequests -- forget any fsyncs and unlinks for a DB */ diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c index 4dc24649df..96480e321d 100644 --- a/src/backend/storage/smgr/smgr.c +++ b/src/backend/storage/smgr/smgr.c @@ -662,6 +662,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..479dcc248e 100644 --- a/src/common/relpath.c +++ b/src/common/relpath.c @@ -34,7 +34,9 @@ const char *const forkNames[] = { "main", /* MAIN_FORKNUM */ "fsm", /* FSM_FORKNUM */ "vm", /* VISIBILITYMAP_FORKNUM */ - "init" /* INIT_FORKNUM */ + "init", /* INIT_FORKNUM */ + "clup", /* CLEANUP_FORKNUM */ + "cln2" /* CLEANUP2_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..040070aa2b 100644 --- a/src/include/common/relpath.h +++ b/src/include/common/relpath.h @@ -43,7 +43,9 @@ typedef enum ForkNumber MAIN_FORKNUM = 0, FSM_FORKNUM, VISIBILITYMAP_FORKNUM, - INIT_FORKNUM + INIT_FORKNUM, + CLEANUP_FORKNUM, + CLEANUP2_FORKNUM /* * NOTE: if you add a new fork, change MAX_FORKNUM and possibly @@ -52,7 +54,7 @@ typedef enum ForkNumber */ } ForkNumber; -#define MAX_FORKNUM INIT_FORKNUM +#define MAX_FORKNUM CLEANUP2_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 fb00fda6a7..ccb0a388f6 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(struct SMgrRelationData *smgr_reln, ForkNumber *forkNum, int nforks, BlockNumber *firstDelBlock); +extern void SetRelationBuffersPersistence(struct SMgrRelationData *srel, + bool permanent, bool isRedo); extern void DropRelFileNodesAllBuffers(struct SMgrRelationData **smgr_reln, int nnodes); extern void DropDatabaseBuffers(Oid dbid); diff --git a/src/include/storage/md.h b/src/include/storage/md.h index 752b440864..3cbbbf2edd 100644 --- a/src/include/storage/md.h +++ b/src/include/storage/md.h @@ -41,6 +41,8 @@ extern void mdtruncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks); extern void mdimmedsync(SMgrRelation reln, ForkNumber forknum); +extern void ForgetRelationForkSyncRequests(RelFileNodeBackend rnode, + ForkNumber forknum); extern void ForgetDatabaseSyncRequests(Oid dbid); extern void DropRelationFiles(RelFileNode *delrels, int ndelrels, bool isRedo); diff --git a/src/include/storage/reinit.h b/src/include/storage/reinit.h index fad1e5c473..b969ba8e86 100644 --- a/src/include/storage/reinit.h +++ b/src/include/storage/reinit.h @@ -23,6 +23,7 @@ extern bool parse_filename_for_nontemp_relation(const char *name, int *oidchars, ForkNumber *fork); #define UNLOGGED_RELATION_CLEANUP 0x0001 -#define UNLOGGED_RELATION_INIT 0x0002 +#define UNLOGGED_RELATION_DROP_BUFFER 0x0002 +#define UNLOGGED_RELATION_INIT 0x0004 #endif /* REINIT_H */ diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h index a6fbf7b6a6..1ac3e4a74a 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(Thu_Jan_14_17_32_17_2021_289)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v5-0002-New-command-ALTER-TABLE-ALL-IN-TABLESPACE-SET-LOG.patch" ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Function scan FDW pushdown @ 2024-11-05 16:11 [email protected] 0 siblings, 1 reply; 3+ messages in thread From: [email protected] @ 2024-11-05 16:11 UTC (permalink / raw) To: Alexander Pyhalov <[email protected]>; +Cc: Ashutosh Bapat <[email protected]>; Ashutosh Bapat <[email protected]>; PostgreSQL Hackers <[email protected]> Alexander Pyhalov писал(а) 2021-10-04 10:42: > Ashutosh Bapat писал 2021-06-15 16:15: >> Hi Alexander, > > Hi. > > The current version of the patch is based on asymetric partition-wise > join. > Currently it is applied after > v19-0001-Asymmetric-partitionwise-join.patch from > on > https://www.postgresql.org/message-id/[email protected] > . > >>> So far I don't know how to visualize actual function expression used >>> in >>> function RTE, as in postgresExplainForeignScan() es->rtable comes >>> from >>> queryDesc->plannedstmt->rtable, and rte->functions is already 0. >> >> The actual function expression will be part of the Remote SQL of >> ForeignScan node so no need to visualize it separately. > > We still need to create tuple description for functions in > get_tupdesc_for_join_scan_tuples(), > so I had to remove setting newrte->functions to NIL in > add_rte_to_flat_rtable(). > With rte->functions in place, there's no issues for explain. > >> >> The patch will have problems when there are multiple foreign tables >> all on different servers or use different FDWs. In such a case the >> function scan's RelOptInfo will get the fpinfo based on the first >> foreign table the function scan is paired with during join planning. >> But that may not be the best foreign table to join. We should be able >> to plan all the possible joins. Current infra to add one fpinfo per >> RelOptInfo won't help there. We need something better. > > I suppose attached version of the patch is more mature. > >> >> The patch targets only postgres FDW, how do you see this working with >> other FDWs? > > Not now. We introduce necessary APIs for other FDWs, but implementing > TryShippableJoinPaths() > doesn't seem straightforward. > >> >> If we come up with the right approach we could use it for 1. pushing >> down queries with IN () clause 2. joining a small local table with a >> large foreign table by sending the local table rows down to the >> foreign server. Hi, This is a long-overdue follow-up to the original patch. Note that this patch contains only the changes required for function scan pushdown, examples and code related to asymmetric join are dropped. In this version, we have fixed some of the issues: 1) Functions returning records are now deparsed correctly 2) I have benchmarked this version alongside the current master using HammerDB and it shows a consistent ~6000 number of orders per minute (NOPM) compared to ~100 NOPM on local setup with 1 virtual user and 10 warehouses The benchmarking was concluded in the following way: 1) schema was created and filled up by buildschema on the first instance 2) foreign schema was imported on the second instance 3) functions created by buildschema (all in namespace 'public') were exported from the first instance and created on the second one 4) HammerDB was connected to the second instance, and the benchmark was run there Note that the HammerDB service functions that were imported to the second instance are not the ones being pushed down, only PostgreSQL built-ins are. The issue with setting newrte->functions to NIL still persists. I've attempted to work around it by adding the rte->functions list to fdw_private field. I stored a copy of rte->functions for each RTE in the order they are listed in simple_rte_array and accessed it where the rte->functions were used in the original patch. While this being pretty straightforward, I found out the hard way that the RTEs are flattened and the original sequential order known at postgresGetForeignPlan() and postgresPlanDirectModify() is altered randomly. It prevents me from looking up the right functions list for the RTE by its rti in postgresExplainForeignScan() and its var->varno in get_tupdesc_for_join_scan_tuples(). I am aware that the rte->functions will now be copied even on instances that don't utilize a FDW, but I don't see a way to solve it. Any suggestions are welcome. Best regards, Gleb Kashkin, Postgres Professional Attachments: [text/x-diff] 0001-Push-join-with-function-scan-to-remote-server.patch (52.0K, ../../[email protected]/2-0001-Push-join-with-function-scan-to-remote-server.patch) download | inline diff: From a59b04e33147a075fdd4c6cef155440689850b83 Mon Sep 17 00:00:00 2001 From: Alexander Pyhalov <[email protected]> Date: Mon, 17 May 2021 19:19:31 +0300 Subject: [PATCH] Push join with function scan to remote server The patch allows pushing joins with function RTEs to PostgreSQL data sources in general and postgres_fdw specifically. Co-authored-by: Gleb Kashkin <[email protected]> --- contrib/postgres_fdw/deparse.c | 187 ++++++-- .../postgres_fdw/expected/postgres_fdw.out | 368 +++++++++++++++ contrib/postgres_fdw/postgres_fdw.c | 423 +++++++++++++++--- contrib/postgres_fdw/postgres_fdw.h | 6 + contrib/postgres_fdw/sql/postgres_fdw.sql | 148 ++++++ src/backend/optimizer/path/joinpath.c | 11 + src/backend/optimizer/plan/setrefs.c | 1 - src/include/foreign/fdwapi.h | 1 + src/test/regress/expected/create_view.out | 6 +- 9 files changed, 1046 insertions(+), 105 deletions(-) diff --git a/contrib/postgres_fdw/deparse.c b/contrib/postgres_fdw/deparse.c index 4680d51733..17228dec47 100644 --- a/contrib/postgres_fdw/deparse.c +++ b/contrib/postgres_fdw/deparse.c @@ -153,6 +153,7 @@ static void deparseConst(Const *node, deparse_expr_cxt *context, int showtype); static void deparseParam(Param *node, deparse_expr_cxt *context); static void deparseSubscriptingRef(SubscriptingRef *node, deparse_expr_cxt *context); static void deparseFuncExpr(FuncExpr *node, deparse_expr_cxt *context); +static void deparseFuncColnames(StringInfo buf, int varno, RangeTblEntry *rte, bool qualify_col); static void deparseOpExpr(OpExpr *node, deparse_expr_cxt *context); static bool isPlainForeignVar(Expr *node, deparse_expr_cxt *context); static void deparseOperatorName(StringInfo buf, Form_pg_operator opform); @@ -1977,13 +1978,54 @@ deparseFromExprForRel(StringInfo buf, PlannerInfo *root, RelOptInfo *foreignrel, { RangeTblEntry *rte = planner_rt_fetch(foreignrel->relid, root); - /* - * Core code already has some lock on each rel being planned, so we - * can use NoLock here. - */ - Relation rel = table_open(rte->relid, NoLock); + Assert(rte->rtekind == RTE_RELATION || rte->rtekind == RTE_FUNCTION); + if (rte->rtekind == RTE_RELATION) + { + /* + * Core code already has some lock on each rel being planned, so + * we can use NoLock here. + */ + Relation rel = table_open(rte->relid, NoLock); - deparseRelation(buf, rel); + deparseRelation(buf, rel); + + table_close(rel, NoLock); + } + else if (rte->rtekind == RTE_FUNCTION) + { + RangeTblFunction *rtfunc; + deparse_expr_cxt context; + ListCell *lc; + bool first = true; + int n; + + n = list_length(rte->functions); + Assert(n >= 1); + + if (n > 1) + appendStringInfoString(buf, "ROWS FROM ("); + + foreach(lc, rte->functions) + { + if (!first) + appendStringInfoString(buf, ", "); + else + first = false; + + rtfunc = (RangeTblFunction *) lfirst(lc); + + context.root = root; + context.foreignrel = foreignrel; + context.scanrel = foreignrel; + context.buf = buf; + context.params_list = params_list; + + deparseExpr((Expr *) rtfunc->funcexpr, &context); + } + + if (n > 1) + appendStringInfoString(buf, ")"); + } /* * Add a unique alias to avoid any conflict in relation names due to @@ -1991,9 +2033,43 @@ deparseFromExprForRel(StringInfo buf, PlannerInfo *root, RelOptInfo *foreignrel, * join. */ if (use_alias) + { appendStringInfo(buf, " %s%d", REL_ALIAS_PREFIX, foreignrel->relid); + if (rte->rtekind == RTE_FUNCTION) + { + appendStringInfo(buf, " ("); + deparseFuncColnames(buf, 0, rte, false); + appendStringInfo(buf, ") "); + } + } + } +} - table_close(rel, NoLock); +/* + * Deparse function columns alias list + */ +static void +deparseFuncColnames(StringInfo buf, int varno, RangeTblEntry *rte, bool qualify_col) +{ + bool first = true; + ListCell *lc; + + Assert(rte); + Assert(rte->rtekind == RTE_FUNCTION); + Assert(rte->eref); + + foreach(lc, rte->eref->colnames) + { + char *colname = strVal(lfirst(lc)); + + if (colname[0] == '\0') + continue; + if (!first) + appendStringInfoString(buf, ","); + if (qualify_col) + ADD_REL_QUALIFIER(buf, varno); + appendStringInfoString(buf, quote_identifier(colname)); + first = false; } } @@ -2717,23 +2793,6 @@ deparseColumnRef(StringInfo buf, int varno, int varattno, RangeTblEntry *rte, /* Required only to be passed down to deparseTargetList(). */ List *retrieved_attrs; - /* - * The lock on the relation will be held by upper callers, so it's - * fine to open it with no lock here. - */ - rel = table_open(rte->relid, NoLock); - - /* - * The local name of the foreign table can not be recognized by the - * foreign server and the table it references on foreign server might - * have different column ordering or different columns than those - * declared locally. Hence we have to deparse whole-row reference as - * ROW(columns referenced locally). Construct this by deparsing a - * "whole row" attribute. - */ - attrs_used = bms_add_member(NULL, - 0 - FirstLowInvalidHeapAttributeNumber); - /* * In case the whole-row reference is under an outer join then it has * to go NULL whenever the rest of the row goes NULL. Deparsing a join @@ -2748,16 +2807,43 @@ deparseColumnRef(StringInfo buf, int varno, int varattno, RangeTblEntry *rte, } appendStringInfoString(buf, "ROW("); - deparseTargetList(buf, rte, varno, rel, false, attrs_used, qualify_col, - &retrieved_attrs); + if (rte->rtekind == RTE_RELATION) + { + /* + * The local name of the foreign table can not be recognized by + * the foreign server and the table it references on foreign + * server might have different column ordering or different + * columns than those declared locally. Hence we have to deparse + * whole-row reference as ROW(columns referenced locally). + * Construct this by deparsing a "whole row" attribute. + */ + attrs_used = bms_add_member(NULL, + 0 - FirstLowInvalidHeapAttributeNumber); + + /* + * The lock on the relation will be held by upper callers, so it's + * fine to open it with no lock here. + */ + rel = table_open(rte->relid, NoLock); + deparseTargetList(buf, rte, varno, rel, false, attrs_used, qualify_col, + &retrieved_attrs); + table_close(rel, NoLock); + bms_free(attrs_used); + } + else if (rte->rtekind == RTE_FUNCTION) + { + /* + * Function call is translated as-is, function returns the same + * columns in the same order as on local server + */ + deparseFuncColnames(buf, varno, rte, qualify_col); + } appendStringInfoChar(buf, ')'); /* Complete the CASE WHEN statement started above. */ if (qualify_col) appendStringInfoString(buf, " END"); - table_close(rel, NoLock); - bms_free(attrs_used); } else { @@ -2772,29 +2858,40 @@ deparseColumnRef(StringInfo buf, int varno, int varattno, RangeTblEntry *rte, * If it's a column of a foreign table, and it has the column_name FDW * option, use that value. */ - options = GetForeignColumnOptions(rte->relid, varattno); - foreach(lc, options) + if (rte->rtekind == RTE_RELATION) { - DefElem *def = (DefElem *) lfirst(lc); - - if (strcmp(def->defname, "column_name") == 0) + options = GetForeignColumnOptions(rte->relid, varattno); + foreach(lc, options) { - colname = defGetString(def); - break; + DefElem *def = (DefElem *) lfirst(lc); + + if (strcmp(def->defname, "column_name") == 0) + { + colname = defGetString(def); + break; + } } - } - /* - * If it's a column of a regular table or it doesn't have column_name - * FDW option, use attribute name. - */ - if (colname == NULL) - colname = get_attname(rte->relid, varattno, false); + /* + * If it's a column of a regular table or it doesn't have + * column_name FDW option, use attribute name. + */ + if (colname == NULL) + colname = get_attname(rte->relid, varattno, false); - if (qualify_col) - ADD_REL_QUALIFIER(buf, varno); + if (qualify_col) + ADD_REL_QUALIFIER(buf, varno); - appendStringInfoString(buf, quote_identifier(colname)); + appendStringInfoString(buf, quote_identifier(colname)); + } + else if (rte->rtekind == RTE_FUNCTION) + { + colname = get_rte_attribute_name(rte, varattno); + + if (qualify_col) + ADD_REL_QUALIFIER(buf, varno); + appendStringInfoString(buf, quote_identifier(colname)); + } } } diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index f2bcd6aa98..686ce1ac61 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -12345,3 +12345,371 @@ SELECT CASE WHEN closed IS NOT false THEN 1 ELSE 0 END -- Clean up \set VERBOSITY default RESET debug_discard_caches; +-- =================================================================== +-- test function scan pushdown +-- =================================================================== +CREATE TABLE base_tbl (a int, b int); +CREATE FOREIGN TABLE remote_tbl (a int, b int) + SERVER loopback OPTIONS (table_name 'base_tbl'); +ALTER FOREIGN TABLE remote_tbl OPTIONS (use_remote_estimate 'true'); +CREATE TABLE base_tbl1 (c int, d text); +CREATE FOREIGN TABLE remote_tbl1 (c int, d text) + SERVER loopback OPTIONS (table_name 'base_tbl1'); +ALTER FOREIGN TABLE remote_tbl1 OPTIONS (use_remote_estimate 'true'); +INSERT INTO remote_tbl SELECT g, g*2 from generate_series(1,1000) g; +INSERT INTO remote_tbl1 SELECT g, 'text'|| g from generate_series(1,500) g; +ANALYZE base_tbl; +ANALYZE base_tbl1; +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl r, unnest(array[2,3,4]) n WHERE r.a = n; + QUERY PLAN +-------------------------------------------------------------------------------------------------------------------------------------- + Foreign Scan + Output: r.a, r.b, n.n + Relations: (public.remote_tbl r) INNER JOIN (pg_catalog.unnest() n) + Remote SQL: SELECT r1.a, r1.b, r2.n FROM (public.base_tbl r1 INNER JOIN unnest('{2,3,4}'::integer[]) r2 (n) ON (((r1.a = r2.n)))) +(4 rows) + +SELECT * FROM remote_tbl r, unnest(array[2,3,4]) n WHERE r.a = n +ORDER BY r.a; + a | b | n +---+---+--- + 2 | 4 | 2 + 3 | 6 | 3 + 4 | 8 | 4 +(3 rows) + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM unnest(array[2,3,4]) n, remote_tbl r WHERE r.a = n; + QUERY PLAN +-------------------------------------------------------------------------------------------------------------------------------------- + Foreign Scan + Output: n.n, r.a, r.b + Relations: (public.remote_tbl r) INNER JOIN (pg_catalog.unnest() n) + Remote SQL: SELECT r1.n, r2.a, r2.b FROM (public.base_tbl r2 INNER JOIN unnest('{2,3,4}'::integer[]) r1 (n) ON (((r2.a = r1.n)))) +(4 rows) + +SELECT * FROM unnest(array[2,3,4]) n, remote_tbl r WHERE r.a = n +ORDER BY r.a; + n | a | b +---+---+--- + 2 | 2 | 4 + 3 | 3 | 6 + 4 | 4 | 8 +(3 rows) + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a; + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ + Foreign Scan + Output: r.a, r.b, r1.c, r1.d, n.n + Relations: ((public.remote_tbl r) INNER JOIN (public.remote_tbl1 r1)) INNER JOIN (pg_catalog.unnest() n) + Remote SQL: SELECT r1.a, r1.b, r2.c, r2.d, r3.n FROM ((public.base_tbl r1 INNER JOIN public.base_tbl1 r2 ON (((r1.a = r2.c)))) INNER JOIN unnest('{3,4}'::integer[]) r3 (n) ON (((r1.a = r3.n)))) +(4 rows) + +SELECT * FROM remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a +ORDER BY r.a; + a | b | c | d | n +---+---+---+-------+--- + 3 | 6 | 3 | text3 | 3 + 4 | 8 | 4 | text4 | 4 +(2 rows) + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT r.*,n from remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a and n > 3; + QUERY PLAN +----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Foreign Scan + Output: r.a, r.b, n.n + Relations: ((public.remote_tbl r) INNER JOIN (public.remote_tbl1 r1)) INNER JOIN (pg_catalog.unnest() n) + Remote SQL: SELECT r1.a, r1.b, r3.n FROM ((public.base_tbl r1 INNER JOIN public.base_tbl1 r2 ON (((r1.a = r2.c)))) INNER JOIN unnest('{3,4}'::integer[]) r3 (n) ON (((r1.a = r3.n)) AND ((r3.n > 3)))) +(4 rows) + +SELECT * from remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a and n > 3; + a | b | c | d | n +---+---+---+-------+--- + 4 | 8 | 4 | text4 | 4 +(1 row) + +CREATE OR REPLACE FUNCTION get_constant_texts() +RETURNS TABLE (text_value text) AS $$ +BEGIN + RETURN QUERY VALUES + ('text1'), + ('text4'); +END; +$$ LANGUAGE plpgsql IMMUTABLE; +ALTER EXTENSION postgres_fdw ADD FUNCTION get_constant_texts(); +ALTER SERVER loopback OPTIONS (extensions 'postgres_fdw'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT r.*, t.n from remote_tbl1 r, ROWS FROM (unnest(array[3,4]), get_constant_texts()) t (n, txt) +WHERE r.c = t.n AND r.d = t.txt; + QUERY PLAN +-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Foreign Scan + Output: r.c, r.d, t.n + Relations: (public.remote_tbl1 r) INNER JOIN (ROWS FROM(pg_catalog.unnest(), public.get_constant_texts()) t) + Remote SQL: SELECT r1.c, r1.d, r2.n FROM (public.base_tbl1 r1 INNER JOIN ROWS FROM (unnest('{3,4}'::integer[]), public.get_constant_texts()) r2 (n,txt) ON (((r1.c = r2.n)) AND ((r2.txt = r1.d)))) +(4 rows) + +SELECT r.*, t.txt from remote_tbl1 r, ROWS FROM (unnest(array[3,4]), get_constant_texts()) t (n, txt) +WHERE r.c = t.n AND r.d = t.txt; + c | d | txt +---+-------+------- + 4 | text4 | text4 +(1 row) + +-- Complex types +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl r JOIN UNNEST(array[box '((2,3),(-2,-3))']) as t(bx) ON a = area(bx); + QUERY PLAN +--------------------------------------------------------------------------------------------------------------------------------------------------- + Foreign Scan + Output: r.a, r.b, t.bx + Relations: (public.remote_tbl r) INNER JOIN (pg_catalog.unnest() t) + Remote SQL: SELECT r1.a, r1.b, r2.bx FROM (public.base_tbl r1 INNER JOIN unnest('{(2,3),(-2,-3)}'::box[]) r2 (bx) ON (((r1.a = area(r2.bx))))) +(4 rows) + +SELECT * FROM remote_tbl r JOIN UNNEST(array[box '((2,3),(-2,-3))']) as t(bx) ON a = area(bx) +ORDER BY r.a; + a | b | bx +----+----+--------------- + 24 | 48 | (2,3),(-2,-3) +(1 row) + +-- DML +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE remote_tbl r SET b=5 FROM UNNEST(array[box '((2,3),(-2,-3))']) AS t (bx) WHERE r.a = area(t.bx) +RETURNING a,b; + QUERY PLAN +---------------------------------------------------------------------------------------------------------------------------------------------------------- + Update on public.remote_tbl r + Output: r.a, r.b + -> Foreign Update + Remote SQL: UPDATE public.base_tbl r1 SET b = 5 FROM unnest('{(2,3),(-2,-3)}'::box[]) r2 (bx) WHERE ((r1.a = area(r2.bx))) RETURNING r1.a, r1.b +(4 rows) + +UPDATE remote_tbl r SET b=5 FROM UNNEST(array[box '((2,3),(-2,-3))']) AS t (bx) WHERE r.a = area(t.bx) +RETURNING a,b; + a | b +----+--- + 24 | 5 +(1 row) + +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE remote_tbl r SET b=CASE WHEN random()>=0 THEN 5 ELSE 0 END FROM UNNEST(array[box '((2,3),(-2,-3))']) AS t (bx) WHERE r.a = area(t.bx) +RETURNING a,b; + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Update on public.remote_tbl r + Output: r.a, r.b + Remote SQL: UPDATE public.base_tbl SET b = $2 WHERE ctid = $1 RETURNING a, b + -> Foreign Scan + Output: CASE WHEN (random() >= '0'::double precision) THEN 5 ELSE 0 END, r.ctid, r.*, t.* + Relations: (public.remote_tbl r) INNER JOIN (pg_catalog.unnest() t) + Remote SQL: SELECT r1.ctid, CASE WHEN (r1.*)::text IS NOT NULL THEN ROW(r1.a, r1.b) END, CASE WHEN (r2.*)::text IS NOT NULL THEN ROW(r2.bx) END FROM (public.base_tbl r1 INNER JOIN unnest('{(2,3),(-2,-3)}'::box[]) r2 (bx) ON (((r1.a = area(r2.bx))))) FOR UPDATE OF r1 + -> Hash Join + Output: r.ctid, r.*, t.* + Hash Cond: ((r.a)::double precision = area(t.bx)) + -> Foreign Scan on public.remote_tbl r + Output: r.ctid, r.*, r.a + Remote SQL: SELECT a, b, ctid FROM public.base_tbl FOR UPDATE + -> Hash + Output: t.*, t.bx + -> Function Scan on pg_catalog.unnest t + Output: t.*, t.bx + Function Call: unnest('{(2,3),(-2,-3)}'::box[]) +(18 rows) + +UPDATE remote_tbl r SET b=CASE WHEN random()>=0 THEN 5 ELSE 0 END FROM UNNEST(array[box '((2,3),(-2,-3))']) AS t (bx) WHERE r.a = area(t.bx) +RETURNING a,b; + a | b +----+--- + 24 | 5 +(1 row) + +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE remote_tbl r SET b=5 FROM UNNEST(array[10,20], array[box '((2,3),(-2,-4))', box '((1,2),(-2,-3))']) AS t (l, bx) WHERE r.a between l and area(t.bx) +RETURNING a,b; + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ + Update on public.remote_tbl r + Output: r.a, r.b + -> Foreign Update + Remote SQL: UPDATE public.base_tbl r1 SET b = 5 FROM ROWS FROM (unnest('{10,20}'::integer[]), unnest('{(2,3),(-2,-4);(1,2),(-2,-3)}'::box[])) r2 (l,bx) WHERE ((r1.a >= r2.l)) AND ((r1.a <= area(r2.bx))) RETURNING r1.a, r1.b +(4 rows) + +UPDATE remote_tbl r SET b=5 FROM UNNEST(array[10,20], array[box '((2,3),(-2,-4))', box '((1,2),(-2,-3))']) AS t (l, bx) WHERE r.a between l and area(t.bx) +RETURNING a,b; + a | b +----+--- + 10 | 5 + 11 | 5 + 12 | 5 + 13 | 5 + 14 | 5 + 15 | 5 + 16 | 5 + 17 | 5 + 18 | 5 + 19 | 5 + 20 | 5 + 21 | 5 + 22 | 5 + 23 | 5 + 25 | 5 + 26 | 5 + 27 | 5 + 28 | 5 + 24 | 5 +(19 rows) + +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE remote_tbl r SET b=CASE WHEN random()>=0 THEN 5 ELSE 0 END FROM UNNEST(array[10,20], array[box '((2,3),(-2,-4))', box '((1,2),(-2,-3))']) AS t (l, bx) WHERE r.a between l and area(t.bx) +RETURNING a,b; + QUERY PLAN +-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Update on public.remote_tbl r + Output: r.a, r.b + Remote SQL: UPDATE public.base_tbl SET b = $2 WHERE ctid = $1 RETURNING a, b + -> Foreign Scan + Output: CASE WHEN (random() >= '0'::double precision) THEN 5 ELSE 0 END, r.ctid, r.*, t.* + Relations: (public.remote_tbl r) INNER JOIN (ROWS FROM(pg_catalog.unnest(), pg_catalog.unnest()) t) + Remote SQL: SELECT r1.ctid, CASE WHEN (r1.*)::text IS NOT NULL THEN ROW(r1.a, r1.b) END, CASE WHEN (r2.*)::text IS NOT NULL THEN ROW(r2.l,r2.bx) END FROM (public.base_tbl r1 INNER JOIN ROWS FROM (unnest('{10,20}'::integer[]), unnest('{(2,3),(-2,-4);(1,2),(-2,-3)}'::box[])) r2 (l,bx) ON (((r1.a >= r2.l)) AND ((r1.a <= area(r2.bx))))) FOR UPDATE OF r1 + -> Nested Loop + Output: r.ctid, r.*, t.* + Join Filter: ((r.a >= t.l) AND ((r.a)::double precision <= area(t.bx))) + -> Foreign Scan on public.remote_tbl r + Output: r.ctid, r.*, r.a + Remote SQL: SELECT a, b, ctid FROM public.base_tbl FOR UPDATE + -> Function Scan on t + Output: t.*, t.l, t.bx + Function Call: unnest('{10,20}'::integer[]), unnest('{(2,3),(-2,-4);(1,2),(-2,-3)}'::box[]) +(16 rows) + +UPDATE remote_tbl r SET b=CASE WHEN random()>=0 THEN 5 ELSE 0 END FROM UNNEST(array[10,20], array[box '((2,3),(-2,-4))', box '((1,2),(-2,-3))']) AS t (l, bx) WHERE r.a between l and area(t.bx) +RETURNING a,b; + a | b +----+--- + 10 | 5 + 11 | 5 + 12 | 5 + 13 | 5 + 14 | 5 + 15 | 5 + 16 | 5 + 17 | 5 + 18 | 5 + 19 | 5 + 20 | 5 + 21 | 5 + 22 | 5 + 23 | 5 + 25 | 5 + 26 | 5 + 27 | 5 + 28 | 5 + 24 | 5 +(19 rows) + +-- Test that local functions are not pushed down +CREATE OR REPLACE FUNCTION f(INTEGER) +RETURNS SETOF INTEGER +LANGUAGE sql AS 'select generate_series(1,$1);' IMMUTABLE; +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl r, f(10) n +WHERE r.a = n; + QUERY PLAN +------------------------------------------------------ + Hash Join + Output: r.a, r.b, (generate_series(1, 10)) + Hash Cond: (r.a = (generate_series(1, 10))) + -> Foreign Scan on public.remote_tbl r + Output: r.a, r.b + Remote SQL: SELECT a, b FROM public.base_tbl + -> Hash + Output: (generate_series(1, 10)) + -> ProjectSet + Output: generate_series(1, 10) + -> Result +(11 rows) + +SELECT * FROM remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a AND n > 3; + a | b | c | d | n +---+---+---+-------+--- + 4 | 8 | 4 | text4 | 4 +(1 row) + +-- Test that a function that returns a record is not pushed down +CREATE OR REPLACE FUNCTION f_ret_record() RETURNS record AS $$ SELECT (1,2)::record $$ language SQL IMMUTABLE; +ALTER EXTENSION postgres_fdw ADD function f_ret_record(); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT s FROM remote_tbl rt, f_ret_record() AS s(a int, b int) +WHERE s.a = rt.a; + QUERY PLAN +----------------------------------------------------------------------------- + Nested Loop + Output: s.* + -> Function Scan on public.f_ret_record s + Output: s.*, s.a + Function Call: f_ret_record() + -> Foreign Scan on public.remote_tbl rt + Output: rt.a, rt.b + Remote SQL: SELECT a FROM public.base_tbl WHERE ((a = $1::integer)) +(8 rows) + +SELECT s FROM remote_tbl rt, f_ret_record() AS s(a int, b int) +WHERE s.a = rt.a; + s +------- + (1,2) +(1 row) + +DROP FUNCTION f(INTEGER); +ALTER EXTENSION postgres_fdw DROP FUNCTION get_constant_texts(); +ALTER EXTENSION postgres_fdw DROP FUNCTION f_ret_record(); +DROP FUNCTION get_constant_texts(); +DROP FUNCTION f_ret_record(); +DROP TABLE base_tbl, base_tbl1; +DROP FOREIGN TABLE remote_tbl, remote_tbl1; +-- Test that function WITH ORDINALITY is not pushed down +CREATE TABLE base_tbl (a int, b int); +CREATE FOREIGN TABLE remote_tbl (a int, b int) SERVER loopback OPTIONS (table_name 'base_tbl'); +INSERT INTO remote_tbl VALUES (1, 2), (2, 3), (3, 4), (5, 6); +ANALYZE remote_tbl; +SET enable_mergejoin TO false; +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl, unnest(ARRAY[1, 2]) WITH ORDINALITY +WHERE a=unnest; + QUERY PLAN +------------------------------------------------------------------------ + Hash Join + Output: remote_tbl.a, remote_tbl.b, unnest.unnest, unnest.ordinality + Hash Cond: (remote_tbl.a = unnest.unnest) + -> Foreign Scan on public.remote_tbl + Output: remote_tbl.a, remote_tbl.b + Remote SQL: SELECT a, b FROM public.base_tbl + -> Hash + Output: unnest.unnest, unnest.ordinality + -> Function Scan on pg_catalog.unnest + Output: unnest.unnest, unnest.ordinality + Function Call: unnest('{1,2}'::integer[]) +(11 rows) + +SELECT * FROM remote_tbl, unnest(ARRAY[1, 2]) WITH ORDINALITY +WHERE a=unnest; + a | b | unnest | ordinality +---+---+--------+------------ + 1 | 2 | 1 | 1 + 2 | 3 | 2 | 2 +(2 rows) + +DROP TABLE base_tbl; +DROP FOREIGN TABLE remote_tbl; +RESET enable_mergejoin; diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 53733d642d..5d0a339ca0 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -36,6 +36,7 @@ #include "optimizer/prep.h" #include "optimizer/restrictinfo.h" #include "optimizer/tlist.h" +#include "optimizer/clauses.h" #include "parser/parsetree.h" #include "postgres_fdw.h" #include "storage/latch.h" @@ -406,6 +407,14 @@ static void postgresGetForeignJoinPaths(PlannerInfo *root, RelOptInfo *innerrel, JoinType jointype, JoinPathExtraData *extra); + +static void postgresTryShippableJoinPaths(PlannerInfo *root, + RelOptInfo *joinrel, + RelOptInfo *outerrel, + RelOptInfo *innerrel, + JoinType jointype, + JoinPathExtraData *extra); + static bool postgresRecheckForeignScan(ForeignScanState *node, TupleTableSlot *slot); static void postgresGetForeignUpperPaths(PlannerInfo *root, @@ -476,7 +485,7 @@ static void store_returning_result(PgFdwModifyState *fmstate, static void finish_foreign_modify(PgFdwModifyState *fmstate); static void deallocate_query(PgFdwModifyState *fmstate); static List *build_remote_returning(Index rtindex, Relation rel, - List *returningList); + List *returningList, Var *tid); static void rebuild_fdw_scan_tlist(ForeignScan *fscan, List *tlist); static void execute_dml_stmt(ForeignScanState *node); static TupleTableSlot *get_returning_data(ForeignScanState *node); @@ -542,6 +551,11 @@ static void merge_fdw_options(PgFdwRelationInfo *fpinfo, const PgFdwRelationInfo *fpinfo_i); static int get_batch_size_option(Relation rel); +static bool is_nonrel_relinfo_ok(PlannerInfo *root, RelOptInfo *foreignrel); +static void init_fpinfo(PlannerInfo *root, + RelOptInfo *baserel, + Oid foreigntableid, + PgFdwRelationInfo *existing_fpinfo); /* * Foreign-data wrapper handler function: return a struct with pointers @@ -597,6 +611,7 @@ postgres_fdw_handler(PG_FUNCTION_ARGS) /* Support functions for join push-down */ routine->GetForeignJoinPaths = postgresGetForeignJoinPaths; + routine->TryShippableJoinPaths = postgresTryShippableJoinPaths; /* Support functions for upper relation push-down */ routine->GetForeignUpperPaths = postgresGetForeignUpperPaths; @@ -621,10 +636,32 @@ static void postgresGetForeignRelSize(PlannerInfo *root, RelOptInfo *baserel, Oid foreigntableid) +{ + init_fpinfo(root, baserel, foreigntableid, NULL); +} + +/* + * init_fpinfo + * + * Either initialize fpinfo based on foreign table or generate one, based on + * existing fpinfo. + * Also estimate # of rows and width of the result of the scan. + * + * We should consider the effect of all baserestrictinfo clauses here, but + * not any join clauses. + */ +static void +init_fpinfo(PlannerInfo *root, + RelOptInfo *baserel, + Oid foreigntableid, + PgFdwRelationInfo *existing_fpinfo) { PgFdwRelationInfo *fpinfo; ListCell *lc; + Assert(existing_fpinfo || foreigntableid != InvalidOid); + Assert(existing_fpinfo == NULL || foreigntableid == InvalidOid); + /* * We use PgFdwRelationInfo to pass various information to subsequent * functions. @@ -635,40 +672,64 @@ postgresGetForeignRelSize(PlannerInfo *root, /* Base foreign tables need to be pushed down always. */ fpinfo->pushdown_safe = true; - /* Look up foreign-table catalog info. */ - fpinfo->table = GetForeignTable(foreigntableid); - fpinfo->server = GetForeignServer(fpinfo->table->serverid); - - /* - * Extract user-settable option values. Note that per-table settings of - * use_remote_estimate, fetch_size and async_capable override per-server - * settings of them, respectively. - */ - fpinfo->use_remote_estimate = false; - fpinfo->fdw_startup_cost = DEFAULT_FDW_STARTUP_COST; - fpinfo->fdw_tuple_cost = DEFAULT_FDW_TUPLE_COST; - fpinfo->shippable_extensions = NIL; - fpinfo->fetch_size = 100; - fpinfo->async_capable = false; - - apply_server_options(fpinfo); - apply_table_options(fpinfo); + if (existing_fpinfo) + { + /* We don't have any table, related to query */ + fpinfo->table = NULL; + fpinfo->server = existing_fpinfo->server; + } + else + { + /* Look up foreign-table catalog info. */ + fpinfo->table = GetForeignTable(foreigntableid); + fpinfo->server = GetForeignServer(fpinfo->table->serverid); + } - /* - * If the table or the server is configured to use remote estimates, - * identify which user to do remote access as during planning. This - * should match what ExecCheckPermissions() does. If we fail due to lack - * of permissions, the query would have failed at runtime anyway. - */ - if (fpinfo->use_remote_estimate) + if (existing_fpinfo) { - Oid userid; + merge_fdw_options(fpinfo, existing_fpinfo, NULL); + fpinfo->user = existing_fpinfo->user; - userid = OidIsValid(baserel->userid) ? baserel->userid : GetUserId(); - fpinfo->user = GetUserMapping(userid, fpinfo->server->serverid); + /* + * Don't try to execute anything on remote server for + * non-relation-based query + */ + fpinfo->use_remote_estimate = false; } else - fpinfo->user = NULL; + { + /* + * Extract user-settable option values. Note that per-table settings + * of use_remote_estimate, fetch_size and async_capable override + * per-server settings of them, respectively. + */ + fpinfo->use_remote_estimate = false; + fpinfo->fdw_startup_cost = DEFAULT_FDW_STARTUP_COST; + fpinfo->fdw_tuple_cost = DEFAULT_FDW_TUPLE_COST; + fpinfo->shippable_extensions = NIL; + fpinfo->fetch_size = 100; + fpinfo->async_capable = false; + fpinfo->is_generated = false; + + apply_server_options(fpinfo); + apply_table_options(fpinfo); + + /* + * If the table or the server is configured to use remote estimates, + * identify which user to do remote access as during planning. This + * should match what ExecCheckPermissions() does. If we fail due to lack + * of permissions, the query would have failed at runtime anyway. + */ + if (fpinfo->use_remote_estimate) + { + Oid userid; + + userid = OidIsValid(baserel->userid) ? baserel->userid : GetUserId(); + fpinfo->user = GetUserMapping(userid, fpinfo->server->serverid); + } + else + fpinfo->user = NULL; + } /* * Identify which baserestrictinfo clauses can be sent to the remote @@ -783,6 +844,9 @@ postgresGetForeignRelSize(PlannerInfo *root, fpinfo->hidden_subquery_rels = NULL; /* Set the relation index. */ fpinfo->relation_index = baserel->relid; + if (existing_fpinfo) + /* Mark fpinfo generated */ + fpinfo->is_generated = true; } /* @@ -1476,13 +1540,72 @@ get_tupdesc_for_join_scan_tuples(ForeignScanState *node) if (!IsA(var, Var) || var->varattno != 0) continue; rte = list_nth(estate->es_range_table, var->varno - 1); - if (rte->rtekind != RTE_RELATION) - continue; - reltype = get_rel_type_id(rte->relid); - if (!OidIsValid(reltype)) - continue; - att->atttypid = reltype; - /* shouldn't need to change anything else */ + if (rte->rtekind == RTE_RELATION) + { + reltype = get_rel_type_id(rte->relid); + if (!OidIsValid(reltype)) + continue; + att->atttypid = reltype; + /* shouldn't need to change anything else */ + } + else if (rte->rtekind == RTE_FUNCTION) + { + RangeTblFunction *rtfunc; + TupleDesc td; + Oid funcrettype; + int num_funcs, + attnum; + ListCell *lc, + *lctype, + *lcname; + bool functype_OK = true; + List *functypes = NIL; + + if (rte->funcordinality) + continue; + + num_funcs = list_length(rte->functions); + Assert(num_funcs >= 0); + + foreach(lc, rte->functions) + { + rtfunc = (RangeTblFunction *) lfirst(lc); + get_expr_result_type(rtfunc->funcexpr, &funcrettype, NULL); + if (!OidIsValid(funcrettype) || funcrettype == RECORDOID) + { + functype_OK = false; + break; + } + functypes = lappend_oid(functypes, funcrettype); + } + if (!functype_OK) + continue; + td = CreateTemplateTupleDesc(num_funcs); + + /* + * funcrettype != RECORD, so we have only one return attribute per + * function + */ + Assert(list_length(rte->eref->colnames) == num_funcs); + attnum = 1; + forthree(lc, rte->functions, lctype, functypes, lcname, rte->eref->colnames) + { + char *colname; + + rtfunc = (RangeTblFunction *) lfirst(lc); + funcrettype = lfirst_oid(lctype); + colname = strVal(lfirst(lcname)); + + TupleDescInitEntry(td, (AttrNumber) attnum, colname, + funcrettype, -1, 0); + TupleDescInitEntryCollation(td, (AttrNumber) attnum, + exprCollation(rtfunc->funcexpr)); + attnum++; + } + + assign_record_type_typmod(td); + att->atttypmod = td->tdtypmod; + } } return tupdesc; } @@ -1518,14 +1641,26 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* * Identify which user to do the remote access as. This should match what - * ExecCheckPermissions() does. + * ExecCheckPermissions() does. In case of a join or aggregate, scan RTEs + * until RTE_RELATION is found. We would get the same result from any. */ userid = OidIsValid(fsplan->checkAsUser) ? fsplan->checkAsUser : GetUserId(); if (fsplan->scan.scanrelid > 0) + { rtindex = fsplan->scan.scanrelid; + rte = exec_rt_fetch(rtindex, estate); + } else - rtindex = bms_next_member(fsplan->fs_base_relids, -1); - rte = exec_rt_fetch(rtindex, estate); + { + rtindex = -1; + while ((rtindex = bms_next_member(fsplan->fs_base_relids, rtindex)) >= 0) + { + rte = exec_rt_fetch(rtindex, estate); + if (rte && rte->rtekind == RTE_RELATION) + break; + } + Assert(rte && rte->rtekind == RTE_RELATION); + } /* Get info about foreign table. */ table = GetForeignTable(rte->relid); @@ -2566,8 +2701,30 @@ postgresPlanDirectModify(PlannerInfo *root, * node below. */ if (fscan->scan.scanrelid == 0) + { + ListCell *lc; + Var *tid_var = NULL; + + /* + * We should explicitly add tableoid to returning list if it's + * requested + */ + foreach(lc, processed_tlist) + { + TargetEntry *tle = lfirst_node(TargetEntry, lc); + Var *var = (Var *) tle->expr; + + if (IsA(var, Var) && (var->varattno == TableOidAttributeNumber) && (strcmp(tle->resname, "tableoid") == 0)) + { + tid_var = var; + break; + } + + } + returningList = build_remote_returning(resultRelation, rel, - returningList); + returningList, tid_var); + } } /* @@ -2889,21 +3046,66 @@ postgresExplainForeignScan(ForeignScanState *node, ExplainState *es) rti += rtoffset; Assert(bms_is_member(rti, plan->fs_base_relids)); rte = rt_fetch(rti, es->rtable); - Assert(rte->rtekind == RTE_RELATION); /* This logic should agree with explain.c's ExplainTargetRel */ - relname = get_rel_name(rte->relid); - if (es->verbose) + if (rte->rtekind == RTE_RELATION) { - char *namespace; - - namespace = get_namespace_name_or_temp(get_rel_namespace(rte->relid)); - appendStringInfo(relations, "%s.%s", - quote_identifier(namespace), - quote_identifier(relname)); + // Note: relname may be uninitialized. + relname = get_rel_name(rte->relid); + if (es->verbose) + { + char *namespace; + + namespace = get_namespace_name_or_temp(get_rel_namespace(rte->relid)); + appendStringInfo(relations, "%s.%s", + quote_identifier(namespace), + quote_identifier(relname)); + } + else + appendStringInfoString(relations, + quote_identifier(relname)); + } + else if (rte->rtekind == RTE_FUNCTION) + { + ListCell *lc; + int n; + bool first = true; + + + n = list_length(rte->functions); + + if (n > 1) + appendStringInfo(relations, "ROWS FROM("); + foreach(lc, rte->functions) + { + RangeTblFunction *rtfunc = (RangeTblFunction *) lfirst(lc); + + if (!first) + appendStringInfoString(relations, ", "); + else + first = false; + + if (IsA(rtfunc->funcexpr, FuncExpr)) + { + FuncExpr *funcexpr = (FuncExpr *) rtfunc->funcexpr; + Oid funcid = funcexpr->funcid; + + relname = get_func_name(funcid); + if (es->verbose) + { + char *namespace; + + namespace = get_namespace_name(get_func_namespace(funcid)); + appendStringInfo(relations, "%s.%s()", + quote_identifier(namespace), + quote_identifier(relname)); + } + else + appendStringInfo(relations, "%s()", quote_identifier(relname)); + } + } + if (n > 1) + appendStringInfo(relations, ")"); } - else - appendStringInfoString(relations, - quote_identifier(relname)); refname = (char *) list_nth(es->rtable_names, rti - 1); if (refname == NULL) refname = rte->eref->aliasname; @@ -4427,7 +4629,7 @@ deallocate_query(PgFdwModifyState *fmstate) * UPDATE/DELETE .. RETURNING on a join directly */ static List * -build_remote_returning(Index rtindex, Relation rel, List *returningList) +build_remote_returning(Index rtindex, Relation rel, List *returningList, Var *tid) { bool have_wholerow = false; List *tlist = NIL; @@ -4436,6 +4638,19 @@ build_remote_returning(Index rtindex, Relation rel, List *returningList) Assert(returningList); + /* + * If tid is requested, add it to the returning list + */ + if (tid) + { + tlist = lappend(tlist, + makeTargetEntry((Expr *) tid, + list_length(tlist) + 1, + NULL, + false)); + + } + vars = pull_var_clause((Node *) returningList, PVC_INCLUDE_PLACEHOLDERS); /* @@ -4727,11 +4942,13 @@ init_returning_filter(PgFdwDirectModifyState *dmstate, if (attrno < 0) { /* - * We don't retrieve system columns other than ctid and oid. + * We don't retrieve system columns other than ctid and oid, + * but locally-generated tableoid can appear in returning + * list. */ if (attrno == SelfItemPointerAttributeNumber) dmstate->ctidAttno = i; - else + else if (attrno != TableOidAttributeNumber) Assert(false); dmstate->hasSystemCols = true; } @@ -5745,6 +5962,63 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) return commands; } +/* + * Determine if foreignrel, not backed by foreign + * table, is fine to push down. + */ +static bool +is_nonrel_relinfo_ok(PlannerInfo *root, RelOptInfo *foreignrel) +{ + RangeTblEntry *rte; + RangeTblFunction *rtfunc; + TupleDesc tupdesc; + Oid funcrettype; + + rte = planner_rt_fetch(foreignrel->relid, root); + + if (!rte) + return false; + + Assert(foreignrel->fdw_private); + + if (rte->rtekind == RTE_FUNCTION) + { + ListCell *lc; + + /* + * WITH ORDINALITY pushdown is not implemented yet. + */ + if (rte->funcordinality) + return false; + + Assert(list_length(rte->functions) >= 1); + foreach(lc, rte->functions) + { + rtfunc = (RangeTblFunction *) lfirst(lc); + + get_expr_result_type(rtfunc->funcexpr, &funcrettype, &tupdesc); + + /* + * Remote server requires a well defined return type for a + * function pushdown. + */ + if (!OidIsValid(funcrettype) || funcrettype == RECORDOID || funcrettype == VOIDOID) + return false; + + if (contain_var_clause(rtfunc->funcexpr) || + contain_mutable_functions(rtfunc->funcexpr) || + contain_subplans(rtfunc->funcexpr)) + return false; + if (!is_foreign_expr(root, foreignrel, (Expr *) rtfunc->funcexpr)) + return false; + } + + return true; + } + + return false; +} + /* * Check if reltarget is safe enough to push down semi-join. Reltarget is not * safe, if it contains references to inner rel relids, which do not belong to @@ -6477,6 +6751,43 @@ postgresGetForeignJoinPaths(PlannerInfo *root, /* XXX Consider parameterized paths for the join relation */ } +/* + * postgresTryShippableJoinPaths + * + * Try to add foreign join of foreign relation with shippable RTE. + */ +static void +postgresTryShippableJoinPaths(PlannerInfo *root, + RelOptInfo *joinrel, + RelOptInfo *outerrel, + RelOptInfo *innerrel, + JoinType jointype, + JoinPathExtraData *extra) +{ + PgFdwRelationInfo *fpinfo_o = (PgFdwRelationInfo *) outerrel->fdw_private; + PgFdwRelationInfo *fpinfo_i = (PgFdwRelationInfo *) innerrel->fdw_private; + + if (fpinfo_o == NULL) + /* Outer path is not foreign relation or foreign JOIN. */ + return; + + if (joinrel->fdwroutine != NULL || innerrel->reloptkind != RELOPT_BASEREL) + return; + + if (fpinfo_i == NULL || fpinfo_i->is_generated) + init_fpinfo(root, innerrel, InvalidOid, fpinfo_o); + + if (!is_nonrel_relinfo_ok(root, innerrel)) + return; + + joinrel->serverid = outerrel->serverid; + joinrel->userid = outerrel->userid; + joinrel->useridiscurrent = outerrel->useridiscurrent; + joinrel->fdwroutine = outerrel->fdwroutine; + + postgresGetForeignJoinPaths(root, joinrel, outerrel, innerrel, jointype, extra); +} + /* * Assess whether the aggregation, grouping and having operations can be pushed * down to the foreign server. As a side effect, save information we obtain in diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index 9e501660d1..772144d611 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -52,6 +52,12 @@ typedef struct PgFdwRelationInfo /* True means that the query_pathkeys is safe to push down */ bool qp_is_pushdown_safe; + /* + * True means that PgFdwRelationInfo is not extracted from catalogs, but + * generated + */ + bool is_generated; + /* Cost and selectivity of local_conds. */ QualCost local_conds_cost; Selectivity local_conds_sel; diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 372fe6dad1..6c717b58b8 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -4278,3 +4278,151 @@ SELECT CASE WHEN closed IS NOT false THEN 1 ELSE 0 END -- Clean up \set VERBOSITY default RESET debug_discard_caches; + +-- =================================================================== +-- test function scan pushdown +-- =================================================================== +CREATE TABLE base_tbl (a int, b int); +CREATE FOREIGN TABLE remote_tbl (a int, b int) + SERVER loopback OPTIONS (table_name 'base_tbl'); +ALTER FOREIGN TABLE remote_tbl OPTIONS (use_remote_estimate 'true'); +CREATE TABLE base_tbl1 (c int, d text); +CREATE FOREIGN TABLE remote_tbl1 (c int, d text) + SERVER loopback OPTIONS (table_name 'base_tbl1'); +ALTER FOREIGN TABLE remote_tbl1 OPTIONS (use_remote_estimate 'true'); + +INSERT INTO remote_tbl SELECT g, g*2 from generate_series(1,1000) g; +INSERT INTO remote_tbl1 SELECT g, 'text'|| g from generate_series(1,500) g; +ANALYZE base_tbl; +ANALYZE base_tbl1; + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl r, unnest(array[2,3,4]) n WHERE r.a = n; + +SELECT * FROM remote_tbl r, unnest(array[2,3,4]) n WHERE r.a = n +ORDER BY r.a; + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM unnest(array[2,3,4]) n, remote_tbl r WHERE r.a = n; + +SELECT * FROM unnest(array[2,3,4]) n, remote_tbl r WHERE r.a = n +ORDER BY r.a; + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a; + +SELECT * FROM remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a +ORDER BY r.a; + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT r.*,n from remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a and n > 3; + +SELECT * from remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a and n > 3; + +CREATE OR REPLACE FUNCTION get_constant_texts() +RETURNS TABLE (text_value text) AS $$ +BEGIN + RETURN QUERY VALUES + ('text1'), + ('text4'); +END; +$$ LANGUAGE plpgsql IMMUTABLE; + +ALTER EXTENSION postgres_fdw ADD FUNCTION get_constant_texts(); +ALTER SERVER loopback OPTIONS (extensions 'postgres_fdw'); + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT r.*, t.n from remote_tbl1 r, ROWS FROM (unnest(array[3,4]), get_constant_texts()) t (n, txt) +WHERE r.c = t.n AND r.d = t.txt; + +SELECT r.*, t.txt from remote_tbl1 r, ROWS FROM (unnest(array[3,4]), get_constant_texts()) t (n, txt) +WHERE r.c = t.n AND r.d = t.txt; + +-- Complex types +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl r JOIN UNNEST(array[box '((2,3),(-2,-3))']) as t(bx) ON a = area(bx); + +SELECT * FROM remote_tbl r JOIN UNNEST(array[box '((2,3),(-2,-3))']) as t(bx) ON a = area(bx) +ORDER BY r.a; + +-- DML +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE remote_tbl r SET b=5 FROM UNNEST(array[box '((2,3),(-2,-3))']) AS t (bx) WHERE r.a = area(t.bx) +RETURNING a,b; + +UPDATE remote_tbl r SET b=5 FROM UNNEST(array[box '((2,3),(-2,-3))']) AS t (bx) WHERE r.a = area(t.bx) +RETURNING a,b; + +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE remote_tbl r SET b=CASE WHEN random()>=0 THEN 5 ELSE 0 END FROM UNNEST(array[box '((2,3),(-2,-3))']) AS t (bx) WHERE r.a = area(t.bx) +RETURNING a,b; + +UPDATE remote_tbl r SET b=CASE WHEN random()>=0 THEN 5 ELSE 0 END FROM UNNEST(array[box '((2,3),(-2,-3))']) AS t (bx) WHERE r.a = area(t.bx) +RETURNING a,b; + +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE remote_tbl r SET b=5 FROM UNNEST(array[10,20], array[box '((2,3),(-2,-4))', box '((1,2),(-2,-3))']) AS t (l, bx) WHERE r.a between l and area(t.bx) +RETURNING a,b; + +UPDATE remote_tbl r SET b=5 FROM UNNEST(array[10,20], array[box '((2,3),(-2,-4))', box '((1,2),(-2,-3))']) AS t (l, bx) WHERE r.a between l and area(t.bx) +RETURNING a,b; + +EXPLAIN (VERBOSE, COSTS OFF) +UPDATE remote_tbl r SET b=CASE WHEN random()>=0 THEN 5 ELSE 0 END FROM UNNEST(array[10,20], array[box '((2,3),(-2,-4))', box '((1,2),(-2,-3))']) AS t (l, bx) WHERE r.a between l and area(t.bx) +RETURNING a,b; + +UPDATE remote_tbl r SET b=CASE WHEN random()>=0 THEN 5 ELSE 0 END FROM UNNEST(array[10,20], array[box '((2,3),(-2,-4))', box '((1,2),(-2,-3))']) AS t (l, bx) WHERE r.a between l and area(t.bx) +RETURNING a,b; + +-- Test that local functions are not pushed down +CREATE OR REPLACE FUNCTION f(INTEGER) +RETURNS SETOF INTEGER +LANGUAGE sql AS 'select generate_series(1,$1);' IMMUTABLE; + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl r, f(10) n +WHERE r.a = n; + +SELECT * FROM remote_tbl r, remote_tbl1 r1, unnest(array[3,4]) n +WHERE r.a = n AND r1.c = r.a AND n > 3; + +-- Test that a function that returns a record is not pushed down +CREATE OR REPLACE FUNCTION f_ret_record() RETURNS record AS $$ SELECT (1,2)::record $$ language SQL IMMUTABLE; +ALTER EXTENSION postgres_fdw ADD function f_ret_record(); + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT s FROM remote_tbl rt, f_ret_record() AS s(a int, b int) +WHERE s.a = rt.a; + +SELECT s FROM remote_tbl rt, f_ret_record() AS s(a int, b int) +WHERE s.a = rt.a; + +DROP FUNCTION f(INTEGER); +ALTER EXTENSION postgres_fdw DROP FUNCTION get_constant_texts(); +ALTER EXTENSION postgres_fdw DROP FUNCTION f_ret_record(); +DROP FUNCTION get_constant_texts(); +DROP FUNCTION f_ret_record(); +DROP TABLE base_tbl, base_tbl1; +DROP FOREIGN TABLE remote_tbl, remote_tbl1; + +-- Test that function WITH ORDINALITY is not pushed down +CREATE TABLE base_tbl (a int, b int); +CREATE FOREIGN TABLE remote_tbl (a int, b int) SERVER loopback OPTIONS (table_name 'base_tbl'); +INSERT INTO remote_tbl VALUES (1, 2), (2, 3), (3, 4), (5, 6); +ANALYZE remote_tbl; +SET enable_mergejoin TO false; + +EXPLAIN (VERBOSE, COSTS OFF) +SELECT * FROM remote_tbl, unnest(ARRAY[1, 2]) WITH ORDINALITY +WHERE a=unnest; + +SELECT * FROM remote_tbl, unnest(ARRAY[1, 2]) WITH ORDINALITY +WHERE a=unnest; + +DROP TABLE base_tbl; +DROP FOREIGN TABLE remote_tbl; +RESET enable_mergejoin; diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c index 3971138480..57430a8165 100644 --- a/src/backend/optimizer/path/joinpath.c +++ b/src/backend/optimizer/path/joinpath.c @@ -333,6 +333,17 @@ add_paths_to_joinrel(PlannerInfo *root, outerrel, innerrel, jointype, &extra); + /* + * If push down of join is not possible we can try to join foreign + * relation with shippable RTE. In this case we have a chance to push down + * this join yet. + */ + else if (outerrel->fdwroutine && + outerrel->fdwroutine->TryShippableJoinPaths) + outerrel->fdwroutine->TryShippableJoinPaths(root, joinrel, + outerrel, innerrel, + jointype, &extra); + /* * 6. Finally, give extensions a chance to manipulate the path list. They * could add new paths (such as CustomPaths) by calling add_path(), or diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c index 91c7c4fe2f..088845c00f 100644 --- a/src/backend/optimizer/plan/setrefs.c +++ b/src/backend/optimizer/plan/setrefs.c @@ -552,7 +552,6 @@ add_rte_to_flat_rtable(PlannerGlobal *glob, List *rteperminfos, newrte->joinleftcols = NIL; newrte->joinrightcols = NIL; newrte->join_using_alias = NULL; - newrte->functions = NIL; newrte->tablefunc = NULL; newrte->values_lists = NIL; newrte->coltypes = NIL; diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index fcde3876b2..0ccd6776a8 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -221,6 +221,7 @@ typedef struct FdwRoutine /* Functions for remote-join planning */ GetForeignJoinPaths_function GetForeignJoinPaths; + GetForeignJoinPaths_function TryShippableJoinPaths; /* Functions for remote upper-relation (post scan/join) planning */ GetForeignUpperPaths_function GetForeignUpperPaths; diff --git a/src/test/regress/expected/create_view.out b/src/test/regress/expected/create_view.out index f551624afb..3fda68bf7f 100644 --- a/src/test/regress/expected/create_view.out +++ b/src/test/regress/expected/create_view.out @@ -1681,10 +1681,10 @@ select pg_get_viewdef('tt14v', true); -- ... and you can even EXPLAIN it ... explain (verbose, costs off) select * from tt14v; - QUERY PLAN ----------------------------------------- + QUERY PLAN +-------------------------------------------- Function Scan on testviewschm2.tt14f t - Output: t.f1, t.f3, t.f4 + Output: t.f1, t."?dropped?column?", t.f4 Function Call: tt14f() (3 rows) -- 2.43.0 ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Function scan FDW pushdown @ 2025-08-05 21:32 Álvaro Herrera <[email protected]> parent: [email protected] 0 siblings, 0 replies; 3+ messages in thread From: Álvaro Herrera @ 2025-08-05 21:32 UTC (permalink / raw) To: PostgreSQL Hackers <[email protected]> Hello, On 2024-Nov-05, [email protected] wrote: > This is a long-overdue follow-up to the original patch. > Note that this patch contains only the changes required for > function scan pushdown, examples and code related to asymmetric > join are dropped. I've marked this as returned with feedback, as several months have passed without a further version; the current one has a number of gotchas, including some problematic coding detected by a compiler warning, as well as unfinished design: > The issue with setting newrte->functions to NIL still persists. > [...] > I am aware that the rte->functions will now be copied even on > instances that don't utilize a FDW, but I don't see a way to solve it. > Any suggestions are welcome. Feel free to reopen this CF entry[1] once you're able to figure this out. [1] https://commitfest.postgresql.org/patch/5470/ -- Álvaro Herrera Breisgau, Deutschland — https://www.EnterpriseDB.com/ ^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2025-08-05 21:32 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 v5 1/2] In-place table persistence change Kyotaro Horiguchi <[email protected]> 2024-11-05 16:11 Re: Function scan FDW pushdown [email protected] 2025-08-05 21:32 ` Re: Function scan FDW pushdown Álvaro 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