Skip to content

Fix walreciever crash with SIGABRT #1979

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
HustonMmmavr wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Fix walreciever crash with SIGABRT #1979

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
222bb04
Fix walreciever crash with SIGABRT
HustonMmmavr Sep 9, 2026
6fed26c
:Revert "Fix walreciever crash with SIGABRT"
HustonMmmavr Sep 21, 2026
a88e568
Fix walreciever crash with SIGABRT
HustonMmmavr Sep 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 1 addition & 4 deletions src/Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 \
Expand Down
43 changes: 7 additions & 36 deletions src/backend/commands/subscriptioncmds.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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));
Expand Down
4 changes: 0 additions & 4 deletions src/backend/replication/Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
7 changes: 6 additions & 1 deletion src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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");
Expand Down
10 changes: 3 additions & 7 deletions src/backend/replication/logical/worker.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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
Expand Down
3 changes: 2 additions & 1 deletion src/backend/replication/walreceiver.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");

Expand Down
18 changes: 9 additions & 9 deletions src/test/walrep/gplibpq.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ static WalReceiverConn *test_connection = NULL;
void
_PG_init(void)
{
libpqwalreceiver_PG_init();
load_file("libpqwalreceiver", false);
}

Datum
Expand All @@ -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)));
Expand Down Expand Up @@ -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)));
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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);
Expand Down
Loading