diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 4c58ad8b07..bf5fdda672 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -101,7 +101,10 @@ parse_subscription_options(List *options, *binary = false; } if (streaming) + { *streaming_given = false; + *streaming = false; + } /* Parse options */ foreach(lc, options) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 1347031d01..9f46b0b34f 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -136,20 +136,6 @@ typedef struct SlotErrCallbackArg int remote_attnum; } SlotErrCallbackArg; -/* - * Stream xid hash entry. Whenever we see a new xid we create this entry in the - * xidhash and along with it create the streaming file and store the fileset handle. - * The subxact file is created iff there is any suxact info under this xid. This - * entry is used on the subsequent streams for the xid to get the corresponding - * fileset handles. - */ -typedef struct StreamXidHash -{ - TransactionId xid; /* xid is the hash key and must be first */ - SharedFileSet *stream_fileset; /* shared file set for stream data */ - SharedFileSet *subxact_fileset; /* shared file set for subxact info */ -} StreamXidHash; - static MemoryContext ApplyMessageContext = NULL; MemoryContext ApplyContext = NULL; @@ -169,14 +155,6 @@ bool in_streamed_transaction = false; static TransactionId stream_xid = InvalidTransactionId; -/* - * Hash table for storing the streaming xid information along with shared file - * set for streaming and subxact files. On every stream start we need to open - * the xid's files and for that we need the shared file set handle. So storing - * it in xid hash make it faster to search. - */ -static HTAB *xidhash = NULL; - /* Buf file handle of the current streaming file. */ static BufFile *stream_fd = NULL; @@ -196,6 +174,8 @@ typedef struct ApplySubXactData SubXactInfo *subxacts; /* sub-xact offset in file */ } ApplySubXactData; +SharedFileSet *fileset; + static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL}; static void subxact_filename(char *path, Oid subid, TransactionId xid); @@ -776,7 +756,6 @@ static void apply_handle_stream_start(StringInfo s) { bool first_segment; - HASHCTL hash_ctl; Assert(!in_streamed_transaction); @@ -793,16 +772,6 @@ apply_handle_stream_start(StringInfo s) /* extract XID of the top-level transaction */ stream_xid = logicalrep_read_stream_start(s, &first_segment); - /* Initialize the xidhash table if we haven't yet */ - if (xidhash == NULL) - { - hash_ctl.keysize = sizeof(TransactionId); - hash_ctl.entrysize = sizeof(StreamXidHash); - hash_ctl.hcxt = ApplyContext; - xidhash = hash_create("StreamXidHash", 1024, &hash_ctl, - HASH_ELEM | HASH_CONTEXT); - } - /* open the spool file for this transaction */ stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment); @@ -885,7 +854,6 @@ apply_handle_stream_abort(StringInfo s) BufFile *fd; bool found = false; char path[MAXPGPATH]; - StreamXidHash *ent; subidx = -1; ensure_transaction(); @@ -916,15 +884,9 @@ apply_handle_stream_abort(StringInfo s) Assert((subidx >= 0) && (subidx < subxact_data.nsubxacts)); - ent = (StreamXidHash *) hash_search(xidhash, - (void *) &xid, - HASH_FIND, - &found); - Assert(found); - /* open the changes file */ changes_filename(path, MyLogicalRepWorker->subid, xid); - fd = BufFileOpenShared(ent->stream_fileset, path, O_RDWR); + fd = BufFileOpenShared(fileset, path, O_RDWR); /* OK, truncate the file at the right offset */ BufFileTruncateShared(fd, subxact_data.subxacts[subidx].fileno, @@ -951,9 +913,7 @@ apply_handle_stream_commit(StringInfo s) int nchanges; char path[MAXPGPATH]; char *buffer = NULL; - bool found; LogicalRepCommitData commit_data; - StreamXidHash *ent; MemoryContext oldcxt; BufFile *fd; @@ -969,12 +929,8 @@ apply_handle_stream_commit(StringInfo s) /* open the spool file for the committed transaction */ changes_filename(path, MyLogicalRepWorker->subid, xid); elog(DEBUG1, "replaying changes from file '%s'", path); - ent = (StreamXidHash *) hash_search(xidhash, - (void *) &xid, - HASH_FIND, - &found); - Assert(found); - fd = BufFileOpenShared(ent->stream_fileset, path, O_RDONLY); + + fd = BufFileOpenShared(fileset, path, O_RDONLY); buffer = palloc(BLCKSZ); initStringInfo(&s2); @@ -2501,61 +2457,14 @@ static void subxact_info_write(Oid subid, TransactionId xid) { char path[MAXPGPATH]; - bool found; Size len; - StreamXidHash *ent; BufFile *fd; Assert(TransactionIdIsValid(xid)); subxact_filename(path, subid, xid); - /* find the xid entry in the xidhash */ - ent = (StreamXidHash *) hash_search(xidhash, - (void *) &xid, - HASH_FIND, - &found); - /* we must found the entry for its top transaction by this time */ - Assert(found); - - /* - * If there is no subtransaction then nothing to do, but if already have - * subxact file then delete that. - */ - if (subxact_data.nsubxacts == 0) - { - if (ent->subxact_fileset) - { - cleanup_subxact_info(); - BufFileDeleteShared(ent->subxact_fileset, path); - pfree(ent->subxact_fileset); - ent->subxact_fileset = NULL; - } - - return; - } - - /* - * Create the subxact file if it not already created, otherwise open the - * existing file. - */ - if (ent->subxact_fileset == NULL) - { - MemoryContext oldctx; - - /* - * We need to maintain shared fileset across multiple stream - * start/stop calls. So, need to allocate it in a persistent context. - */ - oldctx = MemoryContextSwitchTo(ApplyContext); - ent->subxact_fileset = palloc(sizeof(SharedFileSet)); - SharedFileSetInit(ent->subxact_fileset, NULL); - MemoryContextSwitchTo(oldctx); - - fd = BufFileCreateShared(ent->subxact_fileset, path); - } - else - fd = BufFileOpenShared(ent->subxact_fileset, path, O_RDWR); + fd = BufFileOpenShared(fileset, path, O_RDWR); len = sizeof(SubXactInfo) * subxact_data.nsubxacts; @@ -2583,10 +2492,8 @@ static void subxact_info_read(Oid subid, TransactionId xid) { char path[MAXPGPATH]; - bool found; Size len; BufFile *fd; - StreamXidHash *ent; MemoryContext oldctx; Assert(TransactionIdIsValid(xid)); @@ -2594,22 +2501,9 @@ subxact_info_read(Oid subid, TransactionId xid) Assert(subxact_data.nsubxacts == 0); Assert(subxact_data.nsubxacts_max == 0); - /* Find the stream xid entry in the xidhash */ - ent = (StreamXidHash *) hash_search(xidhash, - (void *) &xid, - HASH_FIND, - &found); - - /* - * If subxact_fileset is not valid that mean we don't have any subxact - * info - */ - if (ent->subxact_fileset == NULL) - return; - subxact_filename(path, subid, xid); - fd = BufFileOpenShared(ent->subxact_fileset, path, O_RDONLY); + fd = BufFileOpenShared(fileset, path, O_RDONLY); /* read number of subxact items */ if (BufFileRead(fd, &subxact_data.nsubxacts, @@ -2753,27 +2647,18 @@ static void stream_cleanup_files(Oid subid, TransactionId xid, bool missing_ok) { char path[MAXPGPATH]; - StreamXidHash *ent; - /* Remove the xid entry from the stream xid hash */ - ent = (StreamXidHash *) hash_search(xidhash, - (void *) &xid, - HASH_REMOVE, - NULL); - /* By this time we must have created the transaction entry */ - Assert(ent != NULL); + Assert(fileset != NULL); /* Delete the change file and release the stream fileset memory */ changes_filename(path, subid, xid); - SharedFileSetDeleteAll(ent->stream_fileset); - pfree(ent->stream_fileset); - + BufFileDeleteShared(fileset, path); + /* Delete the subxact file and release the memory, if it exist */ - if (ent->subxact_fileset) + if (subxact_data.nsubxacts > 0) { subxact_filename(path, subid, xid); - SharedFileSetDeleteAll(ent->subxact_fileset); - pfree(ent->subxact_fileset); + BufFileDeleteShared(fileset, path); } } @@ -2793,23 +2678,33 @@ static void stream_open_file(Oid subid, TransactionId xid, bool first_segment) { char path[MAXPGPATH]; - bool found; + char subxact_path[MAXPGPATH]; MemoryContext oldcxt; - StreamXidHash *ent; Assert(in_streamed_transaction); Assert(OidIsValid(subid)); Assert(TransactionIdIsValid(xid)); Assert(stream_fd == NULL); - /* create or find the xid entry in the xidhash */ - ent = (StreamXidHash *) hash_search(xidhash, - (void *) &xid, - HASH_ENTER | HASH_FIND, - &found); - Assert(first_segment || found); + /* + * If shared fileset is not initialized yet then do it now. We need to + * maintain shared fileset across multiple stream start/stop calls. So, + * need to allocate it in a persistent context. + */ + if (fileset == NULL) + { + MemoryContext savectx; + + savectx = MemoryContextSwitchTo(ApplyContext); + fileset = palloc(sizeof(SharedFileSet)); + + SharedFileSetInit(fileset, NULL); + MemoryContextSwitchTo(savectx); + } + changes_filename(path, subid, xid); - elog(DEBUG1, "opening file '%s' for streamed changes", path); + subxact_filename(subxact_path, subid, xid); + elog(DEBUG1, "opening file '%s' for streamed changes", path); /* * Create/open the buffiles under the logical streaming context so that we @@ -2824,25 +2719,11 @@ stream_open_file(Oid subid, TransactionId xid, bool first_segment) */ if (first_segment) { - MemoryContext savectx; - SharedFileSet *fileset; - - /* - * We need to maintain shared fileset across multiple stream - * start/stop calls. So, need to allocate it in a persistent context. - */ - savectx = MemoryContextSwitchTo(ApplyContext); - fileset = palloc(sizeof(SharedFileSet)); - - SharedFileSetInit(fileset, NULL); - MemoryContextSwitchTo(savectx); + BufFile *subxact_fd; stream_fd = BufFileCreateShared(fileset, path); - - /* Remember the fileset for the next stream of the same transaction */ - ent->xid = xid; - ent->stream_fileset = fileset; - ent->subxact_fileset = NULL; + subxact_fd = BufFileCreateShared(fileset, subxact_path); + BufFileClose(subxact_fd); } else { @@ -2850,7 +2731,7 @@ stream_open_file(Oid subid, TransactionId xid, bool first_segment) * Open the file and seek to the end of the file because we always * append the changes file. */ - stream_fd = BufFileOpenShared(ent->stream_fileset, path, O_RDWR); + stream_fd = BufFileOpenShared(fileset, path, O_RDWR); BufFileSeek(stream_fd, 0, 0, SEEK_END); }