POC-v4-0004-Add-support-to-apply-lo_write-operations-in-apply.patch

From f73aab1d1d8f1fa65a6863e8e8b788bee5eda23f Mon Sep 17 00:00:00 2001
From: Nitin Motiani <nitinmotiani@google.com>
Date: Wed, 31 Dec 2025 09:09:20 +0000
Subject: [PATCH v4 4/6] Add support to apply lo_write operations in
 applyworker

* Handle REORDER_BUFFER_CHANGE_LOWRITE operations on apply worker.
* This is done by using lo_put. This patch refactors lo_put to provide
  a version which is not fmgr-callable but is available to the C code.
* This patch currently does not handle large objects in tablesync.
---
 src/backend/libpq/be-fsstubs.c           | 24 +++++++++++++++---------
 src/backend/replication/logical/worker.c | 21 +++++++++++++++++++++
 src/include/libpq/be-fsstubs.h           |  1 +
 3 files changed, 37 insertions(+), 9 deletions(-)

diff --git a/src/backend/libpq/be-fsstubs.c b/src/backend/libpq/be-fsstubs.c
index f026ecd8993..415e4d14728 100644
--- a/src/backend/libpq/be-fsstubs.c
+++ b/src/backend/libpq/be-fsstubs.c
@@ -205,6 +205,20 @@ lo_write(int fd, const char *buf, int len)
 	return status;
 }
 
+void
+lo_put(Oid loOid, int64 offset, const char *str, int len)
+{
+	LargeObjectDesc *loDesc;
+	int			written PG_USED_FOR_ASSERTS_ONLY;
+
+	lo_cleanup_needed = true;
+	loDesc = inv_open(loOid, INV_WRITE, CurrentMemoryContext);
+	inv_seek(loDesc, offset, SEEK_SET);
+	written = inv_write(loDesc, str, len);
+	Assert(written == len);
+	inv_close(loDesc);
+}
+
 Datum
 be_lo_lseek(PG_FUNCTION_ARGS)
 {
@@ -860,18 +874,10 @@ be_lo_put(PG_FUNCTION_ARGS)
 	Oid			loOid = PG_GETARG_OID(0);
 	int64		offset = PG_GETARG_INT64(1);
 	bytea	   *str = PG_GETARG_BYTEA_PP(2);
-	LargeObjectDesc *loDesc;
-	int			written PG_USED_FOR_ASSERTS_ONLY;
 
 	PreventCommandIfReadOnly("lo_put()");
 
-	lo_cleanup_needed = true;
-	loDesc = inv_open(loOid, INV_WRITE, CurrentMemoryContext);
-	inv_seek(loDesc, offset, SEEK_SET);
-	written = inv_write(loDesc, VARDATA_ANY(str), VARSIZE_ANY_EXHDR(str));
-	Assert(written == VARSIZE_ANY_EXHDR(str));
-	inv_close(loDesc);
-
+	lo_put(loOid, offset, VARDATA_ANY(str), VARSIZE_ANY_EXHDR(str));
 	PG_RETURN_VOID();
 }
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index dd6fc38a41e..f4a1b7c8922 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -263,6 +263,7 @@
 #include "commands/trigger.h"
 #include "executor/executor.h"
 #include "executor/execPartition.h"
+#include "libpq/be-fsstubs.h"
 #include "libpq/pqformat.h"
 #include "miscadmin.h"
 #include "optimizer/optimizer.h"
@@ -3170,6 +3171,23 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 	EvalPlanQualEnd(&epqstate);
 }
 
+static void
+apply_handle_lowrite(StringInfo s)
+{
+	Oid			loid;
+	int64		offset;
+	Size		datalen;
+	char	   *data;
+
+	if (handle_streamed_transaction(LOGICAL_REP_MSG_LOWRITE, s))
+		return;
+
+	begin_replication_step();
+	logicalrep_read_lo_write(s, &loid, &offset, &datalen, &data);
+	lo_put(loid, offset, data, datalen);
+	end_replication_step();
+}
+
 /*
  * Try to find a tuple received from the publication side (in 'remoteslot') in
  * the corresponding local relation using either replica identity index,
@@ -3875,6 +3893,9 @@ apply_dispatch(StringInfo s)
 			apply_handle_stream_prepare(s);
 			break;
 
+		case LOGICAL_REP_MSG_LOWRITE:
+			apply_handle_lowrite(s);
+			break;
 		default:
 			ereport(ERROR,
 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
diff --git a/src/include/libpq/be-fsstubs.h b/src/include/libpq/be-fsstubs.h
index 1e402787e35..87b4d85a8e4 100644
--- a/src/include/libpq/be-fsstubs.h
+++ b/src/include/libpq/be-fsstubs.h
@@ -21,6 +21,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);
 
 struct LoBulkWriteItem;
 struct LoBulkPutItem;
-- 
2.54.0.545.g6539524ca2-goog