From 413d02e042692999b0239c8039a95811491ceb77 Mon Sep 17 00:00:00 2001
From: Nitin Motiani <nitinmotiani@google.com>
Date: Mon, 2 Feb 2026 09:05:51 +0000
Subject: [PATCH v3 4/4] 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              | 24 +++++++----
 src/backend/replication/logical/tablesync.c | 47 ++++++++++++++++++++-
 src/include/libpq/be-fsstubs.h              |  1 +
 3 files changed, 62 insertions(+), 10 deletions(-)

diff --git a/src/backend/libpq/be-fsstubs.c b/src/backend/libpq/be-fsstubs.c
index ffabad9b708..9d3fbeb809e 100644
--- a/src/backend/libpq/be-fsstubs.c
+++ b/src/backend/libpq/be-fsstubs.c
@@ -216,6 +216,20 @@ lo_put(Oid loOid, int64 offset, const char *str, int len)
 	inv_close(loDesc);
 }
 
+void
+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);
+}
+
 Datum
 be_lo_lseek(PG_FUNCTION_ARGS)
 {
@@ -847,18 +861,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);
-
+	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 19a3c21a863..62ea0ee416c 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"
@@ -1406,6 +1407,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
@@ -1474,7 +1520,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 39e4e751b45..11c1e29efbc 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 void lo_from_bytea(Oid loOid, bytea *str);
 
 /*
  * Cleanup LOs at xact commit/abort
-- 
2.53.0.rc1.225.gd81095ad13-goog

