buffile_changes.patch

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);
 	}