POC-v4-0006-Enable-tablesync-for-large-objects-for-all-users.patch

From 312ac2ba7031d0996994de3e0b77d95a928b34ae Mon Sep 17 00:00:00 2001
From: Nitin Motiani <nitinmotiani@google.com>
Date: Mon, 2 Feb 2026 09:05:51 +0000
Subject: [PATCH v4 6/6] Enable tablesync for large objects for all users

* Instead of trying to copy pg_largeobject entries, we utilize lo_from_bytea to write the entries.

* The use of lo_from_bytea allows us to create the large object if it doesn't already exist.
  This way we can do the tablesync just using one worker instead of running a separate worker for
  pg_largeobject_metadata. Running one worker helps us avoid any synchronization issues which
  would come up with a separate sync worker creating the large objects and a separate one writing to them.

* This also allows the subscription owner to only replicate the large objects its counterpart owns on the
  publisher. This way multiple subscriptions can be created which only take care of only their own large objects.

* The above logic can also be added to the apply side of the code to only replicate the large
  objects owned by the subscription owner.
---
 src/backend/libpq/be-fsstubs.c              | 26 ++++++++----
 src/backend/replication/logical/tablesync.c | 47 ++++++++++++++++++++-
 src/include/libpq/be-fsstubs.h              |  1 +
 3 files changed, 64 insertions(+), 10 deletions(-)

diff --git a/src/backend/libpq/be-fsstubs.c b/src/backend/libpq/be-fsstubs.c
index 415e4d14728..c27e8cb8efd 100644
--- a/src/backend/libpq/be-fsstubs.c
+++ b/src/backend/libpq/be-fsstubs.c
@@ -219,6 +219,22 @@ lo_put(Oid loOid, int64 offset, const char *str, int len)
 	inv_close(loDesc);
 }
 
+Oid
+lo_from_bytea(Oid loOid, bytea *str)
+{
+	LargeObjectDesc *loDesc;
+	int			written PG_USED_FOR_ASSERTS_ONLY;
+
+	lo_cleanup_needed = true;
+	loOid = inv_create(loOid);
+	loDesc = inv_open(loOid, INV_WRITE, CurrentMemoryContext);
+	written = inv_write(loDesc, VARDATA_ANY(str), VARSIZE_ANY_EXHDR(str));
+	Assert(written == VARSIZE_ANY_EXHDR(str));
+	inv_close(loDesc);
+
+	return loOid;
+}
+
 Datum
 be_lo_lseek(PG_FUNCTION_ARGS)
 {
@@ -850,18 +866,10 @@ be_lo_from_bytea(PG_FUNCTION_ARGS)
 {
 	Oid			loOid = PG_GETARG_OID(0);
 	bytea	   *str = PG_GETARG_BYTEA_PP(1);
-	LargeObjectDesc *loDesc;
-	int			written PG_USED_FOR_ASSERTS_ONLY;
 
 	PreventCommandIfReadOnly("lo_from_bytea()");
 
-	lo_cleanup_needed = true;
-	loOid = inv_create(loOid);
-	loDesc = inv_open(loOid, INV_WRITE, CurrentMemoryContext);
-	written = inv_write(loDesc, VARDATA_ANY(str), VARSIZE_ANY_EXHDR(str));
-	Assert(written == VARSIZE_ANY_EXHDR(str));
-	inv_close(loDesc);
-
+	loOid = lo_from_bytea(loOid, str);
 	PG_RETURN_OID(loOid);
 }
 
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index eb718114297..8f461bec014 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -100,6 +100,7 @@
 #include "catalog/pg_subscription_rel.h"
 #include "catalog/pg_type.h"
 #include "commands/copy.h"
+#include "libpq/be-fsstubs.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
 #include "parser/parse_relation.h"
@@ -1440,6 +1441,51 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 
 	replorigin_session_setup(originid, 0);
 	replorigin_xact_state.origin = originid;
+	if (rel->rd_id == LargeObjectRelationId)
+	{
+		TupleTableSlot *slot;
+		Oid			tableRow[2] = {OIDOID, BYTEAOID};
+
+		PushActiveSnapshot(GetTransactionSnapshot());
+		res = walrcv_exec(LogRepWorkerWalRcvConn, "SELECT m.oid as oid, lo_get(m.oid) FROM pg_largeobject_metadata as m JOIN pg_user as u ON m.lomowner = u.usesysid where u.usename=CURRENT_USER", 2, tableRow);
+		if (res->status != WALRCV_OK_TUPLES)
+			ereport(ERROR,
+					errmsg("could not run large objects command."));
+		slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+		while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+		{
+			Oid			looid;
+			bytea	   *lodata;
+			bool		isnull;
+
+			looid = DatumGetObjectId(slot_getattr(slot, 1, &isnull));
+			Assert(!isnull);
+			lodata = (bytea *) (slot_getattr(slot, 2, &isnull));
+			lo_from_bytea(looid, lodata);
+			ExecClearTuple(slot);
+		}
+		ExecDropSingleTupleTableSlot(slot);
+		walrcv_clear_result(res);
+		PopActiveSnapshot();
+
+		res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL);
+		if (res->status != WALRCV_OK_COMMAND)
+			ereport(ERROR,
+					(errcode(ERRCODE_CONNECTION_FAILURE),
+					 errmsg("table copy could not finish transaction on publisher: %s",
+							res->err)));
+
+		walrcv_clear_result(res);
+		table_close(rel, NoLock);
+		CommandCounterIncrement();
+		UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
+								   MyLogicalRepWorker->relid,
+								   SUBREL_STATE_FINISHEDCOPY,
+								   MyLogicalRepWorker->relstate_lsn,
+								   false);
+		CommitTransactionCommand();
+		goto copy_table_done;
+	}
 
 	/*
 	 * If the user did not opt to run as the owner of the subscription
@@ -1508,7 +1554,6 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	CommitTransactionCommand();
 
 copy_table_done:
-
 	elog(DEBUG1,
 		 "LogicalRepSyncTableStart: '%s' origin_startpos lsn %X/%08X",
 		 originname, LSN_FORMAT_ARGS(*origin_startpos));
diff --git a/src/include/libpq/be-fsstubs.h b/src/include/libpq/be-fsstubs.h
index 87b4d85a8e4..46ad311e401 100644
--- a/src/include/libpq/be-fsstubs.h
+++ b/src/include/libpq/be-fsstubs.h
@@ -22,6 +22,7 @@
 extern int	lo_read(int fd, char *buf, int len);
 extern int	lo_write(int fd, const char *buf, int len);
 extern void lo_put(Oid loOid, int64 offset, const char *str, int len);
+extern Oid lo_from_bytea(Oid loOid, bytea *str);
 
 struct LoBulkWriteItem;
 struct LoBulkPutItem;
-- 
2.54.0.545.g6539524ca2-goog