From 165fcc3a5ed488905cbf8b066913b71f8e7ffe9a Mon Sep 17 00:00:00 2001 From: mingjiegao Date: Sat, 31 Oct 2026 23:06:26 +0800 Subject: [PATCH v1] Fix initial sync of zero-column published tables Initial table synchronization passes an empty column-name list to BeginCopyFrom() when the publisher has no columns. BeginCopyFrom() interprets that as an unspecified list and selects all target columns that are neither dropped nor generated, including columns that exist only on the subscriber. Text COPY can therefore store an empty string instead of the subscriber's default, while binary COPY fails because the row contains zero fields. Add BeginCopyFromExtended() so callers can distinguish an explicitly empty input column list from an unspecified list. Use it in tablesync, leaving the existing BeginCopyFrom() interface and behavior unchanged. Resolve the explicit list before defaults and parser state are initialized, so the usual COPY machinery supplies the subscriber-only values and handles constraints, triggers and partition routing. Add regression tests for initial and incremental zero-column rows in text and binary transfer modes. --- src/backend/commands/copyfrom.c | 26 +++++++- src/backend/replication/logical/tablesync.c | 3 +- src/include/commands/copy.h | 5 ++ src/test/subscription/t/008_diff_schema.pl | 72 +++++++++++++++++++++ 4 files changed, 104 insertions(+), 2 deletions(-) diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index 746b4fdf6c3..534eb1ea343 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -1540,6 +1540,27 @@ BeginCopyFrom(ParseState *pstate, copy_data_source_cb data_source_cb, List *attnamelist, List *options) +{ + return BeginCopyFromExtended(pstate, rel, whereClause, filename, + is_program, data_source_cb, attnamelist, + options, false); +} + +/* + * Like BeginCopyFrom(), but if attnamelist_is_explicit is true, a NIL + * attnamelist means zero input columns rather than all target columns. + * This is needed when a logical replication source has no columns. + */ +CopyFromState +BeginCopyFromExtended(ParseState *pstate, + Relation rel, + Node *whereClause, + const char *filename, + bool is_program, + copy_data_source_cb data_source_cb, + List *attnamelist, + List *options, + bool attnamelist_is_explicit) { CopyFromState cstate; bool pipe = (filename == NULL); @@ -1590,7 +1611,10 @@ BeginCopyFrom(ParseState *pstate, /* process common options or initialization */ /* Generate or convert list of attributes to process */ - cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist); + if (attnamelist_is_explicit && attnamelist == NIL) + cstate->attnumlist = NIL; + else + cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist); num_phys_attrs = tupDesc->natts; diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 4015e861a64..723012fb0c0 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -1211,7 +1211,8 @@ copy_table(Relation rel) NULL, false, false); attnamelist = make_copy_attnamelist(relmapentry); - cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options); + cstate = BeginCopyFromExtended(pstate, rel, NULL, NULL, false, + copy_read_data, attnamelist, options, true); /* Do the copy */ (void) CopyFrom(cstate); diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h index abecfe51098..74203f3547f 100644 --- a/src/include/commands/copy.h +++ b/src/include/commands/copy.h @@ -114,6 +114,11 @@ extern void ProcessCopyOptions(ParseState *pstate, CopyFormatOptions *opts_out, extern CopyFromState BeginCopyFrom(ParseState *pstate, Relation rel, Node *whereClause, const char *filename, bool is_program, copy_data_source_cb data_source_cb, List *attnamelist, List *options); +extern CopyFromState BeginCopyFromExtended(ParseState *pstate, Relation rel, + Node *whereClause, const char *filename, + bool is_program, copy_data_source_cb data_source_cb, + List *attnamelist, List *options, + bool attnamelist_is_explicit); extern void EndCopyFrom(CopyFromState cstate); extern bool NextCopyFrom(CopyFromState cstate, ExprContext *econtext, Datum *values, bool *nulls); diff --git a/src/test/subscription/t/008_diff_schema.pl b/src/test/subscription/t/008_diff_schema.pl index 96e3d6f5888..7b66c58f9d2 100644 --- a/src/test/subscription/t/008_diff_schema.pl +++ b/src/test/subscription/t/008_diff_schema.pl @@ -31,6 +31,78 @@ $node_subscriber->safe_psql('postgres', # Setup logical replication my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; + +# A zero-column publisher still sends rows. Subscriber-only columns must +# get their defaults during initial COPY as well as incremental replication. +foreach my $binary ('false', 'true') +{ + $node_publisher->safe_psql( + 'postgres', q{ + CREATE TABLE tab_nocols (); + INSERT INTO tab_nocols DEFAULT VALUES; + INSERT INTO tab_nocols DEFAULT VALUES; + CREATE PUBLICATION pub_nocols FOR TABLE tab_nocols + WITH (publish = 'insert'); + }); + $node_subscriber->safe_psql('postgres', + "CREATE TABLE tab_nocols (local_value text DEFAULT 'subscriber-default')" + ); + $node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION sub_nocols CONNECTION '$publisher_connstr' PUBLICATION pub_nocols WITH (binary = $binary)" + ); + + # Stop waiting if the copy fails, so that a field-count error does not + # obscure the regression behind a timeout. + $node_subscriber->poll_query_until( + 'postgres', q{ + SELECT EXISTS ( + SELECT 1 FROM pg_subscription_rel + WHERE srsubid = (SELECT oid FROM pg_subscription + WHERE subname = 'sub_nocols') + AND srsubstate = 'r') + OR EXISTS ( + SELECT 1 FROM pg_stat_subscription_stats + WHERE subname = 'sub_nocols' AND sync_table_error_count > 0) + }) or die "timed out waiting for zero-column table synchronization"; + my $ready = $node_subscriber->safe_psql( + 'postgres', q{ + SELECT srsubstate = 'r' FROM pg_subscription_rel + WHERE srsubid = (SELECT oid FROM pg_subscription + WHERE subname = 'sub_nocols') + }); + is($ready, 't', "zero-column table synchronized (binary = $binary)"); + + SKIP: + { + skip 'initial synchronization failed', 2 if $ready ne 't'; + + is( $node_subscriber->safe_psql( + 'postgres', q{ + SELECT count(*) FROM tab_nocols + WHERE local_value = 'subscriber-default' + }), + '2', + "initial zero-column rows use local defaults (binary = $binary)"); + + $node_publisher->safe_psql('postgres', + "INSERT INTO tab_nocols DEFAULT VALUES"); + $node_publisher->wait_for_catchup('sub_nocols'); + is( $node_subscriber->safe_psql( + 'postgres', q{ + SELECT count(*) FROM tab_nocols + WHERE local_value = 'subscriber-default' + }), + '3', + "incremental zero-column rows use local defaults (binary = $binary)" + ); + } + + $node_subscriber->safe_psql('postgres', 'DROP SUBSCRIPTION sub_nocols'); + $node_publisher->safe_psql('postgres', + 'DROP PUBLICATION pub_nocols; DROP TABLE tab_nocols'); + $node_subscriber->safe_psql('postgres', 'DROP TABLE tab_nocols'); +} + $node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_pub FOR ALL TABLES"); base-commit: 77705d1e5a616339c3e92f586bbfa7b6bb562beb -- 2.53.0