diff --git a/src/Makefile b/src/Makefile index 56c6f9b196a..40986f4442b 100644 --- a/src/Makefile +++ b/src/Makefile @@ -13,16 +13,13 @@ top_builddir = .. include Makefile.global include Makefile.mock -# In PostgreSQL, libpqwalreceiver is linked into a separate shared library. -# In GPDB, it's linked as a normal object, check src/backend/replication/Makefile. -# Remove backend/replication/libpqwalreceiver from SUBDIRS here. - SUBDIRS = \ common \ port \ timezone \ backend \ backend/utils/mb/conversion_procs \ + backend/replication/libpqwalreceiver \ backend/snowball \ include \ interfaces \ diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index d495e4d4c83..fe3a3542753 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -682,13 +682,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, publications = stmt->publication; /* Load the library providing us libpq calls. */ - /* - * In GPDB, we build libpqwalreceiver functions, as well as a copy of - * libpq into the backend itself, to support QD-QE communication. See - * src/backend/libpq. - */ - if (!WalReceiverFunctions) - libpqwalreceiver_PG_init(); + load_file("libpqwalreceiver", false); /* Check the connection info string. */ walrcv_check_conninfo(conninfo, opts.passwordrequired && !superuser()); @@ -901,13 +895,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, bool must_use_password; /* Load the library providing us libpq calls. */ - /* - * In GPDB, we build libpqwalreceiver functions, as well as a copy of - * libpq into the backend itself, to support QD-QE communication. See - * src/backend/libpq. - */ - if (!WalReceiverFunctions) - libpqwalreceiver_PG_init(); + load_file("libpqwalreceiver", false); /* Try to connect to the publisher. */ must_use_password = !superuser_arg(sub->owner) && sub->passwordrequired; @@ -1287,13 +1275,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, case ALTER_SUBSCRIPTION_CONNECTION: /* Load the library providing us libpq calls. */ - /* - * In GPDB, we build libpqwalreceiver functions, as well as a copy of - * libpq into the backend itself, to support QD-QE communication. See - * src/backend/libpq. - */ - if (!WalReceiverFunctions) - libpqwalreceiver_PG_init(); + load_file("libpqwalreceiver", false); + /* Check the connection info string. */ walrcv_check_conninfo(stmt->conninfo, sub->passwordrequired && !superuser_arg(sub->owner)); @@ -1757,13 +1740,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) * doing the database operations we won't be able to rollback dropped * slot. */ - /* - * In GPDB, we build libpqwalreceiver functions, as well as a copy of - * libpq into the backend itself, to support QD-QE communication. See - * src/backend/libpq. - */ - if (!WalReceiverFunctions) - libpqwalreceiver_PG_init(); + + load_file("libpqwalreceiver", false); wrconn = walrcv_connect(conninfo, true, must_use_password, subname, &err); @@ -1847,14 +1825,7 @@ ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missi Assert(wrconn); - /* - * Cloudberry: libpqwalreceiver is linked directly into the backend - * (not a separate shared library), so call libpqwalreceiver_PG_init() - * directly instead of load_file(). Guard against double-init since - * callers may have already initialized it. - */ - if (WalReceiverFunctions == NULL) - libpqwalreceiver_PG_init(); + load_file("libpqwalreceiver", false); initStringInfo(&cmd); appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s WAIT", quote_identifier(slotname)); diff --git a/src/backend/replication/Makefile b/src/backend/replication/Makefile index 2550e16b00f..cd5bd2781f7 100644 --- a/src/backend/replication/Makefile +++ b/src/backend/replication/Makefile @@ -29,10 +29,6 @@ OBJS = \ OBJS += gp_replication.o -# In PostgreSQL, libpqwalreceiver is linked into a separate shared library. -# In GPDB, it's linked as a normal object. -OBJS += libpqwalreceiver/libpqwalreceiver.o - SUBDIRS = logical include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 560e432be2c..34fdf5a3e8d 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -36,6 +36,11 @@ #include "utils/pg_lsn.h" #include "utils/tuplestore.h" + +PG_MODULE_MAGIC; + +void _PG_init(void); + /* * In PostgreSQL, this is a dynamically loaded module, because PostgreSQL * doesn't want to link libpq statically into the backend. In GPDB, we have @@ -117,7 +122,7 @@ static char *stringlist_to_identifierstr(PGconn *conn, List *strings); * Module initialization function */ void -libpqwalreceiver_PG_init(void) +_PG_init(void) { if (WalReceiverFunctions != NULL) elog(ERROR, "libpqwalreceiver already loaded"); diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 9bfc161beaf..8e162d66db5 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -4480,13 +4480,6 @@ void InitializeApplyWorker(void) { MemoryContext oldctx; - /* - * In GPDB, we build libpqwalreceiver functions, as well as a copy of - * libpq into the backend itself, to support QD-QE communication. See - * src/backend/libpq. - */ - if (!WalReceiverFunctions) - libpqwalreceiver_PG_init(); /* Run as replica session replication role. */ SetConfigOption("session_replication_role", "replica", @@ -4587,6 +4580,9 @@ ApplyWorkerMain(Datum main_arg) MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time = MyLogicalRepWorker->reply_time = GetCurrentTimestamp(); + /* Load the libpq-specific functions */ + load_file("libpqwalreceiver", false); + /* * Cloudberry: libpqwalreceiver is linked directly into the backend * (not a separate shared library), so skip load_file() and let diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index 317b0312caa..c3584acc98a 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -294,7 +294,8 @@ WalReceiverMain(void) pqsignal(SIGCHLD, SIG_DFL); /* Load the libpq-specific functions */ - libpqwalreceiver_PG_init(); + load_file("libpqwalreceiver", false); + if (WalReceiverFunctions == NULL) elog(ERROR, "libpqwalreceiver didn't initialize correctly"); diff --git a/src/test/walrep/gplibpq.c b/src/test/walrep/gplibpq.c index 84fddbfe3b8..950dd089cbf 100644 --- a/src/test/walrep/gplibpq.c +++ b/src/test/walrep/gplibpq.c @@ -74,7 +74,7 @@ static WalReceiverConn *test_connection = NULL; void _PG_init(void) { - libpqwalreceiver_PG_init(); + load_file("libpqwalreceiver", false); } Datum @@ -85,7 +85,7 @@ test_connect(PG_FUNCTION_ARGS) MemoryContext oldcxt; oldcxt = MemoryContextSwitchTo(TopMemoryContext); - test_connection = walrcv_connect(conninfo, false, "walrcv_test", &err); + test_connection = walrcv_connect(conninfo, false, false, "walrcv_test", &err); if (!test_connection) ereport(ERROR, (errmsg("could not connect to the primary server: %s", err))); @@ -361,7 +361,7 @@ test_xlog_ao(PG_FUNCTION_ARGS) xrecoff = (uint32)startpoint; - conn = walrcv_connect(conninfo, false, "walrcv_test_ao_xlog", &err); + conn = walrcv_connect(conninfo, false, false, "walrcv_test_ao_xlog", &err); if (!conn) ereport(ERROR, (errmsg("could not connect to the primary server: %s", err))); @@ -411,9 +411,9 @@ test_xlog_ao(PG_FUNCTION_ARGS) values[1] = CStringGetTextDatum("XLOG_APPENDONLY_TRUNCATE"); values[2] = Int32GetDatum(result->len); - values[3] = ObjectIdGetDatum(result->target.node.spcNode); - values[4] = ObjectIdGetDatum(result->target.node.dbNode); - values[5] = ObjectIdGetDatum(result->target.node.relNode); + values[3] = ObjectIdGetDatum(result->target.node.spcOid); + values[4] = ObjectIdGetDatum(result->target.node.dbOid); + values[5] = ObjectIdGetDatum(result->target.node.relNumber); values[6] = Int32GetDatum(result->target.segment_filenum); values[7] = Int64GetDatum(result->target.offset); @@ -496,9 +496,9 @@ check_ao_record_present(unsigned char type, char *buf, Size len, xl_ao_target *xlaorecord = (xl_ao_target*) XLogRecGetData(xlogreader); aorecordresult->xrecoff = xlogreader->ReadRecPtr; - aorecordresult->target.node.spcNode = xlaorecord->node.spcNode; - aorecordresult->target.node.dbNode = xlaorecord->node.dbNode; - aorecordresult->target.node.relNode = xlaorecord->node.relNode; + aorecordresult->target.node.spcOid = xlaorecord->node.spcOid; + aorecordresult->target.node.dbOid = xlaorecord->node.dbOid; + aorecordresult->target.node.relNumber = xlaorecord->node.relNumber; aorecordresult->target.segment_filenum = xlaorecord->segment_filenum; aorecordresult->target.offset = xlaorecord->offset; aorecordresult->len = XLogRecGetDataLen(xlogreader);