From 20320ffa7ff16140bcd031dbb81a0e485b6c4c7a Mon Sep 17 00:00:00 2001 From: Greg Burd Date: Wed, 17 Jun 2026 21:43:53 -0400 Subject: [PATCH v48 8/9] Gate HOT-indexed updates on the logical-replication apply path A HOT-indexed update of a replica-identity attribute on a subscriber leaves a stale index leaf that the apply worker's replica-identity lookups must tolerate -- which they do, but only when the subscriber's indexed attributes do not extend past the columns those lookups key on. Add the per-subscription hot_indexed_on_apply option (subhotindexedonapply: off / subset_only (default) / always) and have HeapUpdateHotAllowable consult it when running in an apply worker, comparing the relation's indexed-attribute set against its primary-key attributes: "off" disqualifies HOT-indexed whenever any indexed attribute lies outside the primary key, "subset_only" requires the indexed attributes to be a subset of the primary key, and "always" applies no apply-path gating. Wire the option through CREATE/ALTER SUBSCRIPTION, pg_subscription, pg_dump, and psql's \dRs+, and document it (create_subscription, alter_subscription, catalogs). Cover apply under each mode (039), apply under REPLICA IDENTITY FULL and a non-PK USING INDEX whose key is cycled (040), and decoding of HOT-indexed update chains (test_decoding). Authored-by: Greg Burd --- contrib/test_decoding/Makefile | 2 +- .../test_decoding/expected/hot_indexed.out | 59 +++ contrib/test_decoding/meson.build | 1 + contrib/test_decoding/sql/hot_indexed.sql | 29 ++ doc/src/sgml/catalogs.sgml | 16 + doc/src/sgml/ref/alter_subscription.sgml | 4 + doc/src/sgml/ref/create_subscription.sgml | 55 +++ src/backend/access/heap/README.HOT-INDEXED | 34 ++ src/backend/access/heap/heapam.c | 30 ++ src/backend/catalog/pg_subscription.c | 1 + src/backend/catalog/system_views.sql | 1 + src/backend/commands/subscriptioncmds.c | 43 +- src/backend/replication/logical/worker.c | 33 ++ src/bin/pg_dump/pg_dump.c | 17 + src/bin/pg_dump/pg_dump.h | 1 + src/bin/psql/describe.c | 6 +- src/include/catalog/catversion.h | 2 +- src/include/catalog/pg_subscription.h | 27 ++ src/include/replication/logicalworker.h | 8 + src/test/regress/expected/subscription.out | 176 ++++---- src/test/subscription/meson.build | 2 + .../subscription/t/039_hot_indexed_apply.pl | 416 ++++++++++++++++++ .../t/040_hot_indexed_replica_identity.pl | 110 +++++ 23 files changed, 980 insertions(+), 93 deletions(-) create mode 100644 contrib/test_decoding/expected/hot_indexed.out create mode 100644 contrib/test_decoding/sql/hot_indexed.sql create mode 100644 src/test/subscription/t/039_hot_indexed_apply.pl create mode 100644 src/test/subscription/t/040_hot_indexed_replica_identity.pl diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile index 0111124399a..800216b2ae4 100644 --- a/contrib/test_decoding/Makefile +++ b/contrib/test_decoding/Makefile @@ -5,7 +5,7 @@ PGFILEDESC = "test_decoding - example of a logical decoding output plugin" REGRESS = ddl xact rewrite toast permissions decoding_in_xact \ decoding_into_rel binary prepared replorigin time messages \ - repack spill slot truncate stream stats twophase twophase_stream + repack spill slot truncate stream stats twophase twophase_stream hot_indexed ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \ oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \ twophase_snapshot slot_creation_error catalog_change_snapshot \ diff --git a/contrib/test_decoding/expected/hot_indexed.out b/contrib/test_decoding/expected/hot_indexed.out new file mode 100644 index 00000000000..1e2186fda56 --- /dev/null +++ b/contrib/test_decoding/expected/hot_indexed.out @@ -0,0 +1,59 @@ +-- Logical decoding of HOT-indexed UPDATEs. A HOT-indexed update is an +-- ordinary heap update at the WAL level (the new version is logged in full), +-- so it must decode exactly like any other update. Exercise a chain of +-- HOT-indexed updates under REPLICA IDENTITY FULL so the decoded old tuple and +-- new tuple can both be checked. +SET synchronous_commit = on; +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding'); + ?column? +---------- + init +(1 row) + +CREATE TABLE hi_decode (id int PRIMARY KEY, a int, b int) WITH (fillfactor = 50); +CREATE INDEX hi_decode_a ON hi_decode (a); +CREATE INDEX hi_decode_b ON hi_decode (b); +ALTER TABLE hi_decode REPLICA IDENTITY FULL; +INSERT INTO hi_decode VALUES (1, 10, 100); +-- each update changes one indexed column, so each stays HOT-indexed +UPDATE hi_decode SET a = 11 WHERE id = 1; +UPDATE hi_decode SET b = 101 WHERE id = 1; +UPDATE hi_decode SET a = 12 WHERE id = 1; +-- cycle a away and back (ABA) and then delete +UPDATE hi_decode SET a = 99 WHERE id = 1; +UPDATE hi_decode SET a = 12 WHERE id = 1; +DELETE FROM hi_decode WHERE id = 1; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, + 'include-xids', '0', 'skip-empty-xacts', '1'); + data +------------------------------------------------------------------------------------------------------------------------------------------- + BEGIN + table public.hi_decode: INSERT: id[integer]:1 a[integer]:10 b[integer]:100 + COMMIT + BEGIN + table public.hi_decode: UPDATE: old-key: id[integer]:1 a[integer]:10 b[integer]:100 new-tuple: id[integer]:1 a[integer]:11 b[integer]:100 + COMMIT + BEGIN + table public.hi_decode: UPDATE: old-key: id[integer]:1 a[integer]:11 b[integer]:100 new-tuple: id[integer]:1 a[integer]:11 b[integer]:101 + COMMIT + BEGIN + table public.hi_decode: UPDATE: old-key: id[integer]:1 a[integer]:11 b[integer]:101 new-tuple: id[integer]:1 a[integer]:12 b[integer]:101 + COMMIT + BEGIN + table public.hi_decode: UPDATE: old-key: id[integer]:1 a[integer]:12 b[integer]:101 new-tuple: id[integer]:1 a[integer]:99 b[integer]:101 + COMMIT + BEGIN + table public.hi_decode: UPDATE: old-key: id[integer]:1 a[integer]:99 b[integer]:101 new-tuple: id[integer]:1 a[integer]:12 b[integer]:101 + COMMIT + BEGIN + table public.hi_decode: DELETE: id[integer]:1 a[integer]:12 b[integer]:101 + COMMIT +(21 rows) + +SELECT pg_drop_replication_slot('regression_slot'); + pg_drop_replication_slot +-------------------------- + +(1 row) + +DROP TABLE hi_decode; diff --git a/contrib/test_decoding/meson.build b/contrib/test_decoding/meson.build index ac655853d26..91765ca0e72 100644 --- a/contrib/test_decoding/meson.build +++ b/contrib/test_decoding/meson.build @@ -42,6 +42,7 @@ tests += { 'stats', 'twophase', 'twophase_stream', + 'hot_indexed', ], 'regress_args': [ '--temp-config', files('logical.conf'), diff --git a/contrib/test_decoding/sql/hot_indexed.sql b/contrib/test_decoding/sql/hot_indexed.sql new file mode 100644 index 00000000000..05d7d091b62 --- /dev/null +++ b/contrib/test_decoding/sql/hot_indexed.sql @@ -0,0 +1,29 @@ +-- Logical decoding of HOT-indexed UPDATEs. A HOT-indexed update is an +-- ordinary heap update at the WAL level (the new version is logged in full), +-- so it must decode exactly like any other update. Exercise a chain of +-- HOT-indexed updates under REPLICA IDENTITY FULL so the decoded old tuple and +-- new tuple can both be checked. +SET synchronous_commit = on; + +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding'); + +CREATE TABLE hi_decode (id int PRIMARY KEY, a int, b int) WITH (fillfactor = 50); +CREATE INDEX hi_decode_a ON hi_decode (a); +CREATE INDEX hi_decode_b ON hi_decode (b); +ALTER TABLE hi_decode REPLICA IDENTITY FULL; + +INSERT INTO hi_decode VALUES (1, 10, 100); +-- each update changes one indexed column, so each stays HOT-indexed +UPDATE hi_decode SET a = 11 WHERE id = 1; +UPDATE hi_decode SET b = 101 WHERE id = 1; +UPDATE hi_decode SET a = 12 WHERE id = 1; +-- cycle a away and back (ABA) and then delete +UPDATE hi_decode SET a = 99 WHERE id = 1; +UPDATE hi_decode SET a = 12 WHERE id = 1; +DELETE FROM hi_decode WHERE id = 1; + +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, + 'include-xids', '0', 'skip-empty-xacts', '1'); + +SELECT pg_drop_replication_slot('regression_slot'); +DROP TABLE hi_decode; diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 4b474c13917..4c9aba72ba7 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -8727,6 +8727,22 @@ SCRAM-SHA-256$<iteration count>:&l + + + subhotindexedonapply char + + + Gating mode for the HOT-indexed apply path. Corresponds to the + hot_indexed_on_apply + subscription option: + + o = off + s = subset_only (default) + a = always + + + + subserver oid diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index e4f0b6b16c7..3423c5e7ed1 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -295,6 +295,10 @@ ALTER SUBSCRIPTION name RENAME TO < retain_dead_tuples, max_retention_duration, and wal_receiver_timeout. + The + hot_indexed_on_apply + option can also be altered; the new value takes effect at the apply + worker's next catalog reload. Only a superuser can set password_required = false. diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index 07d5b1bd77c..7f12eed9627 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -602,6 +602,61 @@ CREATE SUBSCRIPTION subscription_name + + + hot_indexed_on_apply (text) + + + Controls whether the subscription's apply worker may take the + HOT-indexed update path when an UPDATE replicated + from the publisher touches an indexed attribute. Because the + subscriber's index set may differ from the publisher's, an + unconstrained HOT-indexed decision on the apply path can produce a + heap chain whose index state disagrees with the upstream row. The + option restricts when the apply worker is allowed to take that path. + + + Accepted values are: + + + off + + + Force non-HOT on apply whenever the subscriber has any indexed + attribute beyond the primary key. This matches the conservative + pre-existing behaviour. + + + + + subset_only + + + Allow the HOT-indexed apply path when the subscriber's + indexed-attr set is a subset of its primary-key attrs (which + includes the no-secondary-index case). This is the default and + captures the common replication-ready schema shape while staying + safe when the subscriber adds indexes the publisher does not + have. + + + + + always + + + Unconditional HOT-indexed eligibility on apply. The operator + takes responsibility for keeping the subscriber's indexed-attr + set compatible with the publisher's; divergent schemas can + produce spurious duplicate-key conflicts for subsequent + inserts on the subscriber. + + + + + + + diff --git a/src/backend/access/heap/README.HOT-INDEXED b/src/backend/access/heap/README.HOT-INDEXED index 2e5e5f4a081..5561498d95f 100644 --- a/src/backend/access/heap/README.HOT-INDEXED +++ b/src/backend/access/heap/README.HOT-INDEXED @@ -257,6 +257,40 @@ per-index recheck outcomes; and pg_relation_hot_indexed_stats() reports per-relation HOT-indexed chain counts. +Logical replication +------------------- + +A HOT-indexed update of a replica-identity attribute on a subscriber leaves a +stale index leaf; the apply worker's replica-identity lookups tolerate that +only when the indexed attributes are covered by the replica identity. The +per-subscription hot_indexed_on_apply option (pg_subscription.subhotindexedonapply, +off / subset_only / always; subset_only is the default) controls this: +HeapUpdateHotAllowable consults GetHotIndexedApplyMode on the apply path and +falls back to non-HOT when the indexed attributes are not exactly the primary +key (off) or not a subset of it (subset_only). + + +Recovery +-------- + +HOT-indexed updates and the prune/collapse use the existing heap UPDATE and +prune/freeze WAL records, so crash recovery replays them with no new record +types. src/test/recovery/t/054_hot_indexed_recovery.pl builds a chain, +crashes without a checkpoint (forcing WAL redo), and verifies the chain walk, +verify_heapam, and vacuum reclamation after restart, with +wal_consistency_checking = 'all' comparing each replayed page to its FPI. + + +Adversarial tests +----------------- + +src/test/isolation/specs/hot_indexed_adversarial.spec exercises the cases the +invariant must satisfy under concurrency: key cycling (X->Y->X), aborted +HOT-indexed updates, concurrent unique inserts against a freed/taken key, +snapshot-safe reclaim of stale leaves, and reader consistency across a +concurrent prune/collapse. + + Appendices ---------- diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 90a5aaa51b3..766f43e45a7 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -4501,6 +4501,36 @@ HeapUpdateHotAllowable(Relation relation, const Bitmapset *modified_idx_attrs) all_idx_attrs = RelationGetIndexAttrBitmapNoCopy(relation, INDEX_ATTR_BITMAP_INDEXED); + /* + * The logical-replication apply path gates HOT-indexed updates on the + * per-subscription hot_indexed_on_apply option. A HOT-indexed update of + * a replica-identity attribute leaves a stale index leaf; the apply + * worker's replica-identity lookups cope with that (see + * RelationFindReplTupleByIndex), but only when the indexed attributes are + * a subset of the replica identity. "off" disqualifies whenever the + * subscriber has any indexed attribute beyond its PK; "subset_only" (the + * default) requires the indexed attributes to be a subset of the PK; + * "always" applies no apply-path gating. + */ + if (IsLogicalWorker()) + { + char mode = GetHotIndexedApplyMode(); + const Bitmapset *pk_attrs = RelationGetIndexAttrBitmapNoCopy(relation, + INDEX_ATTR_BITMAP_PRIMARY_KEY); + + if (mode == LOGICALREP_HOT_INDEXED_OFF) + { + if (!bms_equal(all_idx_attrs, pk_attrs)) + return HEAP_UPDATE_ALL_INDEXES; + } + else if (mode == LOGICALREP_HOT_INDEXED_SUBSET_ONLY) + { + if (!bms_is_subset(all_idx_attrs, pk_attrs)) + return HEAP_UPDATE_ALL_INDEXES; + } + /* LOGICALREP_HOT_INDEXED_ALWAYS: no apply-path gating. */ + } + /* * System catalogs keep classic HOT (an UPDATE touching no non-summarizing * indexed attribute already returned HEAP_HEAP_ONLY_UPDATE above), but do diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index b5cb301db88..f56268edbda 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -132,6 +132,7 @@ GetSubscription(Oid subid, bool missing_ok, bool conninfo_needed, sub->retaindeadtuples = subform->subretaindeadtuples; sub->maxretention = subform->submaxretention; sub->retentionactive = subform->subretentionactive; + sub->hotindexedmode = subform->subhotindexedonapply; if (conninfo_needed) { diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 4ea1fc00659..37edbeb06fe 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1531,6 +1531,7 @@ GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled, subbinary, substream, subtwophasestate, subdisableonerr, subpasswordrequired, subrunasowner, subfailover, subretaindeadtuples, submaxretention, subretentionactive, + subhotindexedonapply, subserver, subslotname, subsynccommit, subwalrcvtimeout, subpublications, suborigin) ON pg_subscription TO public; diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index ee06a726f42..379f0d83522 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -79,6 +79,7 @@ #define SUBOPT_WAL_RECEIVER_TIMEOUT 0x00010000 #define SUBOPT_LSN 0x00020000 #define SUBOPT_ORIGIN 0x00040000 +#define SUBOPT_HOT_INDEXED_ON_APPLY 0x00080000 /* check if the 'val' has 'bits' set */ #define IsSet(val, bits) (((val) & (bits)) == (bits)) @@ -109,6 +110,7 @@ typedef struct SubOpts char *origin; XLogRecPtr lsn; char *wal_receiver_timeout; + char hotindexedmode; } SubOpts; /* @@ -196,6 +198,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->maxretention = 0; if (IsSet(supported_opts, SUBOPT_ORIGIN)) opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY); + if (IsSet(supported_opts, SUBOPT_HOT_INDEXED_ON_APPLY)) + opts->hotindexedmode = LOGICALREP_HOT_INDEXED_SUBSET_ONLY; /* Parse options */ foreach(lc, stmt_options) @@ -436,6 +440,30 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET, false, 0, false); } + else if (IsSet(supported_opts, SUBOPT_HOT_INDEXED_ON_APPLY) && + strcmp(defel->defname, "hot_indexed_on_apply") == 0) + { + char *val; + + if (IsSet(opts->specified_opts, SUBOPT_HOT_INDEXED_ON_APPLY)) + errorConflictingDefElem(defel, pstate); + + opts->specified_opts |= SUBOPT_HOT_INDEXED_ON_APPLY; + val = defGetString(defel); + + if (pg_strcasecmp(val, "off") == 0) + opts->hotindexedmode = LOGICALREP_HOT_INDEXED_OFF; + else if (pg_strcasecmp(val, "subset_only") == 0) + opts->hotindexedmode = LOGICALREP_HOT_INDEXED_SUBSET_ONLY; + else if (pg_strcasecmp(val, "always") == 0) + opts->hotindexedmode = LOGICALREP_HOT_INDEXED_ALWAYS; + else + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("unrecognized value for subscription parameter \"%s\": \"%s\"", + "hot_indexed_on_apply", val), + errhint("Valid values are \"off\", \"subset_only\", and \"always\"."))); + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -674,7 +702,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_RETAIN_DEAD_TUPLES | SUBOPT_MAX_RETENTION_DURATION | - SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN); + SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN | + SUBOPT_HOT_INDEXED_ON_APPLY); parse_subscription_options(pstate, stmt->options, supported_opts, &opts); /* @@ -828,6 +857,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, Int32GetDatum(opts.maxretention); values[Anum_pg_subscription_subretentionactive - 1] = BoolGetDatum(opts.retaindeadtuples); + values[Anum_pg_subscription_subhotindexedonapply - 1] = + CharGetDatum(opts.hotindexedmode); values[Anum_pg_subscription_subserver - 1] = ObjectIdGetDatum(serverid); if (!OidIsValid(serverid)) values[Anum_pg_subscription_subconninfo - 1] = @@ -1499,7 +1530,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, SUBOPT_RETAIN_DEAD_TUPLES | SUBOPT_MAX_RETENTION_DURATION | SUBOPT_WAL_RECEIVER_TIMEOUT | - SUBOPT_ORIGIN); + SUBOPT_ORIGIN | + SUBOPT_HOT_INDEXED_ON_APPLY); break; case ALTER_SUBSCRIPTION_ENABLED: @@ -1858,6 +1890,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, replaces[Anum_pg_subscription_subwalrcvtimeout - 1] = true; } + if (IsSet(opts.specified_opts, SUBOPT_HOT_INDEXED_ON_APPLY)) + { + values[Anum_pg_subscription_subhotindexedonapply - 1] = + CharGetDatum(opts.hotindexedmode); + replaces[Anum_pg_subscription_subhotindexedonapply - 1] = true; + } + update_tuple = true; break; } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 7799266c614..03f6058c67b 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -484,6 +484,14 @@ WalReceiverConn *LogRepWorkerWalRcvConn = NULL; Subscription *MySubscription = NULL; static bool MySubscriptionValid = false; +/* + * Cache of the per-subscription hot_indexed_on_apply mode. The apply worker + * refreshes this after every successful load of MySubscription; readers + * outside worker.c go through GetHotIndexedApplyMode() so they don't need + * visibility into the Subscription struct or the apply worker's globals. + */ +static char hot_indexed_apply_mode = LOGICALREP_HOT_INDEXED_OFF; + static List *on_commit_wakeup_workers_subids = NIL; bool in_remote_transaction = false; @@ -5171,6 +5179,9 @@ maybe_reread_subscription(void) MemoryContextDelete(MySubscription->cxt); MySubscription = newsub; + /* Refresh the cached HOT-indexed apply mode from the new tuple. */ + hot_indexed_apply_mode = MySubscription->hotindexedmode; + /* Change synchronous commit according to the user's wishes */ SetConfigOption("synchronous_commit", MySubscription->synccommit, PGC_BACKEND, PGC_S_OVERRIDE); @@ -5844,6 +5855,12 @@ InitializeLogRepWorker(void) MySubscriptionValid = true; + /* + * Cache the subscription's HOT-indexed apply mode so it is cheap to + * consult from the heap access method (via GetHotIndexedApplyMode()). + */ + hot_indexed_apply_mode = MySubscription->hotindexedmode; + if (!MySubscription->enabled) { ereport(LOG, @@ -6083,6 +6100,22 @@ IsLogicalWorker(void) return MyLogicalRepWorker != NULL; } +/* + * Return the cached HOT-indexed apply mode of the current logical replication + * worker's subscription. + * + * Callers outside worker.c (notably heapam.c's HeapUpdateHotAllowable) use + * this accessor to avoid pulling in worker_internal.h or the Subscription + * struct. Non-apply processes get LOGICALREP_HOT_INDEXED_OFF, which is the + * conservative value; callers are expected to guard with IsLogicalWorker() + * first for clarity, but the accessor is safe either way. + */ +char +GetHotIndexedApplyMode(void) +{ + return hot_indexed_apply_mode; +} + /* * Is current process a logical replication parallel apply worker? */ diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index c56437d6057..bf1a7977be3 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -5184,6 +5184,7 @@ getSubscriptions(Archive *fout) int i_subfailover; int i_subretaindeadtuples; int i_submaxretention; + int i_subhotindexedonapply; int i, ntups; @@ -5281,6 +5282,14 @@ getSubscriptions(Archive *fout) appendPQExpBufferStr(query, " '-1' AS subwalrcvtimeout,\n"); + if (fout->remoteVersion >= 190000) + appendPQExpBufferStr(query, + " s.subhotindexedonapply,\n"); + else + appendPQExpBuffer(query, + " '%c' AS subhotindexedonapply,\n", + LOGICALREP_HOT_INDEXED_SUBSET_ONLY); + if (fout->remoteVersion >= 190000) appendPQExpBufferStr(query, " fs.srvname AS subservername\n"); else @@ -5325,6 +5334,7 @@ getSubscriptions(Archive *fout) i_subfailover = PQfnumber(res, "subfailover"); i_subretaindeadtuples = PQfnumber(res, "subretaindeadtuples"); i_submaxretention = PQfnumber(res, "submaxretention"); + i_subhotindexedonapply = PQfnumber(res, "subhotindexedonapply"); i_subservername = PQfnumber(res, "subservername"); i_subconninfo = PQfnumber(res, "subconninfo"); i_subslotname = PQfnumber(res, "subslotname"); @@ -5368,6 +5378,8 @@ getSubscriptions(Archive *fout) (strcmp(PQgetvalue(res, i, i_subretaindeadtuples), "t") == 0); subinfo[i].submaxretention = atoi(PQgetvalue(res, i, i_submaxretention)); + subinfo[i].subhotindexedonapply = + *(PQgetvalue(res, i, i_subhotindexedonapply)); if (PQgetisnull(res, i, i_subconninfo)) subinfo[i].subconninfo = NULL; else @@ -5645,6 +5657,11 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo) if (subinfo->submaxretention) appendPQExpBuffer(query, ", max_retention_duration = %d", subinfo->submaxretention); + if (subinfo->subhotindexedonapply == LOGICALREP_HOT_INDEXED_OFF) + appendPQExpBufferStr(query, ", hot_indexed_on_apply = off"); + else if (subinfo->subhotindexedonapply == LOGICALREP_HOT_INDEXED_ALWAYS) + appendPQExpBufferStr(query, ", hot_indexed_on_apply = always"); + if (strcmp(subinfo->subsynccommit, "off") != 0) appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit)); diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index 5a6726d8b12..e3e7401df08 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -722,6 +722,7 @@ typedef struct _SubscriptionInfo bool subfailover; bool subretaindeadtuples; int submaxretention; + char subhotindexedonapply; char *subservername; char *subconninfo; char *subslotname; diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c index af3935b0078..92c77d7a359 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -7093,7 +7093,7 @@ describeSubscriptions(const char *pattern, bool verbose) printQueryOpt myopt = pset.popt; static const bool translate_columns[] = {false, false, false, false, false, false, false, false, false, false, false, false, false, false, - false, false, false, false, false, false, false}; + false, false, false, false, false, false, false, false}; if (pset.sversion < 100000) { @@ -7176,6 +7176,10 @@ describeSubscriptions(const char *pattern, bool verbose) ", submaxretention AS \"%s\"\n", gettext_noop("Max retention duration")); + appendPQExpBuffer(&buf, + ", subhotindexedonapply AS \"%s\"\n", + gettext_noop("HOT-indexed on apply")); + appendPQExpBuffer(&buf, ", subretentionactive AS \"%s\"\n", gettext_noop("Retention active")); diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h index e06eb2abbcb..cefdce9d90b 100644 --- a/src/include/catalog/catversion.h +++ b/src/include/catalog/catversion.h @@ -57,6 +57,6 @@ */ /* yyyymmddN */ -#define CATALOG_VERSION_NO 202606302 +#define CATALOG_VERSION_NO 202606303 #endif diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index 48944201889..eb7eb15a9ea 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -92,6 +92,10 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW * exceeded max_retention_duration, when * defined */ + char subhotindexedonapply; /* Per-subscription gating of the HOT- + * indexed apply path. See + * LOGICALREP_HOT_INDEXED_* constants. */ + Oid subserver BKI_LOOKUP_OPT(pg_foreign_server); /* If connection uses * server */ @@ -164,6 +168,9 @@ typedef struct Subscription * and the retention duration has not * exceeded max_retention_duration, when * defined */ + char hotindexedmode; /* Per-subscription gating of the HOT- indexed + * apply path. See LOGICALREP_HOT_INDEXED_* + * constants. */ char *conninfo; /* Connection string to the publisher */ char *slotname; /* Name of the replication slot */ char *synccommit; /* Synchronous commit setting for worker */ @@ -210,6 +217,26 @@ typedef struct Subscription */ #define LOGICALREP_STREAM_PARALLEL 'p' +/* + * Per-subscription gating of the HOT-indexed apply path. Recorded as a + * single-character code in pg_subscription.subhotindexedonapply. + * + * 'o' -- OFF: force non-HOT on apply whenever the subscriber carries any + * indexed attribute beyond the primary key. Matches the conservative + * behaviour before this option was introduced. + * 's' -- SUBSET_ONLY (default for freshly created subscriptions): allow the + * HOT-indexed apply path when the subscriber's full indexed-attr set is + * a subset of its primary-key attrs (which covers the no-secondary- + * index case as well). Safe on matching schemas; falls back to non-HOT + * when the subscriber adds indexes beyond the primary key. + * 'a' -- ALWAYS: unconditional HOT-indexed eligibility on apply. The + * operator accepts responsibility for keeping subscriber and publisher + * indexed-attr sets compatible. + */ +#define LOGICALREP_HOT_INDEXED_OFF 'o' +#define LOGICALREP_HOT_INDEXED_SUBSET_ONLY 's' +#define LOGICALREP_HOT_INDEXED_ALWAYS 'a' + #endif /* EXPOSE_TO_CLIENT_CODE */ extern Subscription *GetSubscription(Oid subid, bool missing_ok, diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h index 7d748a28da8..c9df7d32f2d 100644 --- a/src/include/replication/logicalworker.h +++ b/src/include/replication/logicalworker.h @@ -24,6 +24,14 @@ extern void SequenceSyncWorkerMain(Datum main_arg); extern bool IsLogicalWorker(void); extern bool IsLogicalParallelApplyWorker(void); +/* + * Accessor for the cached hot_indexed_on_apply mode of the current apply + * worker's subscription. Returns a LOGICALREP_HOT_INDEXED_* code (see + * catalog/pg_subscription.h). Non-apply processes always see + * LOGICALREP_HOT_INDEXED_OFF. + */ +extern char GetHotIndexedApplyMode(void); + extern void HandleParallelApplyMessageInterrupt(void); extern void ProcessParallelApplyMessages(void); diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 8dbfac66326..eb7915a43f3 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -139,18 +139,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub4 SET (origin = any); \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub3; @@ -237,10 +237,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar'; ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | test subscription (1 row) ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false); @@ -249,10 +249,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname'); ALTER SUBSCRIPTION regress_testsub SET (password_required = false); ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | s | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription (1 row) ALTER SUBSCRIPTION regress_testsub SET (password_required = true); @@ -268,10 +268,10 @@ ERROR: unrecognized subscription parameter: "create_slot" -- ok ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345'); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | test subscription (1 row) -- ok - with lsn = NONE @@ -280,10 +280,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE); ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0'); ERROR: invalid WAL location (LSN): 0/0 \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription (1 row) BEGIN; @@ -319,10 +319,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = '80s'); ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = 'foobar'); ERROR: invalid value for parameter "wal_receiver_timeout": "foobar" \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | test subscription (1 row) -- rename back to keep the rest simple @@ -351,19 +351,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (binary = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub; @@ -375,27 +375,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - publication already exists @@ -410,10 +410,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); ERROR: publication "testpub1" is already in subscription "regress_testsub" \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - publication used more than once @@ -428,10 +428,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub" -- ok - delete publications ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub; @@ -467,19 +467,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- we can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -489,10 +489,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -505,18 +505,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -529,10 +529,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -549,10 +549,10 @@ NOTICE: max_retention_duration is ineffective when retain_dead_tuples is disabl WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - max_retention_duration must be non-negative @@ -561,10 +561,10 @@ ERROR: max_retention_duration cannot be negative -- ok ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | HOT-indexed on apply | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+----------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | s | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index e71e95c6297..f7ee5b8449a 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -48,6 +48,8 @@ tests += { 't/036_sequences.pl', 't/037_except.pl', 't/038_walsnd_shutdown_timeout.pl', + 't/039_hot_indexed_apply.pl', + 't/040_hot_indexed_replica_identity.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/039_hot_indexed_apply.pl b/src/test/subscription/t/039_hot_indexed_apply.pl new file mode 100644 index 00000000000..fe108e6f081 --- /dev/null +++ b/src/test/subscription/t/039_hot_indexed_apply.pl @@ -0,0 +1,416 @@ + +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Per-subscription hot_indexed_on_apply option: parser, catalog round-trip, +# ALTER behaviour, and apply-path gating under each of the three modes. +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; +use Time::HiRes qw(usleep); + +my $publisher = PostgreSQL::Test::Cluster->new('publisher'); +$publisher->init(allows_streaming => 'logical'); +$publisher->start; + +my $subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$subscriber->init; +$subscriber->start; + +my $pub_conninfo = $publisher->connstr . ' dbname=postgres'; + +# --- Schema ---------------------------------------------------------------- +# tab_extra has an extra btree index beyond the primary key on the +# subscriber side; that is the schema shape that subset_only must demote +# to non-HOT on apply but always must let through. +$publisher->safe_psql('postgres', + q{CREATE TABLE tab_extra (id int PRIMARY KEY, payload int, tag text)}); + +# tab_pk has only the primary key; indexed-attr set is a subset of the PK +# attrs, so subset_only and always should both allow HOT-indexed on apply. +$publisher->safe_psql('postgres', + q{CREATE TABLE tab_pk (id int PRIMARY KEY, payload int)}); + +$publisher->safe_psql('postgres', + q{CREATE PUBLICATION pub FOR TABLE tab_extra, tab_pk}); + +# Subscriber mirrors both tables. tab_extra has the extra secondary index +# only on the subscriber, which is the schema-divergence case the option +# gates. +$subscriber->safe_psql('postgres', + q{CREATE TABLE tab_extra (id int PRIMARY KEY, payload int, tag text)}); +$subscriber->safe_psql('postgres', + q{CREATE INDEX tab_extra_payload_idx ON tab_extra(payload)}); +$subscriber->safe_psql('postgres', + q{CREATE TABLE tab_pk (id int PRIMARY KEY, payload int)}); + +# --- Parser / catalog checks ---------------------------------------------- +# Default on fresh subscription is 's' (subset_only). +$subscriber->safe_psql('postgres', qq{ + CREATE SUBSCRIPTION sub_default + CONNECTION '$pub_conninfo' + PUBLICATION pub + WITH (connect = false, slot_name = NONE, enabled = false, + create_slot = false); +}); +is( $subscriber->safe_psql('postgres', + q{SELECT subhotindexedonapply FROM pg_subscription + WHERE subname = 'sub_default'}), + 's', + 'fresh subscription defaults to subset_only'); + +# Explicit 'always' is stored as 'a'. +$subscriber->safe_psql('postgres', qq{ + CREATE SUBSCRIPTION sub_always_p + CONNECTION '$pub_conninfo' + PUBLICATION pub + WITH (connect = false, slot_name = NONE, enabled = false, + create_slot = false, hot_indexed_on_apply = 'always'); +}); +is( $subscriber->safe_psql('postgres', + q{SELECT subhotindexedonapply FROM pg_subscription + WHERE subname = 'sub_always_p'}), + 'a', + 'CREATE with hot_indexed_on_apply = always stores a'); + +# ALTER SUBSCRIPTION SET updates the column. +$subscriber->safe_psql('postgres', + q{ALTER SUBSCRIPTION sub_default SET (hot_indexed_on_apply = 'off')}); +is( $subscriber->safe_psql('postgres', + q{SELECT subhotindexedonapply FROM pg_subscription + WHERE subname = 'sub_default'}), + 'o', + 'ALTER SUBSCRIPTION SET hot_indexed_on_apply = off stores o'); + +# Unknown values are rejected. +my ($ret, $stdout, $stderr) = $subscriber->psql('postgres', qq{ + CREATE SUBSCRIPTION sub_bogus + CONNECTION '$pub_conninfo' + PUBLICATION pub + WITH (connect = false, slot_name = NONE, enabled = false, + create_slot = false, hot_indexed_on_apply = 'bogus'); +}); +isnt($ret, 0, 'bogus hot_indexed_on_apply value is rejected'); +like($stderr, + qr/unrecognized value for subscription parameter "hot_indexed_on_apply"/, + 'bogus hot_indexed_on_apply value reports the expected error'); + +# Drop the placeholder subscriptions so we can rebuild with real slots. +$subscriber->safe_psql('postgres', 'DROP SUBSCRIPTION sub_default'); +$subscriber->safe_psql('postgres', 'DROP SUBSCRIPTION sub_always_p'); + +# --- Apply-path behaviour ------------------------------------------------- +# Pre-populate both sides identically so we can use copy_data=false and +# avoid duplicate-key conflicts when we recreate subscriptions across the +# three test cases. We update non-overlapping id ranges per case so the +# pg_stat counters segment cleanly. +$publisher->safe_psql('postgres', + q{INSERT INTO tab_extra + SELECT g, 0, 't' FROM generate_series(1, 200) g}); +$publisher->safe_psql('postgres', + q{INSERT INTO tab_pk + SELECT g, 0 FROM generate_series(1, 200) g}); +$subscriber->safe_psql('postgres', + q{INSERT INTO tab_extra + SELECT g, 0, 't' FROM generate_series(1, 200) g}); +$subscriber->safe_psql('postgres', + q{INSERT INTO tab_pk + SELECT g, 0 FROM generate_series(1, 200) g}); + +# Helper: read counters and poll up to 10 s for n_tup_upd to reach a +# minimum target value (the apply worker flushes pgstat asynchronously). +sub poll_counters +{ + my ($node, $table, $upd_target) = @_; + + my $deadline = time() + 10; + my $row = ''; + while (1) + { + $row = $node->safe_psql('postgres', + qq{SELECT coalesce(n_tup_upd, 0), + coalesce(n_tup_hot_upd, 0), + coalesce(n_tup_hot_indexed_upd, 0) + FROM pg_stat_user_tables WHERE relname = '$table'}); + my ($upd) = split /\|/, $row; + last if ($upd + 0) >= $upd_target || time() >= $deadline; + usleep(100_000); + } + my ($upd, $hot, $hot_idx) = split /\|/, $row; + return ($upd + 0, $hot + 0, $hot_idx + 0); +} + +# Helper: fire UPDATEs that touch the indexed payload column on a given +# id range and return the deltas in (n_tup_upd, n_tup_hot_upd, +# n_tup_hot_indexed_upd) on the subscriber. +sub apply_updates_and_read +{ + my ($table, $sub_name, $id_lo, $id_hi) = @_; + + my ($upd0, $hot0, $hotidx0) = + poll_counters($subscriber, $table, 0); + + for my $i ($id_lo .. $id_hi) + { + $publisher->safe_psql('postgres', + "UPDATE $table SET payload = payload + 1 WHERE id = $i"); + } + $publisher->wait_for_catchup($sub_name); + + my $n = $id_hi - $id_lo + 1; + my ($upd1, $hot1, $hotidx1) = + poll_counters($subscriber, $table, $upd0 + $n); + note("$table $sub_name $id_lo..$id_hi: dn_upd=" + . ($upd1 - $upd0) . " dhot=" . ($hot1 - $hot0) + . " dhotidx=" . ($hotidx1 - $hotidx0)); + return ($upd1 - $upd0, $hot1 - $hot0, $hotidx1 - $hotidx0); +} + +# Case 1: off, subscriber-only secondary index. HOT-indexed must be +# suppressed on tab_extra. Plain HOT updates also stay zero because every +# UPDATE touches `payload` which is indexed on the subscriber. +$subscriber->safe_psql('postgres', qq{ + CREATE SUBSCRIPTION sub_off + CONNECTION '$pub_conninfo' + PUBLICATION pub + WITH (slot_name = 'sub_off_slot', create_slot = true, + hot_indexed_on_apply = 'off', copy_data = false); +}); +$publisher->wait_for_catchup('sub_off'); + +my (undef, undef, $off_extra_hotidx) = + apply_updates_and_read('tab_extra', 'sub_off', 1, 20); +is($off_extra_hotidx, 0, + 'hot_indexed_on_apply = off: no HOT-indexed updates on tab_extra'); + +$subscriber->safe_psql('postgres', 'DROP SUBSCRIPTION sub_off'); + +# Case 2: subset_only. On tab_pk (no secondary index, indexed-attr set is +# a subset of PK attrs), classic HOT must fire because `payload` is not +# indexed there. On tab_extra (subscriber's `payload` index is NOT covered +# by the PK), the apply worker must demote to non-HOT just like 'off'. +$subscriber->safe_psql('postgres', qq{ + CREATE SUBSCRIPTION sub_subset + CONNECTION '$pub_conninfo' + PUBLICATION pub + WITH (slot_name = 'sub_subset_slot', create_slot = true, + hot_indexed_on_apply = 'subset_only', copy_data = false); +}); +$publisher->wait_for_catchup('sub_subset'); + +my (undef, $ss_pk_hot, $ss_pk_hotidx) = + apply_updates_and_read('tab_pk', 'sub_subset', 1, 20); +cmp_ok($ss_pk_hot, '>', 0, + 'hot_indexed_on_apply = subset_only: classic HOT fires on tab_pk'); + +my (undef, undef, $ss_extra_hotidx) = + apply_updates_and_read('tab_extra', 'sub_subset', 21, 40); +is($ss_extra_hotidx, 0, + 'hot_indexed_on_apply = subset_only: no HOT-indexed on tab_extra'); + +$subscriber->safe_psql('postgres', 'DROP SUBSCRIPTION sub_subset'); + +# Case 3: always. Unconditional HOT-indexed eligibility. On tab_extra +# updates touching the indexed payload column should now run on the +# HOT-indexed path: n_tup_hot_indexed_upd must increase. +$subscriber->safe_psql('postgres', qq{ + CREATE SUBSCRIPTION sub_always + CONNECTION '$pub_conninfo' + PUBLICATION pub + WITH (slot_name = 'sub_always_slot', create_slot = true, + hot_indexed_on_apply = 'always', copy_data = false); +}); +$publisher->wait_for_catchup('sub_always'); + +my (undef, undef, $al_extra_hotidx) = + apply_updates_and_read('tab_extra', 'sub_always', 41, 80); +cmp_ok($al_extra_hotidx, '>', 0, + 'hot_indexed_on_apply = always: HOT-indexed fires on tab_extra'); + +# ALTER back to off and verify the apply worker picks up the new mode. +$subscriber->safe_psql('postgres', + q{ALTER SUBSCRIPTION sub_always SET (hot_indexed_on_apply = 'off')}); +is( $subscriber->safe_psql('postgres', + q{SELECT subhotindexedonapply FROM pg_subscription + WHERE subname = 'sub_always'}), + 'o', + 'ALTER sub_always SET hot_indexed_on_apply = off persists'); + +# Drive another batch of updates and confirm n_tup_hot_indexed_upd does NOT +# advance after the worker rereads the catalog. +my (undef, undef, $post_alter_hotidx) = + apply_updates_and_read('tab_extra', 'sub_always', 81, 100); +is($post_alter_hotidx, 0, + 'ALTER to off freezes n_tup_hot_indexed_upd after worker reread'); + +$subscriber->safe_psql('postgres', 'DROP SUBSCRIPTION sub_always'); + +# --- Subscriber INSERT-after-replicated-UPDATE per mode ------------------- +# +# Verify that a subscriber INSERT using the OLD value of a replicated +# UPDATE's indexed column succeeds without a spurious unique-violation +# under each apply mode. Use a dedicated table (tab_uk) so the unique +# constraint can be defined up-front and the test does not collide with +# pre-populated rows from the apply-path scenarios above. +# +# Publisher updates row $upd_id changing payload from 0 to 999. The +# subscriber then inserts a fresh row with payload=0 (the pre-update +# value). Under all three modes _bt_check_unique's recheck of the +# conflicting tuple's live key must recognize the stale leaf entry pointing +# at the chain root, so the INSERT succeeds. + +$publisher->safe_psql('postgres', + q{CREATE TABLE tab_uk ( + id int PRIMARY KEY, + payload int, + tag text, + UNIQUE (payload, tag))}); +$subscriber->safe_psql('postgres', + q{CREATE TABLE tab_uk ( + id int PRIMARY KEY, + payload int, + tag text, + UNIQUE (payload, tag))}); +$publisher->safe_psql('postgres', + q{ALTER PUBLICATION pub ADD TABLE tab_uk}); + +for my $mode ('off', 'subset_only', 'always') +{ + my $base_id = ($mode eq 'off') ? 1 + : ($mode eq 'subset_only') ? 100 : 200; + my $upd_id = $base_id + 1; + my $ins_id = $base_id + 2; + + # Seed a row that we will UPDATE on the publisher (payload starts at 0), + # and drain the apply for it before changing payload. + $publisher->safe_psql('postgres', + "INSERT INTO tab_uk VALUES ($upd_id, 0, 'mode_$mode')"); + + $subscriber->safe_psql('postgres', qq{ + CREATE SUBSCRIPTION sub_uk_$mode + CONNECTION '$pub_conninfo' + PUBLICATION pub + WITH (slot_name = 'sub_uk_${mode}_slot', create_slot = true, + hot_indexed_on_apply = '$mode', copy_data = true); + }); + $publisher->wait_for_catchup("sub_uk_$mode"); + + # Publisher UPDATE: payload 0 -> 999. + $publisher->safe_psql('postgres', + "UPDATE tab_uk SET payload = 999 WHERE id = $upd_id"); + $publisher->wait_for_catchup("sub_uk_$mode"); + + # Subscriber INSERT with the OLD payload value but a unique tag. The + # existing chain leaf with key (0, 'mode_$mode') is now stale: the + # live tuple at the chain root has payload=999. _bt_check_unique + # rechecks the conflicting tuple's live key and recognizes the stale + # leaf, allowing this INSERT to succeed. + my ($r, $out, $err) = $subscriber->psql('postgres', + "INSERT INTO tab_uk VALUES ($ins_id, 0, 'fresh_$mode')"); + is($r, 0, + "hot_indexed_on_apply = $mode: " + . "subscriber INSERT with old payload value succeeds"); + like($err, qr/^$/, + "hot_indexed_on_apply = $mode: " + . "INSERT did not raise an error"); + + $subscriber->safe_psql('postgres', + "DROP SUBSCRIPTION sub_uk_$mode"); +} + +# --- always-mode safety with indexed attrs beyond the replica identity ----- +# +# Amit's corner: under hot_indexed_on_apply = 'always' the apply worker may +# run a HOT-indexed update even when the table has an indexed attribute that +# is NOT covered by the replica identity. The read-side staleness mechanism +# must still let the apply worker's later replica-identity lookups find the +# row, and DELETE/UPDATE replication must converge, including when the +# replica-identity column itself is cycled away and back (ABA). +# +# tab_ri: replica identity is a UNIQUE index on rid (not the PK), and there +# is an extra secondary index on payload that is NOT part of the replica +# identity. So a payload change makes the payload leaf stale, and an rid +# ABA cycle makes the rid (replica-identity) leaf stale -- both while +# 'always' keeps the updates on the HOT chain. +$publisher->safe_psql('postgres', q{ + CREATE TABLE tab_ri (id int PRIMARY KEY, rid int NOT NULL, payload int); + CREATE UNIQUE INDEX tab_ri_rid_uk ON tab_ri(rid); + ALTER TABLE tab_ri REPLICA IDENTITY USING INDEX tab_ri_rid_uk; + CREATE PUBLICATION pub_ri FOR TABLE tab_ri; +}); +$subscriber->safe_psql('postgres', q{ + CREATE TABLE tab_ri (id int PRIMARY KEY, rid int NOT NULL, payload int); + CREATE UNIQUE INDEX tab_ri_rid_uk ON tab_ri(rid); + ALTER TABLE tab_ri REPLICA IDENTITY USING INDEX tab_ri_rid_uk; + CREATE INDEX tab_ri_payload_idx ON tab_ri(payload); +}); + +$publisher->safe_psql('postgres', + q{INSERT INTO tab_ri VALUES (1, 10, 0), (2, 20, 0)}); + +$subscriber->safe_psql('postgres', qq{ + CREATE SUBSCRIPTION sub_ri + CONNECTION '$pub_conninfo' + PUBLICATION pub_ri + WITH (slot_name = 'sub_ri_slot', create_slot = true, + hot_indexed_on_apply = 'always', copy_data = true); +}); +# Wait for the initial table COPY to finish, not just streaming catch-up, so +# the seeded rows are present before we start updating them. +$subscriber->wait_for_subscription_sync($publisher, 'sub_ri'); + +# Cycle the replica-identity column away and back (ABA), and also churn the +# extra payload index, all replicated under 'always'. Each step is a +# HOT-indexed update on the subscriber that leaves a stale leaf. +$publisher->safe_psql('postgres', q{ + UPDATE tab_ri SET rid = 11, payload = payload + 1 WHERE id = 1; + UPDATE tab_ri SET rid = 10, payload = payload + 1 WHERE id = 1; -- rid ABA + UPDATE tab_ri SET payload = payload + 1 WHERE id = 2; +}); +$publisher->wait_for_catchup('sub_ri'); + +# Confirm the apply worker actually took the HOT-indexed path on tab_ri (the +# whole point of 'always' with indexed attrs beyond the replica identity). +# Without this the convergence/verify_heapam asserts below could pass +# vacuously if eligibility silently regressed to plain non-HOT. +my (undef, undef, $ri_hotidx) = poll_counters($subscriber, 'tab_ri', 3); +cmp_ok($ri_hotidx, '>', 0, + 'always-mode: HOT-indexed path fired on tab_ri (rid/payload churn)'); + +# A subsequent replicated UPDATE keyed by the replica identity (rid) must +# find the row despite the stale rid/payload leaves the ABA left behind. +$publisher->safe_psql('postgres', + q{UPDATE tab_ri SET payload = 100 WHERE id = 1}); +# And a replicated DELETE resolved through the replica-identity index must +# also find and remove the right row. +$publisher->safe_psql('postgres', q{DELETE FROM tab_ri WHERE id = 2}); +$publisher->wait_for_catchup('sub_ri'); + +is( $subscriber->safe_psql('postgres', + q{SELECT rid || ':' || payload FROM tab_ri WHERE id = 1}), + '10:100', + 'always-mode: replicated UPDATE found the row via RI after rid ABA'); +is( $subscriber->safe_psql('postgres', + q{SELECT count(*) FROM tab_ri WHERE id = 2}), + '0', + 'always-mode: replicated DELETE found the row via RI with stale leaves'); +# Full convergence cross-check. +is( $subscriber->safe_psql('postgres', + q{SELECT string_agg(id || ',' || rid || ',' || payload, ';' ORDER BY id) + FROM tab_ri}), + $publisher->safe_psql('postgres', + q{SELECT string_agg(id || ',' || rid || ',' || payload, ';' ORDER BY id) + FROM tab_ri}), + 'always-mode: tab_ri converges between publisher and subscriber'); + +# verify_heapam finds no corruption in the HOT-indexed chains left behind. +$subscriber->safe_psql('postgres', 'CREATE EXTENSION IF NOT EXISTS amcheck'); +is( $subscriber->safe_psql('postgres', + q{SELECT count(*) FROM verify_heapam('tab_ri')}), + '0', + 'always-mode: verify_heapam clean on tab_ri after stale-leaf churn'); + +$subscriber->safe_psql('postgres', 'DROP SUBSCRIPTION sub_ri'); + +done_testing(); diff --git a/src/test/subscription/t/040_hot_indexed_replica_identity.pl b/src/test/subscription/t/040_hot_indexed_replica_identity.pl new file mode 100644 index 00000000000..f801787b4c0 --- /dev/null +++ b/src/test/subscription/t/040_hot_indexed_replica_identity.pl @@ -0,0 +1,110 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Live logical replication of HOT-indexed updates under non-default replica +# identities. The apply worker locates the row to update or delete via the +# replica identity: a seqscan for REPLICA IDENTITY FULL, and the nominated +# index for REPLICA IDENTITY USING INDEX. On a subscriber whose tables carry +# extra indexes (so apply performs HOT-indexed updates and leaves stale index +# leaves), that lookup must still find the current row -- including after the +# identity column's value is cycled away and back (ABA). +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $publisher = PostgreSQL::Test::Cluster->new('publisher'); +$publisher->init(allows_streaming => 'logical'); +$publisher->start; + +my $subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$subscriber->init; +$subscriber->start; + +my $pub_conninfo = $publisher->connstr . ' dbname=postgres'; + +# tab_full: REPLICA IDENTITY FULL (apply uses a sequential scan). +# tab_idx: REPLICA IDENTITY USING INDEX on a non-PK unique index whose +# column is itself updated, so the apply-side index lookup must +# tolerate stale leaves left by earlier HOT-indexed updates. +$publisher->safe_psql('postgres', q{ + CREATE TABLE tab_full (a int, b int, c int); + ALTER TABLE tab_full REPLICA IDENTITY FULL; + CREATE TABLE tab_idx (k int NOT NULL, v int, w int); + CREATE UNIQUE INDEX tab_idx_k ON tab_idx (k); + ALTER TABLE tab_idx REPLICA IDENTITY USING INDEX tab_idx_k; + CREATE PUBLICATION pub FOR TABLE tab_full, tab_idx; +}); + +# The subscriber adds extra secondary indexes so that an UPDATE changing one +# indexed column stays HOT-indexed on apply. +$subscriber->safe_psql('postgres', q{ + CREATE TABLE tab_full (a int, b int, c int) WITH (fillfactor = 50); + CREATE INDEX tab_full_b ON tab_full (b); + CREATE INDEX tab_full_c ON tab_full (c); + ALTER TABLE tab_full REPLICA IDENTITY FULL; + CREATE TABLE tab_idx (k int NOT NULL, v int, w int) WITH (fillfactor = 50); + CREATE UNIQUE INDEX tab_idx_k ON tab_idx (k); + CREATE INDEX tab_idx_v ON tab_idx (v); + CREATE INDEX tab_idx_w ON tab_idx (w); + ALTER TABLE tab_idx REPLICA IDENTITY USING INDEX tab_idx_k; +}); + +# Allow HOT-indexed updates on the apply path. +$subscriber->safe_psql('postgres', qq{ + CREATE SUBSCRIPTION sub + CONNECTION '$pub_conninfo' + PUBLICATION pub + WITH (hot_indexed_on_apply = always); +}); +$subscriber->wait_for_subscription_sync($publisher, 'sub'); + +# Seed both tables. +$publisher->safe_psql('postgres', q{ + INSERT INTO tab_full VALUES (1, 10, 100), (2, 20, 200); + INSERT INTO tab_idx VALUES (1, 10, 1000), (2, 20, 2000); +}); +$publisher->wait_for_catchup('sub'); + +# A run of single-column updates: each stays HOT-indexed on the subscriber and +# leaves stale leaves, then the identity/PK row is matched again by the next +# change. Include an ABA cycle on the USING INDEX column k (1 -> 3 -> 1). +$publisher->safe_psql('postgres', q{ + UPDATE tab_full SET b = b + 1 WHERE a = 1; + UPDATE tab_full SET c = c + 1 WHERE a = 1; + UPDATE tab_full SET b = b + 1 WHERE a = 1; + UPDATE tab_idx SET v = v + 1 WHERE k = 1; + UPDATE tab_idx SET k = 3 WHERE k = 1; + UPDATE tab_idx SET w = w + 1 WHERE k = 3; + UPDATE tab_idx SET k = 1 WHERE k = 3; + DELETE FROM tab_full WHERE a = 2; + DELETE FROM tab_idx WHERE k = 2; +}); +$publisher->wait_for_catchup('sub'); + +# The subscriber must match the publisher exactly: the RI lookups found the +# right rows across the HOT-indexed chains and the ABA cycle. +my $pub_full = $publisher->safe_psql('postgres', + q{SELECT a, b, c FROM tab_full ORDER BY a}); +my $sub_full = $subscriber->safe_psql('postgres', + q{SELECT a, b, c FROM tab_full ORDER BY a}); +is($sub_full, $pub_full, 'REPLICA IDENTITY FULL: subscriber matches publisher'); + +my $pub_idx = $publisher->safe_psql('postgres', + q{SELECT k, v, w FROM tab_idx ORDER BY k}); +my $sub_idx = $subscriber->safe_psql('postgres', + q{SELECT k, v, w FROM tab_idx ORDER BY k}); +is($sub_idx, $pub_idx, + 'REPLICA IDENTITY USING INDEX: subscriber matches publisher across ABA'); + +# The subscriber's tables must be structurally consistent (stubs recognised). +$subscriber->safe_psql('postgres', q{CREATE EXTENSION amcheck}); +is( $subscriber->safe_psql('postgres', + q{SELECT count(*) FROM verify_heapam('tab_idx')}), + '0', + 'subscriber tab_idx has no heap corruption after HOT-indexed apply'); + +$subscriber->stop; +$publisher->stop; + +done_testing(); -- 2.50.1