public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v8 3/4] Remove globals readSegNo, readOff, readLen 17+ messages / 4 participants [nested] [flat]
* [PATCH v8 3/4] Remove globals readSegNo, readOff, readLen @ 2019-09-10 08:28 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 17+ messages in thread From: Kyotaro Horiguchi @ 2019-09-10 08:28 UTC (permalink / raw) These global variables is functionally duplicated with them in XLogReaderState. Remove the globals. --- src/backend/access/transam/xlog.c | 77 ++++++++++++++++++--------------------- src/include/access/xlogreader.h | 1 + 2 files changed, 36 insertions(+), 42 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index b06e381d73..49e8ca486e 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -783,14 +783,10 @@ static XLogSegNo openLogSegNo = 0; * These variables are used similarly to the ones above, but for reading * the XLOG. Note, however, that readOff generally represents the offset * of the page just read, not the seek position of the FD itself, which - * will be just past that page. readLen indicates how much of the current - * page has been read into readBuf, and readSource indicates where we got - * the currently open file from. + * will be just past that page. readSource indicates where we got the + * currently open file from. */ static int readFile = -1; -static XLogSegNo readSegNo = 0; -static uint32 readOff = 0; -static uint32 readLen = 0; static XLogSource readSource = 0; /* XLOG_FROM_* code */ /* @@ -877,10 +873,12 @@ static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath, static int XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli, int source, bool notfoundOk); static int XLogFileReadAnyTLI(XLogSegNo segno, int emode, int source); -static bool XLogPageRead(XLogReaderState *xlogreader, +static bool XLogPageRead(XLogReaderState *state, bool fetching_ckpt, int emode, bool randAccess); static bool WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, - bool fetching_ckpt, XLogRecPtr tliRecPtr); + bool fetching_ckpt, + XLogRecPtr tliRecPtr, + XLogSegNo readSegNo); static int emode_for_corrupt_record(int emode, XLogRecPtr RecPtr); static void XLogFileClose(void); static void PreallocXlogFiles(XLogRecPtr endptr); @@ -7494,7 +7492,8 @@ StartupXLOG(void) XLogRecPtr pageBeginPtr; pageBeginPtr = EndOfLog - (EndOfLog % XLOG_BLCKSZ); - Assert(readOff == XLogSegmentOffset(pageBeginPtr, wal_segment_size)); + Assert(XLogSegmentOffset(xlogreader->readPagePtr, wal_segment_size) == + XLogSegmentOffset(pageBeginPtr, wal_segment_size)); firstIdx = XLogRecPtrToBufIdx(EndOfLog); @@ -11514,13 +11513,14 @@ CancelBackup(void) * sleep and retry. */ static bool -XLogPageRead(XLogReaderState *xlogreader, +XLogPageRead(XLogReaderState *state, bool fetching_ckpt, int emode, bool randAccess) { - char *readBuf = xlogreader->readBuf; - XLogRecPtr targetPagePtr = xlogreader->readPagePtr; - int reqLen = xlogreader->readLen; - XLogRecPtr targetRecPtr = xlogreader->ReadRecPtr; + char *readBuf = state->readBuf; + XLogRecPtr targetPagePtr = state->readPagePtr; + int reqLen = state->readLen; + int readLen = 0; + XLogRecPtr targetRecPtr = state->ReadRecPtr; uint32 targetPageOff; XLogSegNo targetSegNo PG_USED_FOR_ASSERTS_ONLY; int r; @@ -11533,7 +11533,7 @@ XLogPageRead(XLogReaderState *xlogreader, * is not in the currently open one. */ if (readFile >= 0 && - !XLByteInSeg(targetPagePtr, readSegNo, wal_segment_size)) + !XLByteInSeg(targetPagePtr, state->readSegNo, wal_segment_size)) { /* * Request a restartpoint if we've replayed too much xlog since the @@ -11541,10 +11541,10 @@ XLogPageRead(XLogReaderState *xlogreader, */ if (bgwriterLaunched) { - if (XLogCheckpointNeeded(readSegNo)) + if (XLogCheckpointNeeded(state->readSegNo)) { (void) GetRedoRecPtr(); - if (XLogCheckpointNeeded(readSegNo)) + if (XLogCheckpointNeeded(state->readSegNo)) RequestCheckpoint(CHECKPOINT_CAUSE_XLOG); } } @@ -11554,7 +11554,7 @@ XLogPageRead(XLogReaderState *xlogreader, readSource = 0; } - XLByteToSeg(targetPagePtr, readSegNo, wal_segment_size); + XLByteToSeg(targetPagePtr, state->readSegNo, wal_segment_size); retry: /* See if we need to retrieve more data */ @@ -11563,17 +11563,14 @@ retry: receivedUpto < targetPagePtr + reqLen)) { if (!WaitForWALToBecomeAvailable(targetPagePtr + reqLen, - randAccess, - fetching_ckpt, - targetRecPtr)) + randAccess, fetching_ckpt, + targetRecPtr, state->readSegNo)) { if (readFile >= 0) close(readFile); readFile = -1; - readLen = 0; readSource = 0; - - xlogreader->readLen = -1; + state->readLen = -1; return false; } } @@ -11601,40 +11598,36 @@ retry: else readLen = XLOG_BLCKSZ; - /* Read the requested page */ - readOff = targetPageOff; - pgstat_report_wait_start(WAIT_EVENT_WAL_READ); - r = pg_pread(readFile, readBuf, XLOG_BLCKSZ, (off_t) readOff); + r = pg_pread(readFile, readBuf, XLOG_BLCKSZ, (off_t) targetPageOff); if (r != XLOG_BLCKSZ) { char fname[MAXFNAMELEN]; int save_errno = errno; pgstat_report_wait_end(); - XLogFileName(fname, curFileTLI, readSegNo, wal_segment_size); + XLogFileName(fname, curFileTLI, state->readSegNo, wal_segment_size); if (r < 0) { errno = save_errno; ereport(emode_for_corrupt_record(emode, targetPagePtr + reqLen), (errcode_for_file_access(), errmsg("could not read from log segment %s, offset %u: %m", - fname, readOff))); + fname, targetPageOff))); } else ereport(emode_for_corrupt_record(emode, targetPagePtr + reqLen), (errcode(ERRCODE_DATA_CORRUPTED), errmsg("could not read from log segment %s, offset %u: read %d of %zu", - fname, readOff, r, (Size) XLOG_BLCKSZ))); + fname, targetPageOff, r, (Size) XLOG_BLCKSZ))); goto next_record_is_invalid; } pgstat_report_wait_end(); - Assert(targetSegNo == readSegNo); - Assert(targetPageOff == readOff); + Assert(targetSegNo == state->readSegNo); Assert(reqLen <= readLen); - xlogreader->seg.ws_tli = curFileTLI; + state->seg.ws_tli = curFileTLI; /* * Check the page header immediately, so that we can retry immediately if @@ -11662,15 +11655,15 @@ retry: * Validating the page header is cheap enough that doing it twice * shouldn't be a big deal from a performance point of view. */ - if (!XLogReaderValidatePageHeader(xlogreader, targetPagePtr, readBuf)) + if (!XLogReaderValidatePageHeader(state, targetPagePtr, readBuf)) { - /* reset any error XLogReaderValidatePageHeader() might have set */ - xlogreader->errormsg_buf[0] = '\0'; + /* reset any error StateValidatePageHeader() might have set */ + state->errormsg_buf[0] = '\0'; goto next_record_is_invalid; } - Assert(xlogreader->readPagePtr == targetPagePtr); - xlogreader->readLen = readLen; + Assert(state->readPagePtr == targetPagePtr); + state->readLen = readLen; return true; next_record_is_invalid: @@ -11679,14 +11672,13 @@ next_record_is_invalid: if (readFile >= 0) close(readFile); readFile = -1; - readLen = 0; readSource = 0; /* In standby-mode, keep trying */ if (StandbyMode) goto retry; - xlogreader->readLen = -1; + state->readLen = -1; return false; } @@ -11718,7 +11710,8 @@ next_record_is_invalid: */ static bool WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, - bool fetching_ckpt, XLogRecPtr tliRecPtr) + bool fetching_ckpt, XLogRecPtr tliRecPtr, + XLogSegNo readSegNo) { static TimestampTz last_fail_time = 0; TimestampTz now; diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index 8e39b5c1f6..dc84d39f60 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -133,6 +133,7 @@ struct XLogReaderState * read by reader, which must be larger than * the request, or -1 on error */ TimeLineID readPageTLI; /* TLI for data currently in readBuf */ + XLogSegNo readSegNo; /* Segment # for data currently in readBuf */ char *readBuf; /* buffer to store data */ bool page_verified; /* is the page header on the buffer verified? */ bool record_verified;/* is the current record header verified? */ -- 2.16.3 ----Next_Part(Fri_Sep_27_12_07_26_2019_308)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v8-0004-Change-policy-of-XLog-read-buffer-allocation.patch" ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-02-10 18:17 Tomas Vondra <[email protected]> 0 siblings, 1 reply; 17+ messages in thread From: Tomas Vondra @ 2022-02-10 18:17 UTC (permalink / raw) To: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; +Cc: Amit Kapila <[email protected]>; PostgreSQL Hackers <[email protected]> I've polished & pushed the first part adding sequence decoding infrastructure etc. Attached are the two remaining parts. I plan to wait a day or two and then push the test_decoding part. The last part (for built-in replication) will need more work and maybe rethinking the grammar etc. regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company Attachments: [text/x-patch] 0001-Add-support-for-decoding-sequences-to-test_-20220210.patch (253.2K, ../../[email protected]/2-0001-Add-support-for-decoding-sequences-to-test_-20220210.patch) download | inline diff: From e22abd743b923f8b590948db0eac3e296d1b7b04 Mon Sep 17 00:00:00 2001 From: Tomas Vondra <[email protected]> Date: Fri, 24 Sep 2021 00:42:04 +0200 Subject: [PATCH 1/2] Add support for decoding sequences to test_decoding --- contrib/test_decoding/Makefile | 3 +- contrib/test_decoding/expected/ddl.out | 12 +- .../expected/decoding_in_xact.out | 2 +- .../expected/decoding_into_rel.out | 10 +- contrib/test_decoding/expected/mxact.out | 8 +- .../test_decoding/expected/ondisk_startup.out | 4 +- contrib/test_decoding/expected/replorigin.out | 4 +- contrib/test_decoding/expected/rewrite.out | 4 +- contrib/test_decoding/expected/sequence.out | 327 ++++++++++++++++++ contrib/test_decoding/expected/slot.out | 2 +- contrib/test_decoding/expected/toast.out | 10 +- contrib/test_decoding/expected/truncate.out | 2 +- contrib/test_decoding/specs/mxact.spec | 2 +- .../test_decoding/specs/ondisk_startup.spec | 2 +- contrib/test_decoding/sql/ddl.sql | 12 +- .../test_decoding/sql/decoding_in_xact.sql | 2 +- .../test_decoding/sql/decoding_into_rel.sql | 10 +- contrib/test_decoding/sql/replorigin.sql | 4 +- contrib/test_decoding/sql/rewrite.sql | 4 +- contrib/test_decoding/sql/sequence.sql | 119 +++++++ contrib/test_decoding/sql/slot.sql | 2 +- contrib/test_decoding/sql/toast.sql | 10 +- contrib/test_decoding/sql/truncate.sql | 2 +- contrib/test_decoding/test_decoding.c | 67 ++++ 24 files changed, 569 insertions(+), 55 deletions(-) create mode 100644 contrib/test_decoding/expected/sequence.out create mode 100644 contrib/test_decoding/sql/sequence.sql diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile index 9a31e0b8795..56ddc3abaeb 100644 --- a/contrib/test_decoding/Makefile +++ b/contrib/test_decoding/Makefile @@ -5,7 +5,8 @@ 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 \ - spill slot truncate stream stats twophase twophase_stream + spill slot truncate stream stats twophase twophase_stream \ + sequence ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \ oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \ twophase_snapshot diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out index 4ff0044c787..82898201ca0 100644 --- a/contrib/test_decoding/expected/ddl.out +++ b/contrib/test_decoding/expected/ddl.out @@ -89,7 +89,7 @@ COMMIT; ALTER TABLE replication_example RENAME COLUMN text TO somenum; INSERT INTO replication_example(somedata, somenum) VALUES (4, 1); -- collect all changes -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data --------------------------------------------------------------------------------------------------------------------------- BEGIN @@ -196,7 +196,7 @@ CREATE TABLE tr_unique(id2 serial unique NOT NULL, data int); INSERT INTO tr_unique(data) VALUES(10); ALTER TABLE tr_unique RENAME TO tr_pkey; ALTER TABLE tr_pkey ADD COLUMN id serial primary key; -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-rewrites', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-rewrites', '1', 'include-sequences', '0'); data ----------------------------------------------------------------------------- BEGIN @@ -242,7 +242,7 @@ INSERT INTO tr_oddlength VALUES('ab', 'foo'); COMMIT; /* display results, but hide most of the output */ SELECT count(*), min(data), max(data) -FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1') +FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0') GROUP BY substring(data, 1, 24) ORDER BY 1,2; count | min | max @@ -309,7 +309,7 @@ RELEASE SAVEPOINT c; INSERT INTO tr_sub(path) VALUES ('1-top-2-#1'); RELEASE SAVEPOINT b; COMMIT; -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data ---------------------------------------------------------------------- BEGIN @@ -496,7 +496,7 @@ Options: user_catalog_table=false INSERT INTO replication_metadata(relation, options) VALUES ('zaphod', NULL); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data ------------------------------------------------------------------------------------------------------------------------------------ BEGIN @@ -613,7 +613,7 @@ INSERT INTO toasttable(toasted_col2) SELECT repeat(string_agg(to_char(g.i, 'FM00 UPDATE toasttable SET toasted_col1 = (SELECT string_agg(g.i::text, '') FROM generate_series(1, 2000) g(i)) WHERE id = 1; -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ BEGIN diff --git a/contrib/test_decoding/expected/decoding_in_xact.out b/contrib/test_decoding/expected/decoding_in_xact.out index b65253f4630..6e97b6e34bc 100644 --- a/contrib/test_decoding/expected/decoding_in_xact.out +++ b/contrib/test_decoding/expected/decoding_in_xact.out @@ -58,7 +58,7 @@ SELECT pg_current_xact_id() = '0'; -- don't show yet, haven't committed INSERT INTO nobarf(data) VALUES('2'); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data ----------------------------------------------------------- BEGIN diff --git a/contrib/test_decoding/expected/decoding_into_rel.out b/contrib/test_decoding/expected/decoding_into_rel.out index 8fd3390066d..03966b8b1ca 100644 --- a/contrib/test_decoding/expected/decoding_into_rel.out +++ b/contrib/test_decoding/expected/decoding_into_rel.out @@ -19,7 +19,7 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc CREATE TABLE somechange(id serial primary key); INSERT INTO somechange DEFAULT VALUES; CREATE TABLE changeresult AS - SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT * FROM changeresult; data ------------------------------------------------ @@ -29,9 +29,9 @@ SELECT * FROM changeresult; (3 rows) INSERT INTO changeresult - SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); INSERT INTO changeresult - SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT * FROM changeresult; data -------------------------------------------------------------------------------------------------------------------------------------------------- @@ -63,7 +63,7 @@ DROP TABLE somechange; CREATE FUNCTION slot_changes_wrapper(slot_name name) RETURNS SETOF TEXT AS $$ BEGIN RETURN QUERY - SELECT data FROM pg_logical_slot_peek_changes(slot_name, NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + SELECT data FROM pg_logical_slot_peek_changes(slot_name, NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); END$$ LANGUAGE plpgsql; SELECT * FROM slot_changes_wrapper('regression_slot'); slot_changes_wrapper @@ -84,7 +84,7 @@ SELECT * FROM slot_changes_wrapper('regression_slot'); COMMIT (14 rows) -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- BEGIN diff --git a/contrib/test_decoding/expected/mxact.out b/contrib/test_decoding/expected/mxact.out index 03ad3df0999..c5bc26c8044 100644 --- a/contrib/test_decoding/expected/mxact.out +++ b/contrib/test_decoding/expected/mxact.out @@ -7,7 +7,7 @@ step s0init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_sl init (1 row) -step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false'); +step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'include-sequences', 'false'); data ---- (0 rows) @@ -27,7 +27,7 @@ t (1 row) step s0w: INSERT INTO do_write DEFAULT VALUES; -step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false'); +step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'include-sequences', 'false'); data -------------------------------------------- BEGIN @@ -50,7 +50,7 @@ step s0init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_sl init (1 row) -step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false'); +step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'include-sequences', 'false'); data ---- (0 rows) @@ -71,7 +71,7 @@ t step s0alter: ALTER TABLE do_write ADD column ts timestamptz; step s0w: INSERT INTO do_write DEFAULT VALUES; -step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false'); +step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'include-sequences', 'false'); data ------------------------------------------------------------------------------ BEGIN diff --git a/contrib/test_decoding/expected/ondisk_startup.out b/contrib/test_decoding/expected/ondisk_startup.out index bc7ff071648..3d2fa4a92e0 100644 --- a/contrib/test_decoding/expected/ondisk_startup.out +++ b/contrib/test_decoding/expected/ondisk_startup.out @@ -35,7 +35,7 @@ init step s2c: COMMIT; step s1insert: INSERT INTO do_write DEFAULT VALUES; step s1checkpoint: CHECKPOINT; -step s1start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false'); +step s1start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'include-sequences', 'false'); data -------------------------------------------------------------------- BEGIN @@ -46,7 +46,7 @@ COMMIT step s1insert: INSERT INTO do_write DEFAULT VALUES; step s1alter: ALTER TABLE do_write ADD COLUMN addedbys1 int; step s1insert: INSERT INTO do_write DEFAULT VALUES; -step s1start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false'); +step s1start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'include-sequences', 'false'); data -------------------------------------------------------------------------------------------- BEGIN diff --git a/contrib/test_decoding/expected/replorigin.out b/contrib/test_decoding/expected/replorigin.out index 2e9ef7c823b..fb96aa31727 100644 --- a/contrib/test_decoding/expected/replorigin.out +++ b/contrib/test_decoding/expected/replorigin.out @@ -72,9 +72,9 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_d -- origin tx INSERT INTO origin_tbl(data) VALUES ('will be replicated and decoded and decoded again'); INSERT INTO target_tbl(data) -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); -- as is normal, the insert into target_tbl shows up -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- BEGIN diff --git a/contrib/test_decoding/expected/rewrite.out b/contrib/test_decoding/expected/rewrite.out index b30999c436b..0b5eade41fa 100644 --- a/contrib/test_decoding/expected/rewrite.out +++ b/contrib/test_decoding/expected/rewrite.out @@ -64,7 +64,7 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_d CREATE TABLE replication_example(id SERIAL PRIMARY KEY, somedata int, text varchar(120)); INSERT INTO replication_example(somedata) VALUES (1); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data ---------------------------------------------------------------------------------------------------------- BEGIN @@ -141,7 +141,7 @@ VACUUM FULL pg_proc; VACUUM FULL pg_description; VACUUM FULL pg_shdescription; V INSERT INTO replication_example(somedata, testcolumn1, testcolumn3) VALUES (8, 6, 1); VACUUM FULL pg_proc; VACUUM FULL pg_description; VACUUM FULL pg_shdescription; VACUUM FULL iamalargetable; INSERT INTO replication_example(somedata, testcolumn1, testcolumn3) VALUES (9, 7, 1); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- BEGIN diff --git a/contrib/test_decoding/expected/sequence.out b/contrib/test_decoding/expected/sequence.out new file mode 100644 index 00000000000..cf21f3a1f4f --- /dev/null +++ b/contrib/test_decoding/expected/sequence.out @@ -0,0 +1,327 @@ +-- predictability +SET synchronous_commit = on; +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding'); + ?column? +---------- + init +(1 row) + +CREATE SEQUENCE test_sequence; +-- test the sequence changes by several nextval() calls +SELECT nextval('test_sequence'); + nextval +--------- + 1 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 2 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 4 +(1 row) + +-- test the sequence changes by several ALTER commands +ALTER SEQUENCE test_sequence INCREMENT BY 10; +SELECT nextval('test_sequence'); + nextval +--------- + 14 +(1 row) + +ALTER SEQUENCE test_sequence START WITH 3000; +ALTER SEQUENCE test_sequence MAXVALUE 10000; +ALTER SEQUENCE test_sequence RESTART WITH 4000; +SELECT nextval('test_sequence'); + nextval +--------- + 4000 +(1 row) + +-- test the sequence changes by several setval() calls +SELECT setval('test_sequence', 3500); + setval +-------- + 3500 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3510 +(1 row) + +SELECT setval('test_sequence', 3500, true); + setval +-------- + 3500 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3510 +(1 row) + +SELECT setval('test_sequence', 3500, false); + setval +-------- + 3500 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3500 +(1 row) + +-- show results and drop sequence +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0'); + data +---------------------------------------------------------------------------------------- + BEGIN + sequence public.test_sequence: transactional:1 last_value: 1 log_cnt: 0 is_called:0 + COMMIT + sequence public.test_sequence: transactional:0 last_value: 33 log_cnt: 0 is_called:1 + BEGIN + sequence public.test_sequence: transactional:1 last_value: 4 log_cnt: 0 is_called:1 + COMMIT + sequence public.test_sequence: transactional:0 last_value: 334 log_cnt: 0 is_called:1 + BEGIN + sequence public.test_sequence: transactional:1 last_value: 14 log_cnt: 32 is_called:1 + COMMIT + BEGIN + sequence public.test_sequence: transactional:1 last_value: 14 log_cnt: 0 is_called:1 + COMMIT + BEGIN + sequence public.test_sequence: transactional:1 last_value: 4000 log_cnt: 0 is_called:0 + COMMIT + sequence public.test_sequence: transactional:0 last_value: 4320 log_cnt: 0 is_called:1 + sequence public.test_sequence: transactional:0 last_value: 3500 log_cnt: 0 is_called:1 + sequence public.test_sequence: transactional:0 last_value: 3830 log_cnt: 0 is_called:1 + sequence public.test_sequence: transactional:0 last_value: 3500 log_cnt: 0 is_called:1 + sequence public.test_sequence: transactional:0 last_value: 3830 log_cnt: 0 is_called:1 + sequence public.test_sequence: transactional:0 last_value: 3500 log_cnt: 0 is_called:0 + sequence public.test_sequence: transactional:0 last_value: 3820 log_cnt: 0 is_called:1 +(24 rows) + +DROP SEQUENCE test_sequence; +-- rollback on sequence creation and update +BEGIN; +CREATE SEQUENCE test_sequence; +CREATE TABLE test_table (a INT); +SELECT nextval('test_sequence'); + nextval +--------- + 1 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 2 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3 +(1 row) + +SELECT setval('test_sequence', 3000); + setval +-------- + 3000 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3001 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3002 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3003 +(1 row) + +ALTER SEQUENCE test_sequence RESTART WITH 6000; +INSERT INTO test_table VALUES( (SELECT nextval('test_sequence')) ); +SELECT nextval('test_sequence'); + nextval +--------- + 6001 +(1 row) + +ROLLBACK; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + data +------ +(0 rows) + +-- rollback on table creation with serial column +BEGIN; +CREATE TABLE test_table (a SERIAL, b INT); +INSERT INTO test_table (b) VALUES (100); +INSERT INTO test_table (b) VALUES (200); +INSERT INTO test_table (b) VALUES (300); +SELECT setval('test_table_a_seq', 3000); + setval +-------- + 3000 +(1 row) + +INSERT INTO test_table (b) VALUES (400); +INSERT INTO test_table (b) VALUES (500); +INSERT INTO test_table (b) VALUES (600); +ALTER SEQUENCE test_table_a_seq RESTART WITH 6000; +INSERT INTO test_table (b) VALUES (700); +INSERT INTO test_table (b) VALUES (800); +INSERT INTO test_table (b) VALUES (900); +ROLLBACK; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + data +------ +(0 rows) + +-- rollback on table with serial column +CREATE TABLE test_table (a SERIAL, b INT); +BEGIN; +INSERT INTO test_table (b) VALUES (100); +INSERT INTO test_table (b) VALUES (200); +INSERT INTO test_table (b) VALUES (300); +SELECT setval('test_table_a_seq', 3000); + setval +-------- + 3000 +(1 row) + +INSERT INTO test_table (b) VALUES (400); +INSERT INTO test_table (b) VALUES (500); +INSERT INTO test_table (b) VALUES (600); +ALTER SEQUENCE test_table_a_seq RESTART WITH 6000; +INSERT INTO test_table (b) VALUES (700); +INSERT INTO test_table (b) VALUES (800); +INSERT INTO test_table (b) VALUES (900); +ROLLBACK; +-- check table and sequence values after rollback +SELECT * from test_table_a_seq; + last_value | log_cnt | is_called +------------+---------+----------- + 3003 | 30 | t +(1 row) + +SELECT nextval('test_table_a_seq'); + nextval +--------- + 3004 +(1 row) + +DROP TABLE test_table; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0'); + data +------------------------------------------------------------------------------------------- + BEGIN + sequence public.test_table_a_seq: transactional:1 last_value: 1 log_cnt: 0 is_called:0 + COMMIT + sequence public.test_table_a_seq: transactional:0 last_value: 33 log_cnt: 0 is_called:1 + sequence public.test_table_a_seq: transactional:0 last_value: 3000 log_cnt: 0 is_called:1 + sequence public.test_table_a_seq: transactional:0 last_value: 3033 log_cnt: 0 is_called:1 + BEGIN + COMMIT +(8 rows) + +-- savepoint test on table with serial column +BEGIN; +CREATE TABLE test_table (a SERIAL, b INT); +INSERT INTO test_table (b) VALUES (100); +INSERT INTO test_table (b) VALUES (200); +SAVEPOINT a; +INSERT INTO test_table (b) VALUES (300); +ROLLBACK TO SAVEPOINT a; +DROP TABLE test_table; +COMMIT; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0'); + data +----------------------------------------------------------------------------------------- + BEGIN + sequence public.test_table_a_seq: transactional:1 last_value: 1 log_cnt: 0 is_called:0 + sequence public.test_table_a_seq: transactional:1 last_value: 33 log_cnt: 0 is_called:1 + table public.test_table: INSERT: a[integer]:1 b[integer]:100 + table public.test_table: INSERT: a[integer]:2 b[integer]:200 + COMMIT +(6 rows) + +-- savepoint test on table with serial column +BEGIN; +CREATE SEQUENCE test_sequence; +SELECT nextval('test_sequence'); + nextval +--------- + 1 +(1 row) + +SELECT setval('test_sequence', 3000); + setval +-------- + 3000 +(1 row) + +SELECT nextval('test_sequence'); + nextval +--------- + 3001 +(1 row) + +SAVEPOINT a; +ALTER SEQUENCE test_sequence START WITH 7000; +SELECT setval('test_sequence', 5000); + setval +-------- + 5000 +(1 row) + +ROLLBACK TO SAVEPOINT a; +SELECT * FROM test_sequence; + last_value | log_cnt | is_called +------------+---------+----------- + 3001 | 32 | t +(1 row) + +DROP SEQUENCE test_sequence; +COMMIT; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0'); + data +---------------------------------------------------------------------------------------- + BEGIN + sequence public.test_sequence: transactional:1 last_value: 1 log_cnt: 0 is_called:0 + sequence public.test_sequence: transactional:1 last_value: 33 log_cnt: 0 is_called:1 + sequence public.test_sequence: transactional:1 last_value: 3000 log_cnt: 0 is_called:1 + sequence public.test_sequence: transactional:1 last_value: 3033 log_cnt: 0 is_called:1 + COMMIT +(6 rows) + +SELECT pg_drop_replication_slot('regression_slot'); + pg_drop_replication_slot +-------------------------- + +(1 row) + diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out index 63a9940f73a..93d7b95d47a 100644 --- a/contrib/test_decoding/expected/slot.out +++ b/contrib/test_decoding/expected/slot.out @@ -95,7 +95,7 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot2', 'test_ (1 row) INSERT INTO replication_example(somedata, text) VALUES (1, 3); -SELECT data FROM pg_logical_slot_get_changes('regression_slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data --------------------------------------------------------------------------------------------------------- BEGIN diff --git a/contrib/test_decoding/expected/toast.out b/contrib/test_decoding/expected/toast.out index cd03e9d50a1..a060e3b5e52 100644 --- a/contrib/test_decoding/expected/toast.out +++ b/contrib/test_decoding/expected/toast.out @@ -52,7 +52,7 @@ CREATE TABLE toasted_copy ( ); ALTER TABLE toasted_copy ALTER COLUMN data SET STORAGE EXTERNAL; \copy toasted_copy FROM STDIN -SELECT substr(data, 1, 200) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT substr(data, 1, 200) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); substr ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- BEGIN @@ -316,7 +316,7 @@ SELECT pg_column_size(toasted_key) > 2^16 FROM toasted_several; t (1 row) -SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); regexp_replace ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ BEGIN @@ -327,7 +327,7 @@ SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_sl -- test update of a toasted key without changing it UPDATE toasted_several SET toasted_col1 = toasted_key; UPDATE toasted_several SET toasted_col2 = toasted_col1; -SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); regexp_replace ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ BEGIN @@ -350,7 +350,7 @@ UPDATE toasted_several SET toasted_col1 = toasted_col2 WHERE id = 1; DELETE FROM toasted_several WHERE id = 1; COMMIT; DROP TABLE toasted_several; -SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1') +SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0') WHERE data NOT LIKE '%INSERT: %'; regexp_replace ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ @@ -373,7 +373,7 @@ INSERT INTO tbl1 VALUES(1, repeat('a', 4000)) ; ALTER TABLE tbl1 ADD COLUMN id serial primary key; INSERT INTO tbl2 VALUES(1); commit; -SELECT substr(data, 1, 200) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT substr(data, 1, 200) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); substr ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- BEGIN diff --git a/contrib/test_decoding/expected/truncate.out b/contrib/test_decoding/expected/truncate.out index e64d377214a..961b0ebdc89 100644 --- a/contrib/test_decoding/expected/truncate.out +++ b/contrib/test_decoding/expected/truncate.out @@ -11,7 +11,7 @@ CREATE TABLE tab2 (a int primary key, b int); TRUNCATE tab1; TRUNCATE tab1, tab1 RESTART IDENTITY CASCADE; TRUNCATE tab1, tab2; -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); data ------------------------------------------------------ BEGIN diff --git a/contrib/test_decoding/specs/mxact.spec b/contrib/test_decoding/specs/mxact.spec index ea5b1aa2d67..19f3af14b30 100644 --- a/contrib/test_decoding/specs/mxact.spec +++ b/contrib/test_decoding/specs/mxact.spec @@ -13,7 +13,7 @@ teardown session "s0" setup { SET synchronous_commit=on; } step "s0init" {SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding');} -step "s0start" {SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false');} +step "s0start" {SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'include-sequences', 'false');} step "s0alter" {ALTER TABLE do_write ADD column ts timestamptz; } step "s0w" { INSERT INTO do_write DEFAULT VALUES; } diff --git a/contrib/test_decoding/specs/ondisk_startup.spec b/contrib/test_decoding/specs/ondisk_startup.spec index 96ce87f1a45..fb43cbc5e15 100644 --- a/contrib/test_decoding/specs/ondisk_startup.spec +++ b/contrib/test_decoding/specs/ondisk_startup.spec @@ -16,7 +16,7 @@ session "s1" setup { SET synchronous_commit=on; } step "s1init" {SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding');} -step "s1start" {SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false');} +step "s1start" {SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'include-sequences', 'false');} step "s1insert" { INSERT INTO do_write DEFAULT VALUES; } step "s1checkpoint" { CHECKPOINT; } step "s1alter" { ALTER TABLE do_write ADD COLUMN addedbys1 int; } diff --git a/contrib/test_decoding/sql/ddl.sql b/contrib/test_decoding/sql/ddl.sql index 1b3866d0153..f677460d349 100644 --- a/contrib/test_decoding/sql/ddl.sql +++ b/contrib/test_decoding/sql/ddl.sql @@ -64,7 +64,7 @@ ALTER TABLE replication_example RENAME COLUMN text TO somenum; INSERT INTO replication_example(somedata, somenum) VALUES (4, 1); -- collect all changes -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); ALTER TABLE replication_example ALTER COLUMN somenum TYPE int4 USING (somenum::int4); -- check that this doesn't produce any changes from the heap rewrite @@ -97,7 +97,7 @@ CREATE TABLE tr_unique(id2 serial unique NOT NULL, data int); INSERT INTO tr_unique(data) VALUES(10); ALTER TABLE tr_unique RENAME TO tr_pkey; ALTER TABLE tr_pkey ADD COLUMN id serial primary key; -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-rewrites', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-rewrites', '1', 'include-sequences', '0'); INSERT INTO tr_pkey(data) VALUES(1); --show deletion with primary key @@ -121,7 +121,7 @@ COMMIT; /* display results, but hide most of the output */ SELECT count(*), min(data), max(data) -FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1') +FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0') GROUP BY substring(data, 1, 24) ORDER BY 1,2; @@ -173,7 +173,7 @@ INSERT INTO tr_sub(path) VALUES ('1-top-2-#1'); RELEASE SAVEPOINT b; COMMIT; -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); -- check that we handle xlog assignments correctly BEGIN; @@ -287,7 +287,7 @@ ALTER TABLE replication_metadata SET (user_catalog_table = false); INSERT INTO replication_metadata(relation, options) VALUES ('zaphod', NULL); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); /* * check whether we handle updates/deletes correct with & without a pkey @@ -398,7 +398,7 @@ UPDATE toasttable SET toasted_col1 = (SELECT string_agg(g.i::text, '') FROM generate_series(1, 2000) g(i)) WHERE id = 1; -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); INSERT INTO toasttable(toasted_col1) SELECT string_agg(g.i::text, '') FROM generate_series(1, 2000) g(i); diff --git a/contrib/test_decoding/sql/decoding_in_xact.sql b/contrib/test_decoding/sql/decoding_in_xact.sql index 108782dc2e9..33a9c4a6c77 100644 --- a/contrib/test_decoding/sql/decoding_in_xact.sql +++ b/contrib/test_decoding/sql/decoding_in_xact.sql @@ -32,7 +32,7 @@ BEGIN; SELECT pg_current_xact_id() = '0'; -- don't show yet, haven't committed INSERT INTO nobarf(data) VALUES('2'); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); COMMIT; INSERT INTO nobarf(data) VALUES('3'); diff --git a/contrib/test_decoding/sql/decoding_into_rel.sql b/contrib/test_decoding/sql/decoding_into_rel.sql index 1068cec5888..27d5d08879e 100644 --- a/contrib/test_decoding/sql/decoding_into_rel.sql +++ b/contrib/test_decoding/sql/decoding_into_rel.sql @@ -15,14 +15,14 @@ CREATE TABLE somechange(id serial primary key); INSERT INTO somechange DEFAULT VALUES; CREATE TABLE changeresult AS - SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT * FROM changeresult; INSERT INTO changeresult - SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); INSERT INTO changeresult - SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT * FROM changeresult; DROP TABLE changeresult; @@ -32,11 +32,11 @@ DROP TABLE somechange; CREATE FUNCTION slot_changes_wrapper(slot_name name) RETURNS SETOF TEXT AS $$ BEGIN RETURN QUERY - SELECT data FROM pg_logical_slot_peek_changes(slot_name, NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + SELECT data FROM pg_logical_slot_peek_changes(slot_name, NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); END$$ LANGUAGE plpgsql; SELECT * FROM slot_changes_wrapper('regression_slot'); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT 'stop' FROM pg_drop_replication_slot('regression_slot'); diff --git a/contrib/test_decoding/sql/replorigin.sql b/contrib/test_decoding/sql/replorigin.sql index 2e28a487773..f0f4dd49642 100644 --- a/contrib/test_decoding/sql/replorigin.sql +++ b/contrib/test_decoding/sql/replorigin.sql @@ -41,10 +41,10 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_d -- origin tx INSERT INTO origin_tbl(data) VALUES ('will be replicated and decoded and decoded again'); INSERT INTO target_tbl(data) -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); -- as is normal, the insert into target_tbl shows up -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); INSERT INTO origin_tbl(data) VALUES ('will be replicated, but not decoded again'); diff --git a/contrib/test_decoding/sql/rewrite.sql b/contrib/test_decoding/sql/rewrite.sql index 62dead3a9b1..945c39eb416 100644 --- a/contrib/test_decoding/sql/rewrite.sql +++ b/contrib/test_decoding/sql/rewrite.sql @@ -35,7 +35,7 @@ SELECT pg_relation_size((SELECT reltoastrelid FROM pg_class WHERE oid = 'pg_shde SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding'); CREATE TABLE replication_example(id SERIAL PRIMARY KEY, somedata int, text varchar(120)); INSERT INTO replication_example(somedata) VALUES (1); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); BEGIN; INSERT INTO replication_example(somedata) VALUES (2); @@ -98,7 +98,7 @@ VACUUM FULL pg_proc; VACUUM FULL pg_description; VACUUM FULL pg_shdescription; V INSERT INTO replication_example(somedata, testcolumn1, testcolumn3) VALUES (8, 6, 1); VACUUM FULL pg_proc; VACUUM FULL pg_description; VACUUM FULL pg_shdescription; VACUUM FULL iamalargetable; INSERT INTO replication_example(somedata, testcolumn1, testcolumn3) VALUES (9, 7, 1); -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT pg_drop_replication_slot('regression_slot'); DROP TABLE IF EXISTS replication_example; diff --git a/contrib/test_decoding/sql/sequence.sql b/contrib/test_decoding/sql/sequence.sql new file mode 100644 index 00000000000..42ad9113bf4 --- /dev/null +++ b/contrib/test_decoding/sql/sequence.sql @@ -0,0 +1,119 @@ +-- predictability +SET synchronous_commit = on; +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding'); + +CREATE SEQUENCE test_sequence; + +-- test the sequence changes by several nextval() calls +SELECT nextval('test_sequence'); +SELECT nextval('test_sequence'); +SELECT nextval('test_sequence'); +SELECT nextval('test_sequence'); + +-- test the sequence changes by several ALTER commands +ALTER SEQUENCE test_sequence INCREMENT BY 10; +SELECT nextval('test_sequence'); + +ALTER SEQUENCE test_sequence START WITH 3000; +ALTER SEQUENCE test_sequence MAXVALUE 10000; +ALTER SEQUENCE test_sequence RESTART WITH 4000; +SELECT nextval('test_sequence'); + +-- test the sequence changes by several setval() calls +SELECT setval('test_sequence', 3500); +SELECT nextval('test_sequence'); +SELECT setval('test_sequence', 3500, true); +SELECT nextval('test_sequence'); +SELECT setval('test_sequence', 3500, false); +SELECT nextval('test_sequence'); + +-- show results and drop sequence +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0'); +DROP SEQUENCE test_sequence; + +-- rollback on sequence creation and update +BEGIN; +CREATE SEQUENCE test_sequence; +CREATE TABLE test_table (a INT); +SELECT nextval('test_sequence'); +SELECT nextval('test_sequence'); +SELECT nextval('test_sequence'); +SELECT setval('test_sequence', 3000); +SELECT nextval('test_sequence'); +SELECT nextval('test_sequence'); +SELECT nextval('test_sequence'); +ALTER SEQUENCE test_sequence RESTART WITH 6000; +INSERT INTO test_table VALUES( (SELECT nextval('test_sequence')) ); +SELECT nextval('test_sequence'); +ROLLBACK; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + +-- rollback on table creation with serial column +BEGIN; +CREATE TABLE test_table (a SERIAL, b INT); +INSERT INTO test_table (b) VALUES (100); +INSERT INTO test_table (b) VALUES (200); +INSERT INTO test_table (b) VALUES (300); +SELECT setval('test_table_a_seq', 3000); +INSERT INTO test_table (b) VALUES (400); +INSERT INTO test_table (b) VALUES (500); +INSERT INTO test_table (b) VALUES (600); +ALTER SEQUENCE test_table_a_seq RESTART WITH 6000; +INSERT INTO test_table (b) VALUES (700); +INSERT INTO test_table (b) VALUES (800); +INSERT INTO test_table (b) VALUES (900); +ROLLBACK; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + +-- rollback on table with serial column +CREATE TABLE test_table (a SERIAL, b INT); + +BEGIN; +INSERT INTO test_table (b) VALUES (100); +INSERT INTO test_table (b) VALUES (200); +INSERT INTO test_table (b) VALUES (300); +SELECT setval('test_table_a_seq', 3000); +INSERT INTO test_table (b) VALUES (400); +INSERT INTO test_table (b) VALUES (500); +INSERT INTO test_table (b) VALUES (600); +ALTER SEQUENCE test_table_a_seq RESTART WITH 6000; +INSERT INTO test_table (b) VALUES (700); +INSERT INTO test_table (b) VALUES (800); +INSERT INTO test_table (b) VALUES (900); +ROLLBACK; + +-- check table and sequence values after rollback +SELECT * from test_table_a_seq; +SELECT nextval('test_table_a_seq'); + +DROP TABLE test_table; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0'); + +-- savepoint test on table with serial column +BEGIN; +CREATE TABLE test_table (a SERIAL, b INT); +INSERT INTO test_table (b) VALUES (100); +INSERT INTO test_table (b) VALUES (200); +SAVEPOINT a; +INSERT INTO test_table (b) VALUES (300); +ROLLBACK TO SAVEPOINT a; +DROP TABLE test_table; +COMMIT; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0'); + +-- savepoint test on table with serial column +BEGIN; +CREATE SEQUENCE test_sequence; +SELECT nextval('test_sequence'); +SELECT setval('test_sequence', 3000); +SELECT nextval('test_sequence'); +SAVEPOINT a; +ALTER SEQUENCE test_sequence START WITH 7000; +SELECT setval('test_sequence', 5000); +ROLLBACK TO SAVEPOINT a; +SELECT * FROM test_sequence; +DROP SEQUENCE test_sequence; +COMMIT; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0'); + +SELECT pg_drop_replication_slot('regression_slot'); diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql index 1aa27c56674..70ea1603e67 100644 --- a/contrib/test_decoding/sql/slot.sql +++ b/contrib/test_decoding/sql/slot.sql @@ -50,7 +50,7 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot2', 'test_ INSERT INTO replication_example(somedata, text) VALUES (1, 3); -SELECT data FROM pg_logical_slot_get_changes('regression_slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot1', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT data FROM pg_logical_slot_get_changes('regression_slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); INSERT INTO replication_example(somedata, text) VALUES (1, 4); diff --git a/contrib/test_decoding/sql/toast.sql b/contrib/test_decoding/sql/toast.sql index d1c560a174d..f5d4fc082ed 100644 --- a/contrib/test_decoding/sql/toast.sql +++ b/contrib/test_decoding/sql/toast.sql @@ -264,7 +264,7 @@ ALTER TABLE toasted_copy ALTER COLUMN data SET STORAGE EXTERNAL; 202 untoasted199 203 untoasted200 \. -SELECT substr(data, 1, 200) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT substr(data, 1, 200) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); -- test we can decode "old" tuples bigger than the max heap tuple size correctly DROP TABLE IF EXISTS toasted_several; @@ -287,13 +287,13 @@ UPDATE pg_attribute SET attstorage = 'x' WHERE attrelid = 'toasted_several_pkey' INSERT INTO toasted_several(toasted_key) VALUES(repeat('9876543210', 10000)); SELECT pg_column_size(toasted_key) > 2^16 FROM toasted_several; -SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); -- test update of a toasted key without changing it UPDATE toasted_several SET toasted_col1 = toasted_key; UPDATE toasted_several SET toasted_col2 = toasted_col1; -SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); /* * update with large tuplebuf, in a transaction large enough to force to spool to disk @@ -306,7 +306,7 @@ COMMIT; DROP TABLE toasted_several; -SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1') +SELECT regexp_replace(data, '^(.{100}).*(.{100})$', '\1..\2') FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0') WHERE data NOT LIKE '%INSERT: %'; /* @@ -322,6 +322,6 @@ INSERT INTO tbl1 VALUES(1, repeat('a', 4000)) ; ALTER TABLE tbl1 ADD COLUMN id serial primary key; INSERT INTO tbl2 VALUES(1); commit; -SELECT substr(data, 1, 200) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT substr(data, 1, 200) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT pg_drop_replication_slot('regression_slot'); diff --git a/contrib/test_decoding/sql/truncate.sql b/contrib/test_decoding/sql/truncate.sql index 5633854e0df..b0f36fd681b 100644 --- a/contrib/test_decoding/sql/truncate.sql +++ b/contrib/test_decoding/sql/truncate.sql @@ -10,5 +10,5 @@ TRUNCATE tab1; TRUNCATE tab1, tab1 RESTART IDENTITY CASCADE; TRUNCATE tab1, tab2; -SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'include-sequences', '0'); SELECT pg_drop_replication_slot('regression_slot'); diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c index 42fe91a2f9c..526080f3bbb 100644 --- a/contrib/test_decoding/test_decoding.c +++ b/contrib/test_decoding/test_decoding.c @@ -35,6 +35,7 @@ typedef struct bool include_timestamp; bool skip_empty_xacts; bool only_local; + bool include_sequences; } TestDecodingData; /* @@ -76,6 +77,10 @@ static void pg_decode_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message); +static void pg_decode_sequence(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr sequence_lsn, + Relation rel, bool transactional, + int64 last_value, int64 log_cnt, bool is_called); static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid); @@ -116,6 +121,10 @@ static void pg_decode_stream_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message); +static void pg_decode_stream_sequence(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr sequence_lsn, + Relation rel, bool transactional, + int64 last_value, int64 log_cnt, bool is_called); static void pg_decode_stream_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelations, Relation relations[], @@ -141,6 +150,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->filter_by_origin_cb = pg_decode_filter; cb->shutdown_cb = pg_decode_shutdown; cb->message_cb = pg_decode_message; + cb->sequence_cb = pg_decode_sequence; cb->filter_prepare_cb = pg_decode_filter_prepare; cb->begin_prepare_cb = pg_decode_begin_prepare_txn; cb->prepare_cb = pg_decode_prepare_txn; @@ -153,6 +163,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->stream_commit_cb = pg_decode_stream_commit; cb->stream_change_cb = pg_decode_stream_change; cb->stream_message_cb = pg_decode_stream_message; + cb->stream_sequence_cb = pg_decode_stream_sequence; cb->stream_truncate_cb = pg_decode_stream_truncate; } @@ -173,6 +184,7 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, data->include_xids = true; data->include_timestamp = false; data->skip_empty_xacts = false; + data->include_sequences = true; data->only_local = false; ctx->output_plugin_private = data; @@ -265,6 +277,17 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, errmsg("could not parse value \"%s\" for parameter \"%s\"", strVal(elem->arg), elem->defname))); } + else if (strcmp(elem->defname, "include-sequences") == 0) + { + + if (elem->arg == NULL) + data->include_sequences = false; + else if (!parse_bool(strVal(elem->arg), &data->include_sequences)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("could not parse value \"%s\" for parameter \"%s\"", + strVal(elem->arg), elem->defname))); + } else { ereport(ERROR, @@ -744,6 +767,28 @@ pg_decode_message(LogicalDecodingContext *ctx, OutputPluginWrite(ctx, true); } +static void +pg_decode_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + XLogRecPtr sequence_lsn, Relation rel, + bool transactional, + int64 last_value, int64 log_cnt, bool is_called) +{ + TestDecodingData *data = ctx->output_plugin_private; + + /* return if requested to skip sequences */ + if (!data->include_sequences) + return; + + OutputPluginPrepareWrite(ctx, true); + appendStringInfoString(ctx->out, "sequence "); + appendStringInfoString(ctx->out, + quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(rel))), + RelationGetRelationName(rel))); + appendStringInfo(ctx->out, ": transactional:%d last_value: " INT64_FORMAT " log_cnt: " INT64_FORMAT " is_called:%d", + transactional, last_value, log_cnt, is_called); + OutputPluginWrite(ctx, true); +} + static void pg_decode_stream_start(LogicalDecodingContext *ctx, ReorderBufferTXN *txn) @@ -943,6 +988,28 @@ pg_decode_stream_message(LogicalDecodingContext *ctx, OutputPluginWrite(ctx, true); } +static void +pg_decode_stream_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + XLogRecPtr sequence_lsn, Relation rel, + bool transactional, + int64 last_value, int64 log_cnt, bool is_called) +{ + TestDecodingData *data = ctx->output_plugin_private; + + /* return if requested to skip sequences */ + if (!data->include_sequences) + return; + + OutputPluginPrepareWrite(ctx, true); + appendStringInfoString(ctx->out, "streaming sequence "); + appendStringInfoString(ctx->out, + quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(rel))), + RelationGetRelationName(rel))); + appendStringInfo(ctx->out, ": transactional:%d last_value: " INT64_FORMAT " log_cnt: " INT64_FORMAT " is_called:%d", + transactional, last_value, log_cnt, is_called); + OutputPluginWrite(ctx, true); +} + /* * In streaming mode, we don't display the detailed information of Truncate. * See pg_decode_stream_change. -- 2.34.1 [text/x-patch] 0002-Add-support-for-decoding-sequences-to-built-20220210.patch (77.7K, ../../[email protected]/3-0002-Add-support-for-decoding-sequences-to-built-20220210.patch) download | inline diff: From 0e21d24429c95ac7769a3f66b4693c307cf942aa Mon Sep 17 00:00:00 2001 From: Tomas Vondra <[email protected]> Date: Thu, 10 Feb 2022 15:18:59 +0100 Subject: [PATCH 2/2] Add support for decoding sequences to built-in replication --- doc/src/sgml/catalogs.sgml | 71 ++++ doc/src/sgml/ref/alter_publication.sgml | 24 +- doc/src/sgml/ref/alter_subscription.sgml | 4 +- src/backend/catalog/pg_publication.c | 149 ++++++++- src/backend/catalog/system_views.sql | 10 + src/backend/commands/publicationcmds.c | 350 +++++++++++++++++++- src/backend/commands/sequence.c | 79 +++++ src/backend/commands/subscriptioncmds.c | 272 +++++++++++++++ src/backend/executor/execReplication.c | 2 +- src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/equalfuncs.c | 1 + src/backend/parser/gram.y | 32 ++ src/backend/replication/logical/proto.c | 52 +++ src/backend/replication/logical/tablesync.c | 118 ++++++- src/backend/replication/logical/worker.c | 60 ++++ src/backend/replication/pgoutput/pgoutput.c | 85 ++++- src/backend/utils/cache/relcache.c | 4 +- src/bin/psql/tab-complete.c | 14 +- src/include/catalog/pg_proc.dat | 5 + src/include/catalog/pg_publication.h | 14 + src/include/commands/sequence.h | 1 + src/include/nodes/parsenodes.h | 6 + src/include/replication/logicalproto.h | 19 ++ src/include/replication/pgoutput.h | 1 + src/test/regress/expected/rules.out | 8 + src/test/subscription/t/027_sequences.pl | 196 +++++++++++ 26 files changed, 1542 insertions(+), 36 deletions(-) create mode 100644 src/test/subscription/t/027_sequences.pl diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 879d2dbce03..271dc03e5a2 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -9540,6 +9540,11 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l <entry>prepared transactions</entry> </row> + <row> + <entry><link linkend="view-pg-publication-sequences"><structname>pg_publication_sequences</structname></link></entry> + <entry>publications and their associated sequences</entry> + </row> + <row> <entry><link linkend="view-pg-publication-tables"><structname>pg_publication_tables</structname></link></entry> <entry>publications and their associated tables</entry> @@ -11375,6 +11380,72 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx </sect1> + <sect1 id="view-pg-publication-sequences"> + <title><structname>pg_publication_sequences</structname></title> + + <indexterm zone="view-pg-publication-sequences"> + <primary>pg_publication_sequences</primary> + </indexterm> + + <para> + The view <structname>pg_publication_sequences</structname> provides + information about the mapping between publications and the sequences they + contain. Unlike the underlying catalog + <link linkend="catalog-pg-publication-rel"><structname>pg_publication_rel</structname></link>, + this view expands + publications defined as <literal>FOR ALL SEQUENCES</literal>, so for such + publications there will be a row for each eligible sequence. + </para> + + <table> + <title><structname>pg_publication_sequences</structname> Columns</title> + <tgroup cols="1"> + <thead> + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + Column Type + </para> + <para> + Description + </para></entry> + </row> + </thead> + + <tbody> + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>pubname</structfield> <type>name</type> + (references <link linkend="catalog-pg-publication"><structname>pg_publication</structname></link>.<structfield>pubname</structfield>) + </para> + <para> + Name of publication + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>schemaname</structfield> <type>name</type> + (references <link linkend="catalog-pg-namespace"><structname>pg_namespace</structname></link>.<structfield>nspname</structfield>) + </para> + <para> + Name of schema containing sequence + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>sequencename</structfield> <type>name</type> + (references <link linkend="catalog-pg-class"><structname>pg_class</structname></link>.<structfield>relname</structfield>) + </para> + <para> + Name of sequence + </para></entry> + </row> + </tbody> + </tgroup> + </table> + </sect1> + <sect1 id="view-pg-publication-tables"> <title><structname>pg_publication_tables</structname></title> diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml index 7c7c27bf7ce..9da8274ae2c 100644 --- a/doc/src/sgml/ref/alter_publication.sgml +++ b/doc/src/sgml/ref/alter_publication.sgml @@ -31,7 +31,9 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r <phrase>where <replaceable class="parameter">publication_object</replaceable> is one of:</phrase> TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [, ... ] + SEQUENCE <replaceable class="parameter">sequence_name</replaceable> [ * ] [, ... ] ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ] + ALL SEQUENCES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ] </synopsis> </refsynopsisdiv> @@ -56,7 +58,18 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r </para> <para> - The fourth variant of this command listed in the synopsis can change + The next three variants change which sequences are part of the publication. + The <literal>SET SEQUENCE</literal> clause will replace the list of sequences + in the publication with the specified one. The <literal>ADD SEQUENCE</literal> + and <literal>DROP SEQUENCE</literal> clauses will add and remove one or more + sequences from the publication. Note that adding sequences to a publication + that is already subscribed to will require a <literal>ALTER SUBSCRIPTION + ... REFRESH PUBLICATION</literal> action on the subscribing side in order + to become effective. + </para> + + <para> + The seventh variant of this command listed in the synopsis can change all of the publication properties specified in <xref linkend="sql-createpublication"/>. Properties not mentioned in the command retain their previous settings. @@ -123,6 +136,15 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r </listitem> </varlistentry> + <varlistentry> + <term><replaceable class="parameter">sequence_name</replaceable></term> + <listitem> + <para> + Name of an existing sequence. + </para> + </listitem> + </varlistentry> + <varlistentry> <term><literal>SET ( <replaceable class="parameter">publication_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] )</literal></term> <listitem> diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 0b027cc3462..8f28cf03f40 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -147,7 +147,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < <listitem> <para> Fetch missing table information from publisher. This will start - replication of tables that were added to the subscribed-to publications + replication of tables and sequences that were added to the subscribed-to publications since <command>CREATE SUBSCRIPTION</command> or the last invocation of <command>REFRESH PUBLICATION</command>. </para> @@ -164,7 +164,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < Specifies whether to copy pre-existing data in the publications that are being subscribed to when the replication starts. The default is <literal>true</literal>. (Previously-subscribed - tables are not copied.) + tables and sequences are not copied.) </para> </listitem> </varlistentry> diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index e14ca2f5630..1a9e05ba98b 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -54,7 +54,8 @@ check_publication_add_relation(Relation targetrel) { /* Must be a regular or partitioned table */ if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION && - RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE) + RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE && + RelationGetForm(targetrel)->relkind != RELKIND_SEQUENCE) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("cannot add relation \"%s\" to publication", @@ -131,7 +132,8 @@ static bool is_publishable_class(Oid relid, Form_pg_class reltuple) { return (reltuple->relkind == RELKIND_RELATION || - reltuple->relkind == RELKIND_PARTITIONED_TABLE) && + reltuple->relkind == RELKIND_PARTITIONED_TABLE || + reltuple->relkind == RELKIND_SEQUENCE) && !IsCatalogRelationOid(relid) && reltuple->relpersistence == RELPERSISTENCE_PERMANENT && relid >= FirstNormalObjectId; @@ -503,6 +505,11 @@ GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) Form_pg_publication_rel pubrel; pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + + /* skip sequences here */ + if (get_rel_relkind(pubrel->prrelid) == RELKIND_SEQUENCE) + continue; + result = GetPubPartitionOptionRelations(result, pub_partopt, pubrel->prrelid); } @@ -517,6 +524,49 @@ GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) return result; } +/* + * Gets list of relation oids for a publication (sequences only). + * + * This should only be used for normal publications, the FOR ALL TABLES + * should use GetAllSequencesPublicationRelations(). + */ +List * +GetPublicationSequenceRelations(Oid pubid) +{ + List *result; + Relation pubrelsrel; + ScanKeyData scankey; + SysScanDesc scan; + HeapTuple tup; + + /* Find all publications associated with the relation. */ + pubrelsrel = table_open(PublicationRelRelationId, AccessShareLock); + + ScanKeyInit(&scankey, + Anum_pg_publication_rel_prpubid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(pubid)); + + scan = systable_beginscan(pubrelsrel, PublicationRelPrrelidPrpubidIndexId, + true, NULL, 1, &scankey); + + result = NIL; + while (HeapTupleIsValid(tup = systable_getnext(scan))) + { + Form_pg_publication_rel pubrel; + + pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + + if (get_rel_relkind(pubrel->prrelid) == RELKIND_SEQUENCE) + result = lappend_oid(result, pubrel->prrelid); + } + + systable_endscan(scan); + table_close(pubrelsrel, AccessShareLock); + + return result; +} + /* * Gets list of publication oids for publications marked as FOR ALL TABLES. */ @@ -762,6 +812,46 @@ GetAllSchemaPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) return result; } +/* + * Gets list of all relation published by FOR ALL TABLES publication(s). + * + * If the publication publishes partition changes via their respective root + * partitioned tables, we must exclude partitions in favor of including the + * root partitioned tables. + */ +List * +GetAllSequencesPublicationRelations(void) +{ + Relation classRel; + ScanKeyData key[1]; + TableScanDesc scan; + HeapTuple tuple; + List *result = NIL; + + classRel = table_open(RelationRelationId, AccessShareLock); + + ScanKeyInit(&key[0], + Anum_pg_class_relkind, + BTEqualStrategyNumber, F_CHAREQ, + CharGetDatum(RELKIND_SEQUENCE)); + + scan = table_beginscan_catalog(classRel, 1, key); + + while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL) + { + Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple); + Oid relid = relForm->oid; + + if (is_publishable_class(relid, relForm)) + result = lappend_oid(result, relid); + } + + table_endscan(scan); + + table_close(classRel, AccessShareLock); + return result; +} + /* * Get publication using oid * @@ -784,10 +874,12 @@ GetPublication(Oid pubid) pub->oid = pubid; pub->name = pstrdup(NameStr(pubform->pubname)); pub->alltables = pubform->puballtables; + pub->allsequences = pubform->puballsequences; pub->pubactions.pubinsert = pubform->pubinsert; pub->pubactions.pubupdate = pubform->pubupdate; pub->pubactions.pubdelete = pubform->pubdelete; pub->pubactions.pubtruncate = pubform->pubtruncate; + pub->pubactions.pubsequence = pubform->pubsequence; pub->pubviaroot = pubform->pubviaroot; ReleaseSysCache(tup); @@ -937,3 +1029,56 @@ pg_get_publication_tables(PG_FUNCTION_ARGS) SRF_RETURN_DONE(funcctx); } + +/* + * Returns Oids of sequences in a publication. + */ +Datum +pg_get_publication_sequences(PG_FUNCTION_ARGS) +{ + FuncCallContext *funcctx; + char *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0)); + Publication *publication; + List *sequences; + + /* stuff done only on the first call of the function */ + if (SRF_IS_FIRSTCALL()) + { + MemoryContext oldcontext; + + /* create a function context for cross-call persistence */ + funcctx = SRF_FIRSTCALL_INIT(); + + /* switch to memory context appropriate for multiple function calls */ + oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx); + + publication = GetPublicationByName(pubname, false); + + /* + * Publications support partitioned tables, although all changes are + * replicated using leaf partition identity and schema, so we only + * need those. + */ + if (publication->allsequences) + sequences = GetAllSequencesPublicationRelations(); + else + sequences = GetPublicationSequenceRelations(publication->oid); + + funcctx->user_fctx = (void *) sequences; + + MemoryContextSwitchTo(oldcontext); + } + + /* stuff done on every call of the function */ + funcctx = SRF_PERCALL_SETUP(); + sequences = (List *) funcctx->user_fctx; + + if (funcctx->call_cntr < list_length(sequences)) + { + Oid relid = list_nth_oid(sequences, funcctx->call_cntr); + + SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid)); + } + + SRF_RETURN_DONE(funcctx); +} diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 3cb69b1f87b..b5cc33aca34 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -374,6 +374,16 @@ CREATE VIEW pg_publication_tables AS pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace) WHERE C.oid = GPT.relid; +CREATE VIEW pg_publication_sequences AS + SELECT + P.pubname AS pubname, + N.nspname AS schemaname, + C.relname AS sequencename + FROM pg_publication P, + LATERAL pg_get_publication_sequences(P.pubname) GPT, + pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace) + WHERE C.oid = GPT.relid; + CREATE VIEW pg_locks AS SELECT * FROM pg_lock_status() AS L; diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 0e4bb97fb73..3bc2e8ccb66 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -16,6 +16,7 @@ #include "access/genam.h" #include "access/htup_details.h" +#include "access/relation.h" #include "access/table.h" #include "access/xact.h" #include "catalog/catalog.h" @@ -59,6 +60,12 @@ static void PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists, AlterPublicationStmt *stmt); static void PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok); +static List *OpenSequenceList(List *sequences); +static void CloseSequenceList(List *rels); +static void PublicationAddSequences(Oid pubid, List *rels, bool if_not_exists, + AlterPublicationStmt *stmt); +static void PublicationDropSequences(Oid pubid, List *rels, bool missing_ok); + static void parse_publication_options(ParseState *pstate, List *options, @@ -77,6 +84,7 @@ parse_publication_options(ParseState *pstate, pubactions->pubupdate = true; pubactions->pubdelete = true; pubactions->pubtruncate = true; + pubactions->pubsequence = true; *publish_via_partition_root = false; /* Parse options */ @@ -101,6 +109,7 @@ parse_publication_options(ParseState *pstate, pubactions->pubupdate = false; pubactions->pubdelete = false; pubactions->pubtruncate = false; + pubactions->pubsequence = false; *publish_given = true; publish = defGetString(defel); @@ -123,6 +132,8 @@ parse_publication_options(ParseState *pstate, pubactions->pubdelete = true; else if (strcmp(publish_opt, "truncate") == 0) pubactions->pubtruncate = true; + else if (strcmp(publish_opt, "sequence") == 0) + pubactions->pubsequence = true; else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -149,7 +160,9 @@ parse_publication_options(ParseState *pstate, */ static void ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate, - List **rels, List **schemas) + List **tables, List **sequences, + List **tables_schemas, List **sequences_schemas, + List **schemas) { ListCell *cell; PublicationObjSpec *pubobj; @@ -167,12 +180,23 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate, switch (pubobj->pubobjtype) { case PUBLICATIONOBJ_TABLE: - *rels = lappend(*rels, pubobj->pubtable); + *tables = lappend(*tables, pubobj->pubtable); + break; + case PUBLICATIONOBJ_SEQUENCE: + *sequences = lappend(*sequences, pubobj->pubtable); break; case PUBLICATIONOBJ_TABLES_IN_SCHEMA: schemaid = get_namespace_oid(pubobj->name, false); /* Filter out duplicates if user specifies "sch1, sch1" */ + *tables_schemas = list_append_unique_oid(*tables_schemas, schemaid); + *schemas = list_append_unique_oid(*schemas, schemaid); + break; + case PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA: + schemaid = get_namespace_oid(pubobj->name, false); + + /* Filter out duplicates if user specifies "sch1, sch1" */ + *sequences_schemas = list_append_unique_oid(*sequences_schemas, schemaid); *schemas = list_append_unique_oid(*schemas, schemaid); break; case PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA: @@ -186,6 +210,21 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate, list_free(search_path); /* Filter out duplicates if user specifies "sch1, sch1" */ + *tables_schemas = list_append_unique_oid(*tables_schemas, schemaid); + *schemas = list_append_unique_oid(*schemas, schemaid); + break; + case PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA: + search_path = fetch_search_path(false); + if (search_path == NIL) /* nothing valid in search_path? */ + ereport(ERROR, + errcode(ERRCODE_UNDEFINED_SCHEMA), + errmsg("no schema has been selected for CURRENT_SCHEMA")); + + schemaid = linitial_oid(search_path); + list_free(search_path); + + /* Filter out duplicates if user specifies "sch1, sch1" */ + *sequences_schemas = list_append_unique_oid(*sequences_schemas, schemaid); *schemas = list_append_unique_oid(*schemas, schemaid); break; default: @@ -251,7 +290,10 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) bool publish_via_partition_root_given; bool publish_via_partition_root; AclResult aclresult; - List *relations = NIL; + List *tables = NIL; + List *sequences = NIL; + List *tables_schemaidlist = NIL; + List *sequences_schemaidlist = NIL; List *schemaidlist = NIL; /* must have CREATE privilege on database */ @@ -306,6 +348,8 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) BoolGetDatum(pubactions.pubdelete); values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(pubactions.pubtruncate); + values[Anum_pg_publication_pubsequence - 1] = + BoolGetDatum(pubactions.pubsequence); values[Anum_pg_publication_pubviaroot - 1] = BoolGetDatum(publish_via_partition_root); @@ -330,26 +374,40 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) } else { - ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations, + ObjectsInPublicationToOids(stmt->pubobjects, pstate, + &tables, &sequences, + &tables_schemaidlist, + &sequences_schemaidlist, &schemaidlist); /* FOR ALL TABLES IN SCHEMA requires superuser */ - if (list_length(schemaidlist) > 0 && !superuser()) + if (list_length(tables_schemaidlist) > 0 && !superuser()) ereport(ERROR, errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("must be superuser to create FOR ALL TABLES IN SCHEMA publication")); - if (list_length(relations) > 0) + if (list_length(tables) > 0) { List *rels; - rels = OpenTableList(relations); - CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist, + rels = OpenTableList(tables); + CheckObjSchemaNotAlreadyInPublication(rels, tables_schemaidlist, PUBLICATIONOBJ_TABLE); PublicationAddTables(puboid, rels, true, NULL); CloseTableList(rels); } + if (list_length(sequences) > 0) + { + List *rels; + + rels = OpenSequenceList(sequences); + CheckObjSchemaNotAlreadyInPublication(rels, sequences_schemaidlist, + PUBLICATIONOBJ_SEQUENCE); + PublicationAddTables(puboid, rels, true, NULL); + CloseSequenceList(rels); + } + if (list_length(schemaidlist) > 0) { /* @@ -653,12 +711,13 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt, */ static void CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, - List *tables, List *schemaidlist) + List *tables, List *tables_schemaidlist, + List *sequences, List *sequences_schemaidlist) { Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup); if ((stmt->action == AP_AddObjects || stmt->action == AP_SetObjects) && - schemaidlist && !superuser()) + (tables_schemaidlist || sequences_schemaidlist) && !superuser()) ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("must be superuser to add or set schemas"))); @@ -667,13 +726,24 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, * Check that user is allowed to manipulate the publication tables in * schema */ - if (schemaidlist && pubform->puballtables) + if (tables_schemaidlist && pubform->puballtables) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("publication \"%s\" is defined as FOR ALL TABLES", NameStr(pubform->pubname)), errdetail("Tables from schema cannot be added to, dropped from, or set on FOR ALL TABLES publications."))); + /* + * Check that user is allowed to manipulate the publication sequences in + * schema + */ + if (sequences_schemaidlist && pubform->puballsequences) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("publication \"%s\" is defined as FOR ALL SEQUENCES", + NameStr(pubform->pubname)), + errdetail("Sequences from schema cannot be added to, dropped from, or set on FOR ALL SEQUENCES publications."))); + /* Check that user is allowed to manipulate the publication tables. */ if (tables && pubform->puballtables) ereport(ERROR, @@ -681,6 +751,108 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, errmsg("publication \"%s\" is defined as FOR ALL TABLES", NameStr(pubform->pubname)), errdetail("Tables cannot be added to or dropped from FOR ALL TABLES publications."))); + + /* Check that user is allowed to manipulate the publication tables. */ + if (sequences && pubform->puballsequences) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("publication \"%s\" is defined as FOR ALL SEQUENCES", + NameStr(pubform->pubname)), + errdetail("Sequences cannot be added to or dropped from FOR ALL SEQUENCES publications."))); +} + +/* + * Add or remove sequence to/from publication. + */ +static void +AlterPublicationSequences(AlterPublicationStmt *stmt, HeapTuple tup, + List *sequences, List *schemaidlist) +{ + List *rels = NIL; + Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup); + Oid pubid = pubform->oid; + + /* + * It is quite possible that for the SET case user has not specified any + * tables in which case we need to remove all the existing tables. + */ + if (!sequences && stmt->action != AP_SetObjects) + return; + + rels = OpenSequenceList(sequences); + + if (stmt->action == AP_AddObjects) + { + List *schemas = NIL; + + /* + * Check if the relation is member of the existing schema in the + * publication or member of the schema list specified. + */ + schemas = list_concat_copy(schemaidlist, GetPublicationSchemas(pubid)); + CheckObjSchemaNotAlreadyInPublication(rels, schemas, + PUBLICATIONOBJ_SEQUENCE); + PublicationAddSequences(pubid, rels, false, stmt); + } + else if (stmt->action == AP_DropObjects) + PublicationDropSequences(pubid, rels, false); + else /* DEFELEM_SET */ + { + List *oldrelids = GetPublicationRelations(pubid, + PUBLICATION_PART_ROOT); + List *delrels = NIL; + ListCell *oldlc; + + CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist, + PUBLICATIONOBJ_SEQUENCE); + + /* Calculate which relations to drop. */ + foreach(oldlc, oldrelids) + { + Oid oldrelid = lfirst_oid(oldlc); + ListCell *newlc; + bool found = false; + + foreach(newlc, rels) + { + PublicationRelInfo *newpubrel; + + newpubrel = (PublicationRelInfo *) lfirst(newlc); + if (RelationGetRelid(newpubrel->relation) == oldrelid) + { + found = true; + break; + } + } + /* Not yet in the list, open it and add to the list */ + if (!found) + { + Relation oldrel; + PublicationRelInfo *pubrel; + + /* Wrap relation into PublicationRelInfo */ + oldrel = table_open(oldrelid, ShareUpdateExclusiveLock); + + pubrel = palloc(sizeof(PublicationRelInfo)); + pubrel->relation = oldrel; + + delrels = lappend(delrels, pubrel); + } + } + + /* And drop them. */ + PublicationDropSequences(pubid, delrels, true); + + /* + * Don't bother calculating the difference for adding, we'll catch and + * skip existing ones when doing catalog update. + */ + PublicationAddSequences(pubid, rels, true, stmt); + + CloseSequenceList(delrels); + } + + CloseSequenceList(rels); } /* @@ -718,13 +890,21 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt) AlterPublicationOptions(pstate, stmt, rel, tup); else { - List *relations = NIL; + List *tables = NIL; + List *sequences = NIL; + List *tables_schemaidlist = NIL; + List *sequences_schemaidlist = NIL; List *schemaidlist = NIL; - ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations, + ObjectsInPublicationToOids(stmt->pubobjects, pstate, + &tables, &sequences, + &tables_schemaidlist, + &sequences_schemaidlist, &schemaidlist); - CheckAlterPublication(stmt, tup, relations, schemaidlist); + CheckAlterPublication(stmt, tup, + tables, tables_schemaidlist, + sequences, sequences_schemaidlist); /* * Lock the publication so nobody else can do anything with it. This @@ -749,7 +929,9 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt) errmsg("publication \"%s\" does not exist", stmt->pubname)); - AlterPublicationTables(stmt, tup, relations, schemaidlist); + AlterPublicationTables(stmt, tup, tables, tables_schemaidlist); + AlterPublicationSequences(stmt, tup, sequences, sequences_schemaidlist); + AlterPublicationSchemas(stmt, tup, schemaidlist); } @@ -1157,6 +1339,144 @@ PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok) } } +/* + * Open relations specified by a PublicationTable list. + * In the returned list of PublicationRelInfo, tables are locked + * in ShareUpdateExclusiveLock mode in order to add them to a publication. + */ +static List * +OpenSequenceList(List *sequences) +{ + List *relids = NIL; + List *rels = NIL; + ListCell *lc; + + /* + * Open, share-lock, and check all the explicitly-specified relations + */ + foreach(lc, sequences) + { + PublicationTable *s = lfirst_node(PublicationTable, lc); + Relation rel; + Oid myrelid; + PublicationRelInfo *pub_rel; + + /* Allow query cancel in case this takes a long time */ + CHECK_FOR_INTERRUPTS(); + + rel = table_openrv(s->relation, ShareUpdateExclusiveLock); + myrelid = RelationGetRelid(rel); + + /* + * Filter out duplicates if user specifies "foo, foo". + * + * Note that this algorithm is known to not be very efficient (O(N^2)) + * but given that it only works on list of tables given to us by user + * it's deemed acceptable. + */ + if (list_member_oid(relids, myrelid)) + { + table_close(rel, ShareUpdateExclusiveLock); + continue; + } + + pub_rel = palloc(sizeof(PublicationRelInfo)); + pub_rel->relation = rel; + rels = lappend(rels, pub_rel); + relids = lappend_oid(relids, myrelid); + } + + list_free(relids); + + return rels; +} + +/* + * Close all relations in the list. + */ +static void +CloseSequenceList(List *rels) +{ + ListCell *lc; + + foreach(lc, rels) + { + PublicationRelInfo *pub_rel; + + pub_rel = (PublicationRelInfo *) lfirst(lc); + table_close(pub_rel->relation, NoLock); + } +} + +/* + * Add listed tables to the publication. + */ +static void +PublicationAddSequences(Oid pubid, List *rels, bool if_not_exists, + AlterPublicationStmt *stmt) +{ + ListCell *lc; + + Assert(!stmt || !stmt->for_all_sequences); + + foreach(lc, rels) + { + PublicationRelInfo *pub_rel = (PublicationRelInfo *) lfirst(lc); + Relation rel = pub_rel->relation; + ObjectAddress obj; + + /* Must be owner of the sequence or superuser. */ + if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind), + RelationGetRelationName(rel)); + + obj = publication_add_relation(pubid, pub_rel, if_not_exists); + if (stmt) + { + EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress, + (Node *) stmt); + + InvokeObjectPostCreateHook(PublicationRelRelationId, + obj.objectId, 0); + } + } +} + +/* + * Remove listed sequences from the publication. + */ +static void +PublicationDropSequences(Oid pubid, List *rels, bool missing_ok) +{ + ObjectAddress obj; + ListCell *lc; + Oid prid; + + foreach(lc, rels) + { + Relation rel = (Relation) lfirst(lc); + Oid relid = RelationGetRelid(rel); + + prid = GetSysCacheOid2(PUBLICATIONRELMAP, Anum_pg_publication_rel_oid, + ObjectIdGetDatum(relid), + ObjectIdGetDatum(pubid)); + if (!OidIsValid(prid)) + { + if (missing_ok) + continue; + + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("relation \"%s\" is not part of the publication", + RelationGetRelationName(rel)))); + } + + ObjectAddressSet(obj, PublicationRelRelationId, prid); + performDeletion(&obj, DROP_CASCADE, 0); + } +} + + /* * Internal workhorse for changing a publication owner */ diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c index ab592ce2f15..fe4f21ec438 100644 --- a/src/backend/commands/sequence.c +++ b/src/backend/commands/sequence.c @@ -336,6 +336,85 @@ ResetSequence(Oid seq_relid) relation_close(seq_rel, NoLock); } +/* + * Reset a sequence to its initial value. + * + * The change is made transactionally, so that on failure of the current + * transaction, the sequence will be restored to its previous state. + * We do that by creating a whole new relfilenode for the sequence; so this + * works much like the rewriting forms of ALTER TABLE. + * + * Caller is assumed to have acquired AccessExclusiveLock on the sequence, + * which must not be released until end of transaction. Caller is also + * responsible for permissions checking. + */ +void +ResetSequence2(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called) +{ + Relation seq_rel; + SeqTable elm; + Form_pg_sequence_data seq; + Buffer buf; + HeapTupleData seqdatatuple; + HeapTuple tuple; + + /* + * Read the old sequence. This does a bit more work than really + * necessary, but it's simple, and we do want to double-check that it's + * indeed a sequence. + */ + init_sequence(seq_relid, &elm, &seq_rel); + (void) read_seq_tuple(seq_rel, &buf, &seqdatatuple); + + /* + * Copy the existing sequence tuple. + */ + tuple = heap_copytuple(&seqdatatuple); + + /* Now we're done with the old page */ + UnlockReleaseBuffer(buf); + + /* + * Modify the copied tuple to execute the restart (compare the RESTART + * action in AlterSequence) + */ + seq = (Form_pg_sequence_data) GETSTRUCT(tuple); + seq->last_value = last_value; + seq->is_called = is_called; + seq->log_cnt = log_cnt; + + /* + * Create a new storage file for the sequence. + */ + RelationSetNewRelfilenode(seq_rel, seq_rel->rd_rel->relpersistence); + + /* + * Ensure sequence's relfrozenxid is at 0, since it won't contain any + * unfrozen XIDs. Same with relminmxid, since a sequence will never + * contain multixacts. + */ + Assert(seq_rel->rd_rel->relfrozenxid == InvalidTransactionId); + Assert(seq_rel->rd_rel->relminmxid == InvalidMultiXactId); + + /* + * Insert the modified tuple into the new storage file. + * + * XXX Maybe this should also use created=true, just like the other places + * calling fill_seq_with_data. That's probably needed for correct cascading + * replication. + * + * XXX That'd mean all fill_seq_with_data callers use created=true, making + * the parameter unnecessary. + */ + fill_seq_with_data(seq_rel, tuple); + + /* Clear local cache so that we don't think we have cached numbers */ + /* Note that we do not change the currval() state */ + elm->cached = elm->last; + + relation_close(seq_rel, NoLock); +} + /* * Initialize a sequence's relation with the specified tuple as content */ diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 3ef6607d246..5beb67e7652 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -85,6 +85,7 @@ typedef struct SubOpts } SubOpts; static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); +static List *fetch_sequence_list(WalReceiverConn *wrconn, List *publications); static void check_duplicates_in_publist(List *publist, Datum *datums); static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname); static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); @@ -496,6 +497,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, char *err; WalReceiverConn *wrconn; List *tables; + List *sequences; ListCell *lc; char table_state; @@ -534,6 +536,26 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, InvalidXLogRecPtr); } + /* + * Get the sequence list from publisher and build local sequence + * status info. + */ + sequences = fetch_sequence_list(wrconn, publications); + foreach(lc, sequences) + { + RangeVar *rv = (RangeVar *) lfirst(lc); + Oid relid; + + relid = RangeVarGetRelid(rv, AccessShareLock, false); + + /* Check for supported relkind. */ + CheckSubscriptionRelkind(get_rel_relkind(relid), + rv->schemaname, rv->relname); + + AddSubscriptionRelState(subid, relid, table_state, + InvalidXLogRecPtr); + } + /* * If requested, create permanent slot for the subscription. We * won't use the initial snapshot for anything, so no need to @@ -706,6 +728,10 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) { Oid relid = subrel_local_oids[off]; + /* XXX ignore sequences - maybe do this in GetSubscriptionRelations? */ + if (get_rel_relkind(relid) == RELKIND_SEQUENCE) + continue; + if (!bsearch(&relid, pubrel_local_oids, list_length(pubrel_names), sizeof(Oid), oid_cmp)) { @@ -797,6 +823,183 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); } } + + /* + * XXX now do the same thing for sequences, maybe before the preceding + * block, or earlier? + */ + + /* Get the table list from publisher. */ + pubrel_names = fetch_sequence_list(wrconn, sub->publications); + + /* Get local table list. */ + subrel_states = GetSubscriptionRelations(sub->oid); + + /* + * Build qsorted array of local table oids for faster lookup. This can + * potentially contain all tables in the database so speed of lookup + * is important. + */ + subrel_local_oids = palloc(list_length(subrel_states) * sizeof(Oid)); + off = 0; + foreach(lc, subrel_states) + { + SubscriptionRelState *relstate = (SubscriptionRelState *) lfirst(lc); + + subrel_local_oids[off++] = relstate->relid; + } + qsort(subrel_local_oids, list_length(subrel_states), + sizeof(Oid), oid_cmp); + + /* + * Rels that we want to remove from subscription and drop any slots + * and origins corresponding to them. + */ + sub_remove_rels = palloc(list_length(subrel_states) * sizeof(SubRemoveRels)); + + /* + * Walk over the remote tables and try to match them to locally known + * tables. If the table is not known locally create a new state for + * it. + * + * Also builds array of local oids of remote tables for the next step. + */ + off = 0; + pubrel_local_oids = palloc(list_length(pubrel_names) * sizeof(Oid)); + + foreach(lc, pubrel_names) + { + RangeVar *rv = (RangeVar *) lfirst(lc); + Oid relid; + + relid = RangeVarGetRelid(rv, AccessShareLock, false); + + /* Check for supported relkind. */ + CheckSubscriptionRelkind(get_rel_relkind(relid), + rv->schemaname, rv->relname); + + pubrel_local_oids[off++] = relid; + + if (!bsearch(&relid, subrel_local_oids, + list_length(subrel_states), sizeof(Oid), oid_cmp)) + { + AddSubscriptionRelState(sub->oid, relid, + copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY, + InvalidXLogRecPtr); + ereport(DEBUG1, + (errmsg_internal("table \"%s.%s\" added to subscription \"%s\"", + rv->schemaname, rv->relname, sub->name))); + } + } + + /* + * Next remove state for tables we should not care about anymore using + * the data we collected above + */ + qsort(pubrel_local_oids, list_length(pubrel_names), + sizeof(Oid), oid_cmp); + + remove_rel_len = 0; + for (off = 0; off < list_length(subrel_states); off++) + { + Oid relid = subrel_local_oids[off]; + + /* XXX ignore non-sequences - maybe do this in GetSubscriptionRelations? */ + if (get_rel_relkind(relid) != RELKIND_SEQUENCE) + continue; + + if (!bsearch(&relid, pubrel_local_oids, + list_length(pubrel_names), sizeof(Oid), oid_cmp)) + { + char state; + XLogRecPtr statelsn; + + /* + * Lock pg_subscription_rel with AccessExclusiveLock to + * prevent any race conditions with the apply worker + * re-launching workers at the same time this code is trying + * to remove those tables. + * + * Even if new worker for this particular rel is restarted it + * won't be able to make any progress as we hold exclusive + * lock on subscription_rel till the transaction end. It will + * simply exit as there is no corresponding rel entry. + * + * This locking also ensures that the state of rels won't + * change till we are done with this refresh operation. + */ + if (!rel) + rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock); + + /* Last known rel state. */ + state = GetSubscriptionRelState(sub->oid, relid, &statelsn); + + sub_remove_rels[remove_rel_len].relid = relid; + sub_remove_rels[remove_rel_len++].state = state; + + RemoveSubscriptionRel(sub->oid, relid); + + logicalrep_worker_stop(sub->oid, relid); + + /* + * For READY state, we would have already dropped the + * tablesync origin. + */ + if (state != SUBREL_STATE_READY) + { + char originname[NAMEDATALEN]; + + /* + * Drop the tablesync's origin tracking if exists. + * + * It is possible that the origin is not yet created for + * tablesync worker, this can happen for the states before + * SUBREL_STATE_FINISHEDCOPY. The apply worker can also + * concurrently try to drop the origin and by this time + * the origin might be already removed. For these reasons, + * passing missing_ok = true. + */ + ReplicationOriginNameForTablesync(sub->oid, relid, originname, + sizeof(originname)); + replorigin_drop_by_name(originname, true, false); + } + + ereport(DEBUG1, + (errmsg_internal("table \"%s.%s\" removed from subscription \"%s\"", + get_namespace_name(get_rel_namespace(relid)), + get_rel_name(relid), + sub->name))); + } + } + + /* + * Drop the tablesync slots associated with removed tables. This has + * to be at the end because otherwise if there is an error while doing + * the database operations we won't be able to rollback dropped slots. + */ + for (off = 0; off < remove_rel_len; off++) + { + if (sub_remove_rels[off].state != SUBREL_STATE_READY && + sub_remove_rels[off].state != SUBREL_STATE_SYNCDONE) + { + char syncslotname[NAMEDATALEN] = {0}; + + /* + * For READY/SYNCDONE states we know the tablesync slot has + * already been dropped by the tablesync worker. + * + * For other states, there is no certainty, maybe the slot + * does not exist yet. Also, if we fail after removing some of + * the slots, next time, it will again try to drop already + * dropped slots and fail. For these reasons, we allow + * missing_ok = true for the drop. + */ + ReplicationSlotNameForTablesync(sub->oid, sub_remove_rels[off].relid, + syncslotname, sizeof(syncslotname)); + ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); + } + } + } PG_FINALLY(); { @@ -1616,6 +1819,75 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) return tablelist; } +/* + * Get the list of sequences which belong to specified publications on the + * publisher connection. + */ +static List * +fetch_sequence_list(WalReceiverConn *wrconn, List *publications) +{ + WalRcvExecResult *res; + StringInfoData cmd; + TupleTableSlot *slot; + Oid tableRow[2] = {TEXTOID, TEXTOID}; + ListCell *lc; + bool first; + List *tablelist = NIL; + + Assert(list_length(publications) > 0); + + initStringInfo(&cmd); + appendStringInfoString(&cmd, "SELECT DISTINCT s.schemaname, s.sequencename\n" + " FROM pg_catalog.pg_publication_sequences s\n" + " WHERE s.pubname IN ("); + first = true; + foreach(lc, publications) + { + char *pubname = strVal(lfirst(lc)); + + if (first) + first = false; + else + appendStringInfoString(&cmd, ", "); + + appendStringInfoString(&cmd, quote_literal_cstr(pubname)); + } + appendStringInfoChar(&cmd, ')'); + + res = walrcv_exec(wrconn, cmd.data, 2, tableRow); + pfree(cmd.data); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errmsg("could not receive list of replicated tables from the publisher: %s", + res->err))); + + /* Process tables. */ + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + { + char *nspname; + char *relname; + bool isnull; + RangeVar *rv; + + nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); + Assert(!isnull); + + rv = makeRangeVar(nspname, relname, -1); + tablelist = lappend(tablelist, rv); + + ExecClearTuple(slot); + } + ExecDropSingleTupleTableSlot(slot); + + walrcv_clear_result(res); + + return tablelist; +} + /* * This is to report the connection failure while dropping replication slots. * Here, we report the WARNING for all tablesync slots so that user can drop diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index 313c87398b2..78f14119d98 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -608,7 +608,7 @@ void CheckSubscriptionRelkind(char relkind, const char *nspname, const char *relname) { - if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE) + if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE && relkind != RELKIND_SEQUENCE) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("cannot use relation \"%s.%s\" as logical replication target", diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 6bd95bbce24..8b7e9710401 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -4852,6 +4852,7 @@ _copyCreatePublicationStmt(const CreatePublicationStmt *from) COPY_NODE_FIELD(options); COPY_NODE_FIELD(pubobjects); COPY_SCALAR_FIELD(for_all_tables); + COPY_SCALAR_FIELD(for_all_sequences); return newnode; } diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c index 4126516222b..55a7dbbddf3 100644 --- a/src/backend/nodes/equalfuncs.c +++ b/src/backend/nodes/equalfuncs.c @@ -2337,6 +2337,7 @@ _equalAlterPublicationStmt(const AlterPublicationStmt *a, COMPARE_NODE_FIELD(options); COMPARE_NODE_FIELD(pubobjects); COMPARE_SCALAR_FIELD(for_all_tables); + COMPARE_SCALAR_FIELD(for_all_sequences); COMPARE_SCALAR_FIELD(action); return true; diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index c4f32425060..bbde765cf56 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -9771,6 +9771,26 @@ PublicationObjSpec: $$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA; $$->location = @5; } + | SEQUENCE relation_expr + { + $$ = makeNode(PublicationObjSpec); + $$->pubobjtype = PUBLICATIONOBJ_SEQUENCE; + $$->pubtable = makeNode(PublicationTable); + $$->pubtable->relation = $2; + } + | ALL SEQUENCES IN_P SCHEMA ColId + { + $$ = makeNode(PublicationObjSpec); + $$->pubobjtype = PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA; + $$->name = $5; + $$->location = @5; + } + | ALL SEQUENCES IN_P SCHEMA CURRENT_SCHEMA + { + $$ = makeNode(PublicationObjSpec); + $$->pubobjtype = PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA; + $$->location = @5; + } | ColId { $$ = makeNode(PublicationObjSpec); @@ -10106,6 +10126,12 @@ UnlistenStmt: } ; +/* + * FIXME + * + * opt_publication_for_sequences and publication_for_sequences should be + * copies for sequences + */ /***************************************************************************** * @@ -10114,6 +10140,12 @@ UnlistenStmt: * BEGIN / COMMIT / ROLLBACK * (also older versions END / ABORT) * + * ALTER PUBLICATION name ADD SEQUENCE sequence [, sequence2] + * + * ALTER PUBLICATION name DROP SEQUENCE sequence [, sequence2] + * + * ALTER PUBLICATION name SET SEQUENCE sequence [, sequence2] + * *****************************************************************************/ TransactionStmt: diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 953942692ce..e8ead1387ae 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -647,6 +647,56 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, pq_sendbytes(out, message, sz); } +/* + * Write SEQUENCE to stream + */ +void +logicalrep_write_sequence(StringInfo out, Relation rel, TransactionId xid, + XLogRecPtr lsn, bool transactional, + int64 last_value, int64 log_cnt, bool is_called) +{ + uint8 flags = 0; + char *relname; + + pq_sendbyte(out, LOGICAL_REP_MSG_SEQUENCE); + + /* transaction ID (if not valid, we're not streaming) */ + if (TransactionIdIsValid(xid)) + pq_sendint32(out, xid); + + pq_sendint8(out, flags); + pq_sendint64(out, lsn); + + logicalrep_write_namespace(out, RelationGetNamespace(rel)); + relname = RelationGetRelationName(rel); + pq_sendstring(out, relname); + + pq_sendint8(out, transactional); + pq_sendint64(out, last_value); + pq_sendint64(out, log_cnt); + pq_sendint8(out, is_called); +} + +/* + * Read SEQUENCE from the stream. + */ +void +logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata) +{ + /* XXX skipping flags and lsn */ + pq_getmsgint(in, 1); + pq_getmsgint64(in); + + /* Read relation name from stream */ + seqdata->nspname = pstrdup(logicalrep_read_namespace(in)); + seqdata->seqname = pstrdup(pq_getmsgstring(in)); + + seqdata->transactional = pq_getmsgint(in, 1); + seqdata->last_value = pq_getmsgint64(in); + seqdata->log_cnt = pq_getmsgint64(in); + seqdata->is_called = pq_getmsgint(in, 1); +} + /* * Write relation description to the output stream. */ @@ -1203,6 +1253,8 @@ logicalrep_message_type(LogicalRepMsgType action) return "STREAM ABORT"; case LOGICAL_REP_MSG_STREAM_PREPARE: return "STREAM PREPARE"; + case LOGICAL_REP_MSG_SEQUENCE: + return "SEQUENCE"; } elog(ERROR, "invalid logical replication message type \"%c\"", action); diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index e596b69d466..c85867ee46b 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -100,6 +100,7 @@ #include "catalog/pg_subscription_rel.h" #include "catalog/pg_type.h" #include "commands/copy.h" +#include "commands/sequence.h" #include "miscadmin.h" #include "parser/parse_relation.h" #include "pgstat.h" @@ -359,6 +360,12 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn) * * If the synchronization position is reached (SYNCDONE), then the table can * be marked as READY and is no longer tracked. + * + * XXX This needs to handle sequences too - after AlterSubscription_refresh + * starts caring about sequences, GetSubscriptionNotReadyRelations won't + * return just tables, and we'll have to sync them here. Not sure it's worth + * creating a new "sync" worker per sequence, maybe we should just sync them + * in the current process (it's pretty light-weight). */ static void process_syncing_tables_for_apply(XLogRecPtr current_lsn) @@ -873,6 +880,99 @@ copy_table(Relation rel) logicalrep_rel_close(relmapentry, NoLock); } + + +/* + * FIXME add comment + */ +static void +fetch_sequence_data(char *nspname, char *relname, + int64 *last_value, int64 *log_cnt, bool *is_called) +{ + WalRcvExecResult *res; + StringInfoData cmd; + TupleTableSlot *slot; + Oid tableRow[3] = {INT8OID, INT8OID, BOOLOID}; + + initStringInfo(&cmd); + appendStringInfo(&cmd, "SELECT last_value, log_cnt, is_called\n" + " FROM %s", quote_qualified_identifier(nspname, relname)); + + res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 3, tableRow); + pfree(cmd.data); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errmsg("could not receive list of replicated tables from the publisher: %s", + res->err))); + + /* Process the sequence. */ + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + { + bool isnull; + + *last_value = DatumGetInt64(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + + *log_cnt = DatumGetInt64(slot_getattr(slot, 2, &isnull)); + Assert(!isnull); + + *is_called = DatumGetBool(slot_getattr(slot, 3, &isnull)); + Assert(!isnull); + + ExecClearTuple(slot); + } + ExecDropSingleTupleTableSlot(slot); + + walrcv_clear_result(res); +} + +/* + * Copy existing data of a sequence from publisher. + * + * Caller is responsible for locking the local relation. + */ +static void +copy_sequence(Relation rel) +{ + LogicalRepRelMapEntry *relmapentry; + LogicalRepRelation lrel; + StringInfoData cmd; + int64 last_value = 0, + log_cnt = 0; + bool is_called = 0; + + /* Get the publisher relation info. */ + fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)), + RelationGetRelationName(rel), &lrel); + + /* Put the relation into relmap. */ + logicalrep_relmap_update(&lrel); + + /* Map the publisher relation to local one. */ + relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock); + Assert(rel == relmapentry->localrel); + + /* Start copy on the publisher. */ + initStringInfo(&cmd); + + Assert(lrel.relkind == RELKIND_SEQUENCE); + + fetch_sequence_data(lrel.nspname, lrel.relname, &last_value, &log_cnt, &is_called); + + elog(WARNING, "sequence %s info last_value %ld log_cnt %ld is_called %d", + quote_qualified_identifier(lrel.nspname, lrel.relname), + last_value, log_cnt, is_called); + + ResetSequence2(RelationGetRelid(rel), last_value, log_cnt, is_called); + + logicalrep_rel_close(relmapentry, NoLock); +} + + + + /* * Determine the tablesync slot name. * @@ -1134,10 +1234,20 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) originname))); } - /* Now do the initial data copy */ - PushActiveSnapshot(GetTransactionSnapshot()); - copy_table(rel); - PopActiveSnapshot(); + if (get_rel_relkind(RelationGetRelid(rel)) == RELKIND_SEQUENCE) + { + /* Now do the initial sequence copy */ + PushActiveSnapshot(GetTransactionSnapshot()); + copy_sequence(rel); + PopActiveSnapshot(); + } + else + { + /* Now do the initial data copy */ + PushActiveSnapshot(GetTransactionSnapshot()); + copy_table(rel); + PopActiveSnapshot(); + } res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL); if (res->status != WALRCV_OK_COMMAND) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index d77bb32bb9e..68708d3907f 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -144,6 +144,7 @@ #include "catalog/pg_tablespace.h" #include "commands/tablecmds.h" #include "commands/tablespace.h" +#include "commands/sequence.h" #include "commands/trigger.h" #include "executor/executor.h" #include "executor/execPartition.h" @@ -1093,6 +1094,61 @@ apply_handle_origin(StringInfo s) errmsg_internal("ORIGIN message sent out of order"))); } +/* + * Handle SEQUENCE message. + */ +static void +apply_handle_sequence(StringInfo s) +{ + LogicalRepSequence seq; + Oid relid; + + if (handle_streamed_transaction(LOGICAL_REP_MSG_SEQUENCE, s)) + return; + + logicalrep_read_sequence(s, &seq); + + /* + * Non-transactional sequence updates should not be part of a remote + * transaction. There should not be any running transaction. + */ + Assert((!seq.transactional) || in_remote_transaction); + Assert(!(!seq.transactional && in_remote_transaction)); + Assert(!(!seq.transactional && IsTransactionState())); + + /* + * Make sure we're in a transaction (needed by ResetSequence2). For + * non-transactional updates we're guaranteed to start a new one, + * and we'll commit it at the end. + */ + if (!IsTransactionState()) + { + StartTransactionCommand(); + maybe_reread_subscription(); + } + + relid = RangeVarGetRelid(makeRangeVar(seq.nspname, + seq.seqname, -1), + RowExclusiveLock, false); + + /* lock the sequence in AccessExclusiveLock, as expected by ResetSequence2 */ + elog(WARNING, "locking sequence %d in exclusive mode", relid); + LockRelationOid(relid, AccessExclusiveLock); + + elog(WARNING, "applying sequence %s.%s transactional %d last_value %ld log_cnt %ld is_called %d", + seq.nspname, seq.seqname, seq.transactional, seq.last_value, seq.log_cnt, seq.is_called); + + /* apply the sequence change */ + ResetSequence2(relid, seq.last_value, seq.log_cnt, seq.is_called); + + /* + * Commit the per-stream transaction (we only do this when not in + * remote transaction, i.e. for non-transactional sequence updates. + */ + if (!in_remote_transaction) + CommitTransactionCommand(); +} + /* * Handle STREAM START message. */ @@ -2421,6 +2477,10 @@ apply_dispatch(StringInfo s) */ break; + case LOGICAL_REP_MSG_SEQUENCE: + apply_handle_sequence(s); + return; + case LOGICAL_REP_MSG_STREAM_START: apply_handle_stream_start(s); break; diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 6df705f90ff..b82c6b10305 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -49,6 +49,10 @@ static void pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message); +static void pgoutput_sequence(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr sequence_lsn, + Relation rel, bool transactional, + int64 last_value, int64 log_cnt, bool is_called); static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, RepOriginId origin_id); static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, @@ -161,6 +165,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->change_cb = pgoutput_change; cb->truncate_cb = pgoutput_truncate; cb->message_cb = pgoutput_message; + cb->sequence_cb = pgoutput_sequence; cb->commit_cb = pgoutput_commit_txn; cb->begin_prepare_cb = pgoutput_begin_prepare_txn; @@ -177,6 +182,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->stream_commit_cb = pgoutput_stream_commit; cb->stream_change_cb = pgoutput_change; cb->stream_message_cb = pgoutput_message; + cb->stream_sequence_cb = pgoutput_sequence; cb->stream_truncate_cb = pgoutput_truncate; /* transaction streaming - two-phase commit */ cb->stream_prepare_cb = pgoutput_stream_prepare_txn; @@ -190,6 +196,7 @@ parse_output_parameters(List *options, PGOutputData *data) bool publication_names_given = false; bool binary_option_given = false; bool messages_option_given = false; + bool sequences_option_given = false; bool streaming_given = false; bool two_phase_option_given = false; @@ -197,6 +204,7 @@ parse_output_parameters(List *options, PGOutputData *data) data->streaming = false; data->messages = false; data->two_phase = false; + data->sequences = true; foreach(lc, options) { @@ -262,6 +270,16 @@ parse_output_parameters(List *options, PGOutputData *data) data->messages = defGetBoolean(defel); } + else if (strcmp(defel->defname, "sequences") == 0) + { + if (sequences_option_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + sequences_option_given = true; + + data->sequences = defGetBoolean(defel); + } else if (strcmp(defel->defname, "streaming") == 0) { if (streaming_given) @@ -858,6 +876,51 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, OutputPluginWrite(ctx, true); } +static void +pgoutput_sequence(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr sequence_lsn, + Relation rel, bool transactional, + int64 last_value, int64 log_cnt, bool is_called) +{ + PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; + TransactionId xid = InvalidTransactionId; + RelationSyncEntry *relentry; + + if (!data->sequences) + return; + + if (!is_publishable_relation(rel)) + return; + + /* + * Remember the xid for the message in streaming mode. See + * pgoutput_change. + */ + if (in_streaming) + xid = txn->xid; + + relentry = get_rel_sync_entry(data, RelationGetRelid(rel)); + + /* + * First check the sequence filter. + * + * We handle just REORDER_BUFFER_CHANGE_SEQUENCE here. + */ + if (!relentry->pubactions.pubsequence) + return; + + OutputPluginPrepareWrite(ctx, true); + logicalrep_write_sequence(ctx->out, + rel, + xid, + sequence_lsn, + transactional, + last_value, + log_cnt, + is_called); + OutputPluginWrite(ctx, true); +} + /* * Currently we always forward. */ @@ -1141,7 +1204,8 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->schema_sent = false; entry->streamed_txns = NIL; entry->pubactions.pubinsert = entry->pubactions.pubupdate = - entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false; + entry->pubactions.pubdelete = entry->pubactions.pubtruncate = + entry->pubactions.pubsequence = false; entry->publish_as_relid = InvalidOid; entry->map = NULL; /* will be set by maybe_send_schema() if * needed */ @@ -1163,6 +1227,7 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) Oid publish_as_relid = relid; bool am_partition = get_rel_relispartition(relid); char relkind = get_rel_relkind(relid); + bool is_sequence = (get_rel_relkind(relid) == RELKIND_SEQUENCE); /* Reload publications if needed before use. */ if (!publications_valid) @@ -1191,6 +1256,7 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->pubactions.pubupdate = false; entry->pubactions.pubdelete = false; entry->pubactions.pubtruncate = false; + entry->pubactions.pubsequence = false; if (entry->map) { /* @@ -1213,12 +1279,23 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) Publication *pub = lfirst(lc); bool publish = false; - if (pub->alltables) + if (pub->alltables && (!is_sequence)) { publish = true; if (pub->pubviaroot && am_partition) publish_as_relid = llast_oid(get_partition_ancestors(relid)); } + else if (pub->allsequences && is_sequence) + { + publish = true; + } + + /* if a sequence, just cross-check the list of publications */ + if (!publish && is_sequence) + { + if (list_member_oid(pubids, pub->oid)) + publish = true; + } if (!publish) { @@ -1275,10 +1352,12 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->pubactions.pubupdate |= pub->pubactions.pubupdate; entry->pubactions.pubdelete |= pub->pubactions.pubdelete; entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate; + entry->pubactions.pubsequence |= pub->pubactions.pubsequence; } if (entry->pubactions.pubinsert && entry->pubactions.pubupdate && - entry->pubactions.pubdelete && entry->pubactions.pubtruncate) + entry->pubactions.pubdelete && entry->pubactions.pubtruncate && + entry->pubactions.pubsequence) break; } diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 2707fed12f4..45a8b3e490a 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -5586,6 +5586,7 @@ GetRelationPublicationActions(Relation relation) pubactions->pubupdate |= pubform->pubupdate; pubactions->pubdelete |= pubform->pubdelete; pubactions->pubtruncate |= pubform->pubtruncate; + pubactions->pubsequence |= pubform->pubsequence; ReleaseSysCache(tup); @@ -5594,7 +5595,8 @@ GetRelationPublicationActions(Relation relation) * other publications. */ if (pubactions->pubinsert && pubactions->pubupdate && - pubactions->pubdelete && pubactions->pubtruncate) + pubactions->pubdelete && pubactions->pubtruncate && + pubactions->pubsequence) break; } diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index 98882272130..4900c48ff19 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1782,20 +1782,20 @@ psql_completion(const char *text, int start, int end) COMPLETE_WITH("ADD", "DROP", "OWNER TO", "RENAME TO", "SET"); /* ALTER PUBLICATION <name> ADD */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD")) - COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE"); - else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") || - (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") && + COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE|SEQUENCE"); + else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE|SEQUENCE") || + (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE|SEQUENCE") && ends_with(prev_wd, ','))) COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables); - else if (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE")) + else if (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE|SEQUENCE")) COMPLETE_WITH(","); /* ALTER PUBLICATION <name> DROP */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP")) - COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE"); + COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE|SEQUENCE"); /* ALTER PUBLICATION <name> SET */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET")) - COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "TABLE"); - else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES", "IN", "SCHEMA")) + COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "TABLE|SEQUENCE"); + else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA")) COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas " AND nspname NOT LIKE E'pg\\\\_%'", "CURRENT_SCHEMA"); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 62f36daa981..a66eb8109d0 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11546,6 +11546,11 @@ provolatile => 's', prorettype => 'oid', proargtypes => 'text', proallargtypes => '{text,oid}', proargmodes => '{i,o}', proargnames => '{pubname,relid}', prosrc => 'pg_get_publication_tables' }, +{ oid => '8000', descr => 'get OIDs of sequences in a publication', + proname => 'pg_get_publication_sequences', prorows => '1000', proretset => 't', + provolatile => 's', prorettype => 'oid', proargtypes => 'text', + proallargtypes => '{text,oid}', proargmodes => '{i,o}', + proargnames => '{pubname,relid}', prosrc => 'pg_get_publication_sequences' }, { oid => '6121', descr => 'returns whether a relation can be part of a publication', proname => 'pg_relation_is_publishable', provolatile => 's', diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 841b9b6c253..e56286772f4 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -40,6 +40,12 @@ CATALOG(pg_publication,6104,PublicationRelationId) */ bool puballtables; + /* + * indicates that this is special publication which should encompass all + * sequences in the database (except for the unlogged and temp ones) + */ + bool puballsequences; + /* true if inserts are published */ bool pubinsert; @@ -52,6 +58,9 @@ CATALOG(pg_publication,6104,PublicationRelationId) /* true if truncates are published */ bool pubtruncate; + /* true if sequences are published */ + bool pubsequence; + /* true if partition changes are published using root schema */ bool pubviaroot; } FormData_pg_publication; @@ -72,6 +81,7 @@ typedef struct PublicationActions bool pubupdate; bool pubdelete; bool pubtruncate; + bool pubsequence; } PublicationActions; typedef struct Publication @@ -79,6 +89,7 @@ typedef struct Publication Oid oid; char *name; bool alltables; + bool allsequences; bool pubviaroot; PublicationActions pubactions; } Publication; @@ -121,6 +132,9 @@ extern List *GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt, Oid relid); +extern List *GetAllSequencesPublicationRelations(void); +extern List *GetPublicationSequenceRelations(Oid pubid); + extern bool is_publishable_relation(Relation rel); extern bool is_schema_publication(Oid pubid); extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *targetrel, diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h index 9fecc41954e..d8c255a7af5 100644 --- a/src/include/commands/sequence.h +++ b/src/include/commands/sequence.h @@ -60,6 +60,7 @@ extern ObjectAddress DefineSequence(ParseState *pstate, CreateSeqStmt *stmt); extern ObjectAddress AlterSequence(ParseState *pstate, AlterSeqStmt *stmt); extern void DeleteSequenceTuple(Oid relid); extern void ResetSequence(Oid seq_relid); +extern void ResetSequence2(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called); extern void ResetSequenceCaches(void); extern void seq_redo(XLogReaderState *rptr); diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 37fcc4c9b5a..4a990364e4a 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3656,6 +3656,10 @@ typedef enum PublicationObjSpecType PUBLICATIONOBJ_TABLES_IN_SCHEMA, /* All tables in schema */ PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA, /* All tables in first element of * search_path */ + PUBLICATIONOBJ_SEQUENCE, /* Sequence type */ + PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA, /* Sequences in schema type */ + PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA, /* Get the first element of + * search_path */ PUBLICATIONOBJ_CONTINUATION /* Continuation of previous type */ } PublicationObjSpecType; @@ -3675,6 +3679,7 @@ typedef struct CreatePublicationStmt List *options; /* List of DefElem nodes */ List *pubobjects; /* Optional list of publication objects */ bool for_all_tables; /* Special publication for all tables in db */ + bool for_all_sequences; /* Special publication for all sequences in db */ } CreatePublicationStmt; typedef enum AlterPublicationAction @@ -3698,6 +3703,7 @@ typedef struct AlterPublicationStmt */ List *pubobjects; /* Optional list of publication objects */ bool for_all_tables; /* Special publication for all tables in db */ + bool for_all_sequences; /* Special publication for all sequences in db */ AlterPublicationAction action; /* What action to perform with the given * objects */ } AlterPublicationStmt; diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 22fffaca62d..8f8c325522d 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -60,6 +60,7 @@ typedef enum LogicalRepMsgType LOGICAL_REP_MSG_RELATION = 'R', LOGICAL_REP_MSG_TYPE = 'Y', LOGICAL_REP_MSG_MESSAGE = 'M', + LOGICAL_REP_MSG_SEQUENCE = 'X', /* FIXME change */ LOGICAL_REP_MSG_BEGIN_PREPARE = 'b', LOGICAL_REP_MSG_PREPARE = 'P', LOGICAL_REP_MSG_COMMIT_PREPARED = 'K', @@ -117,6 +118,18 @@ typedef struct LogicalRepTyp char *typname; /* name of the remote type */ } LogicalRepTyp; +/* Sequence info */ +typedef struct LogicalRepSequence +{ + Oid remoteid; /* unique id of the remote sequence */ + char *nspname; /* schema name of remote sequence */ + char *seqname; /* name of the remote sequence */ + bool transactional; + int64 last_value; + int64 log_cnt; + bool is_called; +} LogicalRepSequence; + /* Transaction info */ typedef struct LogicalRepBeginData { @@ -227,6 +240,12 @@ extern List *logicalrep_read_truncate(StringInfo in, bool *cascade, bool *restart_seqs); extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, const char *message); +extern void logicalrep_write_sequence(StringInfo out, Relation rel, + TransactionId xid, XLogRecPtr lsn, + bool transactional, + int64 last_value, int64 log_cnt, + bool is_called); +extern void logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata); extern void logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel); extern LogicalRepRelation *logicalrep_read_rel(StringInfo in); diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h index 78aa9151ef5..a6f6843ada6 100644 --- a/src/include/replication/pgoutput.h +++ b/src/include/replication/pgoutput.h @@ -28,6 +28,7 @@ typedef struct PGOutputData bool streaming; bool messages; bool two_phase; + bool sequences; } PGOutputData; #endif /* PGOUTPUT_H */ diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 1420288d67b..3f50e100f8d 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1429,6 +1429,14 @@ pg_prepared_xacts| SELECT p.transaction, FROM ((pg_prepared_xact() p(transaction, gid, prepared, ownerid, dbid) LEFT JOIN pg_authid u ON ((p.ownerid = u.oid))) LEFT JOIN pg_database d ON ((p.dbid = d.oid))); +pg_publication_sequences| SELECT p.pubname, + n.nspname AS schemaname, + c.relname AS sequencename + FROM pg_publication p, + LATERAL pg_get_publication_sequences((p.pubname)::text) gpt(relid), + (pg_class c + JOIN pg_namespace n ON ((n.oid = c.relnamespace))) + WHERE (c.oid = gpt.relid); pg_publication_tables| SELECT p.pubname, n.nspname AS schemaname, c.relname AS tablename diff --git a/src/test/subscription/t/027_sequences.pl b/src/test/subscription/t/027_sequences.pl new file mode 100644 index 00000000000..fd941619071 --- /dev/null +++ b/src/test/subscription/t/027_sequences.pl @@ -0,0 +1,196 @@ + +# Copyright (c) 2021, PostgreSQL Global Development Group + +# This tests that sequences are replicated correctly by logical replication +use strict; +use warnings; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More tests => 6; + +# Initialize publisher node +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +# Create subscriber node +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init(allows_streaming => 'logical'); +$node_subscriber->start; + +# Create some preexisting content on publisher +my $ddl = qq( + CREATE SEQUENCE s; +); + +# Setup structure on the publisher +$node_publisher->safe_psql('postgres', $ddl); + +# Create some the same structure on subscriber, and an extra sequence that +# we'll create on the publisher later +$ddl = qq( + CREATE SEQUENCE s; + CREATE SEQUENCE s2; +); + +$node_subscriber->safe_psql('postgres', $ddl); + +# Setup logical replication +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION seq_pub"); + +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION seq_pub ADD SEQUENCE s"); + +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION seq_sub CONNECTION '$publisher_connstr' PUBLICATION seq_pub WITH (slot_name = seq_sub_slot)" +); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Wait for initial sync to finish as well +my $synced_query = + "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');"; +$node_subscriber->poll_query_until('postgres', $synced_query) + or die "Timed out while waiting for subscriber to synchronize data"; + +# Insert initial test data +$node_publisher->safe_psql( + 'postgres', qq( + -- generate a number of values using the sequence + SELECT nextval('s') FROM generate_series(1,100); +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +my $result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s; +)); + +is( $result, '132|0|t', + 'check replicated sequence values on subscriber'); + + +# advance the sequence in a rolled-back transaction - should not be replicated +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + SELECT nextval('s') FROM generate_series(1,100); + ROLLBACK; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s; +)); + +is( $result, '231|0|t', + 'check replicated sequence values on subscriber'); + + +# create a new sequence and roll it back - should not be replicated, due to +# the transactional behavior +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + CREATE SEQUENCE s2; + ALTER PUBLICATION seq_pub ADD SEQUENCE s2; + SELECT nextval('s2') FROM generate_series(1,100); + ROLLBACK; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s2; +)); + +is( $result, '1|0|f', + 'check replicated sequence values on subscriber'); + + +# create a new sequence, advance it in a rolled-back transaction, but commit +# the create - the advance should be replicated nevertheless +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + CREATE SEQUENCE s2; + ALTER PUBLICATION seq_pub ADD SEQUENCE s2; + SAVEPOINT sp1; + SELECT nextval('s2') FROM generate_series(1,100); + ROLLBACK TO sp1; + COMMIT; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Wait for sync of the second sequence we just added to finish +$synced_query = + "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');"; +$node_subscriber->poll_query_until('postgres', $synced_query) + or die "Timed out while waiting for subscriber to synchronize data"; + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s2; +)); + +is( $result, '132|0|t', + 'check replicated sequence values on subscriber'); + + +# advance the new sequence in a transaction, and roll it back - in this case +# it should not be replicated at commit +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + SELECT nextval('s2') FROM generate_series(1,100); + ROLLBACK; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s2; +)); + +is( $result, '132|0|t', + 'check replicated sequence values on subscriber'); + + +# advance the sequence in a subtransaction - the subtransaction gets rolled +# back, but commit the main one - the changes should still be replicated +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + SAVEPOINT s1; + SELECT nextval('s2') FROM generate_series(1,100); + ROLLBACK TO s1; + COMMIT; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s2; +)); + +is( $result, '330|0|t', + 'check replicated sequence values on subscriber'); + + +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); -- 2.34.1 ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-02-12 00:34 Tomas Vondra <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 1 reply; 17+ messages in thread From: Tomas Vondra @ 2022-02-12 00:34 UTC (permalink / raw) To: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; +Cc: Amit Kapila <[email protected]>; PostgreSQL Hackers <[email protected]> On 2/10/22 19:17, Tomas Vondra wrote: > I've polished & pushed the first part adding sequence decoding > infrastructure etc. Attached are the two remaining parts. > > I plan to wait a day or two and then push the test_decoding part. The > last part (for built-in replication) will need more work and maybe > rethinking the grammar etc. > I've pushed the second part, adding sequences to test_decoding. Here's the remaining part, rebased, with a small tweak in the TAP test to eliminate the issue with not waiting for sequence increments. I've kept the tweak in a separate patch, so that we can throw it away easily if we happen to resolve the issue. regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company Attachments: [text/x-patch] 0001-Add-support-for-decoding-sequences-to-built-20220212.patch (77.7K, ../../[email protected]/2-0001-Add-support-for-decoding-sequences-to-built-20220212.patch) download | inline diff: From c456270469cdb6ad769455eb3d16aea6db2c02af Mon Sep 17 00:00:00 2001 From: Tomas Vondra <[email protected]> Date: Thu, 10 Feb 2022 15:18:59 +0100 Subject: [PATCH 1/2] Add support for decoding sequences to built-in replication --- doc/src/sgml/catalogs.sgml | 71 ++++ doc/src/sgml/ref/alter_publication.sgml | 24 +- doc/src/sgml/ref/alter_subscription.sgml | 4 +- src/backend/catalog/pg_publication.c | 149 ++++++++- src/backend/catalog/system_views.sql | 10 + src/backend/commands/publicationcmds.c | 350 +++++++++++++++++++- src/backend/commands/sequence.c | 79 +++++ src/backend/commands/subscriptioncmds.c | 272 +++++++++++++++ src/backend/executor/execReplication.c | 2 +- src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/equalfuncs.c | 1 + src/backend/parser/gram.y | 32 ++ src/backend/replication/logical/proto.c | 52 +++ src/backend/replication/logical/tablesync.c | 118 ++++++- src/backend/replication/logical/worker.c | 60 ++++ src/backend/replication/pgoutput/pgoutput.c | 85 ++++- src/backend/utils/cache/relcache.c | 4 +- src/bin/psql/tab-complete.c | 14 +- src/include/catalog/pg_proc.dat | 5 + src/include/catalog/pg_publication.h | 14 + src/include/commands/sequence.h | 1 + src/include/nodes/parsenodes.h | 6 + src/include/replication/logicalproto.h | 19 ++ src/include/replication/pgoutput.h | 1 + src/test/regress/expected/rules.out | 8 + src/test/subscription/t/028_sequences.pl | 196 +++++++++++ 26 files changed, 1542 insertions(+), 36 deletions(-) create mode 100644 src/test/subscription/t/028_sequences.pl diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 879d2dbce03..271dc03e5a2 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -9540,6 +9540,11 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l <entry>prepared transactions</entry> </row> + <row> + <entry><link linkend="view-pg-publication-sequences"><structname>pg_publication_sequences</structname></link></entry> + <entry>publications and their associated sequences</entry> + </row> + <row> <entry><link linkend="view-pg-publication-tables"><structname>pg_publication_tables</structname></link></entry> <entry>publications and their associated tables</entry> @@ -11375,6 +11380,72 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx </sect1> + <sect1 id="view-pg-publication-sequences"> + <title><structname>pg_publication_sequences</structname></title> + + <indexterm zone="view-pg-publication-sequences"> + <primary>pg_publication_sequences</primary> + </indexterm> + + <para> + The view <structname>pg_publication_sequences</structname> provides + information about the mapping between publications and the sequences they + contain. Unlike the underlying catalog + <link linkend="catalog-pg-publication-rel"><structname>pg_publication_rel</structname></link>, + this view expands + publications defined as <literal>FOR ALL SEQUENCES</literal>, so for such + publications there will be a row for each eligible sequence. + </para> + + <table> + <title><structname>pg_publication_sequences</structname> Columns</title> + <tgroup cols="1"> + <thead> + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + Column Type + </para> + <para> + Description + </para></entry> + </row> + </thead> + + <tbody> + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>pubname</structfield> <type>name</type> + (references <link linkend="catalog-pg-publication"><structname>pg_publication</structname></link>.<structfield>pubname</structfield>) + </para> + <para> + Name of publication + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>schemaname</structfield> <type>name</type> + (references <link linkend="catalog-pg-namespace"><structname>pg_namespace</structname></link>.<structfield>nspname</structfield>) + </para> + <para> + Name of schema containing sequence + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>sequencename</structfield> <type>name</type> + (references <link linkend="catalog-pg-class"><structname>pg_class</structname></link>.<structfield>relname</structfield>) + </para> + <para> + Name of sequence + </para></entry> + </row> + </tbody> + </tgroup> + </table> + </sect1> + <sect1 id="view-pg-publication-tables"> <title><structname>pg_publication_tables</structname></title> diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml index 7c7c27bf7ce..9da8274ae2c 100644 --- a/doc/src/sgml/ref/alter_publication.sgml +++ b/doc/src/sgml/ref/alter_publication.sgml @@ -31,7 +31,9 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r <phrase>where <replaceable class="parameter">publication_object</replaceable> is one of:</phrase> TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [, ... ] + SEQUENCE <replaceable class="parameter">sequence_name</replaceable> [ * ] [, ... ] ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ] + ALL SEQUENCES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ] </synopsis> </refsynopsisdiv> @@ -56,7 +58,18 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r </para> <para> - The fourth variant of this command listed in the synopsis can change + The next three variants change which sequences are part of the publication. + The <literal>SET SEQUENCE</literal> clause will replace the list of sequences + in the publication with the specified one. The <literal>ADD SEQUENCE</literal> + and <literal>DROP SEQUENCE</literal> clauses will add and remove one or more + sequences from the publication. Note that adding sequences to a publication + that is already subscribed to will require a <literal>ALTER SUBSCRIPTION + ... REFRESH PUBLICATION</literal> action on the subscribing side in order + to become effective. + </para> + + <para> + The seventh variant of this command listed in the synopsis can change all of the publication properties specified in <xref linkend="sql-createpublication"/>. Properties not mentioned in the command retain their previous settings. @@ -123,6 +136,15 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r </listitem> </varlistentry> + <varlistentry> + <term><replaceable class="parameter">sequence_name</replaceable></term> + <listitem> + <para> + Name of an existing sequence. + </para> + </listitem> + </varlistentry> + <varlistentry> <term><literal>SET ( <replaceable class="parameter">publication_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] )</literal></term> <listitem> diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 0b027cc3462..8f28cf03f40 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -147,7 +147,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < <listitem> <para> Fetch missing table information from publisher. This will start - replication of tables that were added to the subscribed-to publications + replication of tables and sequences that were added to the subscribed-to publications since <command>CREATE SUBSCRIPTION</command> or the last invocation of <command>REFRESH PUBLICATION</command>. </para> @@ -164,7 +164,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < Specifies whether to copy pre-existing data in the publications that are being subscribed to when the replication starts. The default is <literal>true</literal>. (Previously-subscribed - tables are not copied.) + tables and sequences are not copied.) </para> </listitem> </varlistentry> diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index e14ca2f5630..1a9e05ba98b 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -54,7 +54,8 @@ check_publication_add_relation(Relation targetrel) { /* Must be a regular or partitioned table */ if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION && - RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE) + RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE && + RelationGetForm(targetrel)->relkind != RELKIND_SEQUENCE) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("cannot add relation \"%s\" to publication", @@ -131,7 +132,8 @@ static bool is_publishable_class(Oid relid, Form_pg_class reltuple) { return (reltuple->relkind == RELKIND_RELATION || - reltuple->relkind == RELKIND_PARTITIONED_TABLE) && + reltuple->relkind == RELKIND_PARTITIONED_TABLE || + reltuple->relkind == RELKIND_SEQUENCE) && !IsCatalogRelationOid(relid) && reltuple->relpersistence == RELPERSISTENCE_PERMANENT && relid >= FirstNormalObjectId; @@ -503,6 +505,11 @@ GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) Form_pg_publication_rel pubrel; pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + + /* skip sequences here */ + if (get_rel_relkind(pubrel->prrelid) == RELKIND_SEQUENCE) + continue; + result = GetPubPartitionOptionRelations(result, pub_partopt, pubrel->prrelid); } @@ -517,6 +524,49 @@ GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) return result; } +/* + * Gets list of relation oids for a publication (sequences only). + * + * This should only be used for normal publications, the FOR ALL TABLES + * should use GetAllSequencesPublicationRelations(). + */ +List * +GetPublicationSequenceRelations(Oid pubid) +{ + List *result; + Relation pubrelsrel; + ScanKeyData scankey; + SysScanDesc scan; + HeapTuple tup; + + /* Find all publications associated with the relation. */ + pubrelsrel = table_open(PublicationRelRelationId, AccessShareLock); + + ScanKeyInit(&scankey, + Anum_pg_publication_rel_prpubid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(pubid)); + + scan = systable_beginscan(pubrelsrel, PublicationRelPrrelidPrpubidIndexId, + true, NULL, 1, &scankey); + + result = NIL; + while (HeapTupleIsValid(tup = systable_getnext(scan))) + { + Form_pg_publication_rel pubrel; + + pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + + if (get_rel_relkind(pubrel->prrelid) == RELKIND_SEQUENCE) + result = lappend_oid(result, pubrel->prrelid); + } + + systable_endscan(scan); + table_close(pubrelsrel, AccessShareLock); + + return result; +} + /* * Gets list of publication oids for publications marked as FOR ALL TABLES. */ @@ -762,6 +812,46 @@ GetAllSchemaPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) return result; } +/* + * Gets list of all relation published by FOR ALL TABLES publication(s). + * + * If the publication publishes partition changes via their respective root + * partitioned tables, we must exclude partitions in favor of including the + * root partitioned tables. + */ +List * +GetAllSequencesPublicationRelations(void) +{ + Relation classRel; + ScanKeyData key[1]; + TableScanDesc scan; + HeapTuple tuple; + List *result = NIL; + + classRel = table_open(RelationRelationId, AccessShareLock); + + ScanKeyInit(&key[0], + Anum_pg_class_relkind, + BTEqualStrategyNumber, F_CHAREQ, + CharGetDatum(RELKIND_SEQUENCE)); + + scan = table_beginscan_catalog(classRel, 1, key); + + while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL) + { + Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple); + Oid relid = relForm->oid; + + if (is_publishable_class(relid, relForm)) + result = lappend_oid(result, relid); + } + + table_endscan(scan); + + table_close(classRel, AccessShareLock); + return result; +} + /* * Get publication using oid * @@ -784,10 +874,12 @@ GetPublication(Oid pubid) pub->oid = pubid; pub->name = pstrdup(NameStr(pubform->pubname)); pub->alltables = pubform->puballtables; + pub->allsequences = pubform->puballsequences; pub->pubactions.pubinsert = pubform->pubinsert; pub->pubactions.pubupdate = pubform->pubupdate; pub->pubactions.pubdelete = pubform->pubdelete; pub->pubactions.pubtruncate = pubform->pubtruncate; + pub->pubactions.pubsequence = pubform->pubsequence; pub->pubviaroot = pubform->pubviaroot; ReleaseSysCache(tup); @@ -937,3 +1029,56 @@ pg_get_publication_tables(PG_FUNCTION_ARGS) SRF_RETURN_DONE(funcctx); } + +/* + * Returns Oids of sequences in a publication. + */ +Datum +pg_get_publication_sequences(PG_FUNCTION_ARGS) +{ + FuncCallContext *funcctx; + char *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0)); + Publication *publication; + List *sequences; + + /* stuff done only on the first call of the function */ + if (SRF_IS_FIRSTCALL()) + { + MemoryContext oldcontext; + + /* create a function context for cross-call persistence */ + funcctx = SRF_FIRSTCALL_INIT(); + + /* switch to memory context appropriate for multiple function calls */ + oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx); + + publication = GetPublicationByName(pubname, false); + + /* + * Publications support partitioned tables, although all changes are + * replicated using leaf partition identity and schema, so we only + * need those. + */ + if (publication->allsequences) + sequences = GetAllSequencesPublicationRelations(); + else + sequences = GetPublicationSequenceRelations(publication->oid); + + funcctx->user_fctx = (void *) sequences; + + MemoryContextSwitchTo(oldcontext); + } + + /* stuff done on every call of the function */ + funcctx = SRF_PERCALL_SETUP(); + sequences = (List *) funcctx->user_fctx; + + if (funcctx->call_cntr < list_length(sequences)) + { + Oid relid = list_nth_oid(sequences, funcctx->call_cntr); + + SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid)); + } + + SRF_RETURN_DONE(funcctx); +} diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 3cb69b1f87b..b5cc33aca34 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -374,6 +374,16 @@ CREATE VIEW pg_publication_tables AS pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace) WHERE C.oid = GPT.relid; +CREATE VIEW pg_publication_sequences AS + SELECT + P.pubname AS pubname, + N.nspname AS schemaname, + C.relname AS sequencename + FROM pg_publication P, + LATERAL pg_get_publication_sequences(P.pubname) GPT, + pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace) + WHERE C.oid = GPT.relid; + CREATE VIEW pg_locks AS SELECT * FROM pg_lock_status() AS L; diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 0e4bb97fb73..3bc2e8ccb66 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -16,6 +16,7 @@ #include "access/genam.h" #include "access/htup_details.h" +#include "access/relation.h" #include "access/table.h" #include "access/xact.h" #include "catalog/catalog.h" @@ -59,6 +60,12 @@ static void PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists, AlterPublicationStmt *stmt); static void PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok); +static List *OpenSequenceList(List *sequences); +static void CloseSequenceList(List *rels); +static void PublicationAddSequences(Oid pubid, List *rels, bool if_not_exists, + AlterPublicationStmt *stmt); +static void PublicationDropSequences(Oid pubid, List *rels, bool missing_ok); + static void parse_publication_options(ParseState *pstate, List *options, @@ -77,6 +84,7 @@ parse_publication_options(ParseState *pstate, pubactions->pubupdate = true; pubactions->pubdelete = true; pubactions->pubtruncate = true; + pubactions->pubsequence = true; *publish_via_partition_root = false; /* Parse options */ @@ -101,6 +109,7 @@ parse_publication_options(ParseState *pstate, pubactions->pubupdate = false; pubactions->pubdelete = false; pubactions->pubtruncate = false; + pubactions->pubsequence = false; *publish_given = true; publish = defGetString(defel); @@ -123,6 +132,8 @@ parse_publication_options(ParseState *pstate, pubactions->pubdelete = true; else if (strcmp(publish_opt, "truncate") == 0) pubactions->pubtruncate = true; + else if (strcmp(publish_opt, "sequence") == 0) + pubactions->pubsequence = true; else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -149,7 +160,9 @@ parse_publication_options(ParseState *pstate, */ static void ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate, - List **rels, List **schemas) + List **tables, List **sequences, + List **tables_schemas, List **sequences_schemas, + List **schemas) { ListCell *cell; PublicationObjSpec *pubobj; @@ -167,12 +180,23 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate, switch (pubobj->pubobjtype) { case PUBLICATIONOBJ_TABLE: - *rels = lappend(*rels, pubobj->pubtable); + *tables = lappend(*tables, pubobj->pubtable); + break; + case PUBLICATIONOBJ_SEQUENCE: + *sequences = lappend(*sequences, pubobj->pubtable); break; case PUBLICATIONOBJ_TABLES_IN_SCHEMA: schemaid = get_namespace_oid(pubobj->name, false); /* Filter out duplicates if user specifies "sch1, sch1" */ + *tables_schemas = list_append_unique_oid(*tables_schemas, schemaid); + *schemas = list_append_unique_oid(*schemas, schemaid); + break; + case PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA: + schemaid = get_namespace_oid(pubobj->name, false); + + /* Filter out duplicates if user specifies "sch1, sch1" */ + *sequences_schemas = list_append_unique_oid(*sequences_schemas, schemaid); *schemas = list_append_unique_oid(*schemas, schemaid); break; case PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA: @@ -186,6 +210,21 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate, list_free(search_path); /* Filter out duplicates if user specifies "sch1, sch1" */ + *tables_schemas = list_append_unique_oid(*tables_schemas, schemaid); + *schemas = list_append_unique_oid(*schemas, schemaid); + break; + case PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA: + search_path = fetch_search_path(false); + if (search_path == NIL) /* nothing valid in search_path? */ + ereport(ERROR, + errcode(ERRCODE_UNDEFINED_SCHEMA), + errmsg("no schema has been selected for CURRENT_SCHEMA")); + + schemaid = linitial_oid(search_path); + list_free(search_path); + + /* Filter out duplicates if user specifies "sch1, sch1" */ + *sequences_schemas = list_append_unique_oid(*sequences_schemas, schemaid); *schemas = list_append_unique_oid(*schemas, schemaid); break; default: @@ -251,7 +290,10 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) bool publish_via_partition_root_given; bool publish_via_partition_root; AclResult aclresult; - List *relations = NIL; + List *tables = NIL; + List *sequences = NIL; + List *tables_schemaidlist = NIL; + List *sequences_schemaidlist = NIL; List *schemaidlist = NIL; /* must have CREATE privilege on database */ @@ -306,6 +348,8 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) BoolGetDatum(pubactions.pubdelete); values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(pubactions.pubtruncate); + values[Anum_pg_publication_pubsequence - 1] = + BoolGetDatum(pubactions.pubsequence); values[Anum_pg_publication_pubviaroot - 1] = BoolGetDatum(publish_via_partition_root); @@ -330,26 +374,40 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) } else { - ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations, + ObjectsInPublicationToOids(stmt->pubobjects, pstate, + &tables, &sequences, + &tables_schemaidlist, + &sequences_schemaidlist, &schemaidlist); /* FOR ALL TABLES IN SCHEMA requires superuser */ - if (list_length(schemaidlist) > 0 && !superuser()) + if (list_length(tables_schemaidlist) > 0 && !superuser()) ereport(ERROR, errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("must be superuser to create FOR ALL TABLES IN SCHEMA publication")); - if (list_length(relations) > 0) + if (list_length(tables) > 0) { List *rels; - rels = OpenTableList(relations); - CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist, + rels = OpenTableList(tables); + CheckObjSchemaNotAlreadyInPublication(rels, tables_schemaidlist, PUBLICATIONOBJ_TABLE); PublicationAddTables(puboid, rels, true, NULL); CloseTableList(rels); } + if (list_length(sequences) > 0) + { + List *rels; + + rels = OpenSequenceList(sequences); + CheckObjSchemaNotAlreadyInPublication(rels, sequences_schemaidlist, + PUBLICATIONOBJ_SEQUENCE); + PublicationAddTables(puboid, rels, true, NULL); + CloseSequenceList(rels); + } + if (list_length(schemaidlist) > 0) { /* @@ -653,12 +711,13 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt, */ static void CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, - List *tables, List *schemaidlist) + List *tables, List *tables_schemaidlist, + List *sequences, List *sequences_schemaidlist) { Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup); if ((stmt->action == AP_AddObjects || stmt->action == AP_SetObjects) && - schemaidlist && !superuser()) + (tables_schemaidlist || sequences_schemaidlist) && !superuser()) ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("must be superuser to add or set schemas"))); @@ -667,13 +726,24 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, * Check that user is allowed to manipulate the publication tables in * schema */ - if (schemaidlist && pubform->puballtables) + if (tables_schemaidlist && pubform->puballtables) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("publication \"%s\" is defined as FOR ALL TABLES", NameStr(pubform->pubname)), errdetail("Tables from schema cannot be added to, dropped from, or set on FOR ALL TABLES publications."))); + /* + * Check that user is allowed to manipulate the publication sequences in + * schema + */ + if (sequences_schemaidlist && pubform->puballsequences) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("publication \"%s\" is defined as FOR ALL SEQUENCES", + NameStr(pubform->pubname)), + errdetail("Sequences from schema cannot be added to, dropped from, or set on FOR ALL SEQUENCES publications."))); + /* Check that user is allowed to manipulate the publication tables. */ if (tables && pubform->puballtables) ereport(ERROR, @@ -681,6 +751,108 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, errmsg("publication \"%s\" is defined as FOR ALL TABLES", NameStr(pubform->pubname)), errdetail("Tables cannot be added to or dropped from FOR ALL TABLES publications."))); + + /* Check that user is allowed to manipulate the publication tables. */ + if (sequences && pubform->puballsequences) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("publication \"%s\" is defined as FOR ALL SEQUENCES", + NameStr(pubform->pubname)), + errdetail("Sequences cannot be added to or dropped from FOR ALL SEQUENCES publications."))); +} + +/* + * Add or remove sequence to/from publication. + */ +static void +AlterPublicationSequences(AlterPublicationStmt *stmt, HeapTuple tup, + List *sequences, List *schemaidlist) +{ + List *rels = NIL; + Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup); + Oid pubid = pubform->oid; + + /* + * It is quite possible that for the SET case user has not specified any + * tables in which case we need to remove all the existing tables. + */ + if (!sequences && stmt->action != AP_SetObjects) + return; + + rels = OpenSequenceList(sequences); + + if (stmt->action == AP_AddObjects) + { + List *schemas = NIL; + + /* + * Check if the relation is member of the existing schema in the + * publication or member of the schema list specified. + */ + schemas = list_concat_copy(schemaidlist, GetPublicationSchemas(pubid)); + CheckObjSchemaNotAlreadyInPublication(rels, schemas, + PUBLICATIONOBJ_SEQUENCE); + PublicationAddSequences(pubid, rels, false, stmt); + } + else if (stmt->action == AP_DropObjects) + PublicationDropSequences(pubid, rels, false); + else /* DEFELEM_SET */ + { + List *oldrelids = GetPublicationRelations(pubid, + PUBLICATION_PART_ROOT); + List *delrels = NIL; + ListCell *oldlc; + + CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist, + PUBLICATIONOBJ_SEQUENCE); + + /* Calculate which relations to drop. */ + foreach(oldlc, oldrelids) + { + Oid oldrelid = lfirst_oid(oldlc); + ListCell *newlc; + bool found = false; + + foreach(newlc, rels) + { + PublicationRelInfo *newpubrel; + + newpubrel = (PublicationRelInfo *) lfirst(newlc); + if (RelationGetRelid(newpubrel->relation) == oldrelid) + { + found = true; + break; + } + } + /* Not yet in the list, open it and add to the list */ + if (!found) + { + Relation oldrel; + PublicationRelInfo *pubrel; + + /* Wrap relation into PublicationRelInfo */ + oldrel = table_open(oldrelid, ShareUpdateExclusiveLock); + + pubrel = palloc(sizeof(PublicationRelInfo)); + pubrel->relation = oldrel; + + delrels = lappend(delrels, pubrel); + } + } + + /* And drop them. */ + PublicationDropSequences(pubid, delrels, true); + + /* + * Don't bother calculating the difference for adding, we'll catch and + * skip existing ones when doing catalog update. + */ + PublicationAddSequences(pubid, rels, true, stmt); + + CloseSequenceList(delrels); + } + + CloseSequenceList(rels); } /* @@ -718,13 +890,21 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt) AlterPublicationOptions(pstate, stmt, rel, tup); else { - List *relations = NIL; + List *tables = NIL; + List *sequences = NIL; + List *tables_schemaidlist = NIL; + List *sequences_schemaidlist = NIL; List *schemaidlist = NIL; - ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations, + ObjectsInPublicationToOids(stmt->pubobjects, pstate, + &tables, &sequences, + &tables_schemaidlist, + &sequences_schemaidlist, &schemaidlist); - CheckAlterPublication(stmt, tup, relations, schemaidlist); + CheckAlterPublication(stmt, tup, + tables, tables_schemaidlist, + sequences, sequences_schemaidlist); /* * Lock the publication so nobody else can do anything with it. This @@ -749,7 +929,9 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt) errmsg("publication \"%s\" does not exist", stmt->pubname)); - AlterPublicationTables(stmt, tup, relations, schemaidlist); + AlterPublicationTables(stmt, tup, tables, tables_schemaidlist); + AlterPublicationSequences(stmt, tup, sequences, sequences_schemaidlist); + AlterPublicationSchemas(stmt, tup, schemaidlist); } @@ -1157,6 +1339,144 @@ PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok) } } +/* + * Open relations specified by a PublicationTable list. + * In the returned list of PublicationRelInfo, tables are locked + * in ShareUpdateExclusiveLock mode in order to add them to a publication. + */ +static List * +OpenSequenceList(List *sequences) +{ + List *relids = NIL; + List *rels = NIL; + ListCell *lc; + + /* + * Open, share-lock, and check all the explicitly-specified relations + */ + foreach(lc, sequences) + { + PublicationTable *s = lfirst_node(PublicationTable, lc); + Relation rel; + Oid myrelid; + PublicationRelInfo *pub_rel; + + /* Allow query cancel in case this takes a long time */ + CHECK_FOR_INTERRUPTS(); + + rel = table_openrv(s->relation, ShareUpdateExclusiveLock); + myrelid = RelationGetRelid(rel); + + /* + * Filter out duplicates if user specifies "foo, foo". + * + * Note that this algorithm is known to not be very efficient (O(N^2)) + * but given that it only works on list of tables given to us by user + * it's deemed acceptable. + */ + if (list_member_oid(relids, myrelid)) + { + table_close(rel, ShareUpdateExclusiveLock); + continue; + } + + pub_rel = palloc(sizeof(PublicationRelInfo)); + pub_rel->relation = rel; + rels = lappend(rels, pub_rel); + relids = lappend_oid(relids, myrelid); + } + + list_free(relids); + + return rels; +} + +/* + * Close all relations in the list. + */ +static void +CloseSequenceList(List *rels) +{ + ListCell *lc; + + foreach(lc, rels) + { + PublicationRelInfo *pub_rel; + + pub_rel = (PublicationRelInfo *) lfirst(lc); + table_close(pub_rel->relation, NoLock); + } +} + +/* + * Add listed tables to the publication. + */ +static void +PublicationAddSequences(Oid pubid, List *rels, bool if_not_exists, + AlterPublicationStmt *stmt) +{ + ListCell *lc; + + Assert(!stmt || !stmt->for_all_sequences); + + foreach(lc, rels) + { + PublicationRelInfo *pub_rel = (PublicationRelInfo *) lfirst(lc); + Relation rel = pub_rel->relation; + ObjectAddress obj; + + /* Must be owner of the sequence or superuser. */ + if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind), + RelationGetRelationName(rel)); + + obj = publication_add_relation(pubid, pub_rel, if_not_exists); + if (stmt) + { + EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress, + (Node *) stmt); + + InvokeObjectPostCreateHook(PublicationRelRelationId, + obj.objectId, 0); + } + } +} + +/* + * Remove listed sequences from the publication. + */ +static void +PublicationDropSequences(Oid pubid, List *rels, bool missing_ok) +{ + ObjectAddress obj; + ListCell *lc; + Oid prid; + + foreach(lc, rels) + { + Relation rel = (Relation) lfirst(lc); + Oid relid = RelationGetRelid(rel); + + prid = GetSysCacheOid2(PUBLICATIONRELMAP, Anum_pg_publication_rel_oid, + ObjectIdGetDatum(relid), + ObjectIdGetDatum(pubid)); + if (!OidIsValid(prid)) + { + if (missing_ok) + continue; + + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("relation \"%s\" is not part of the publication", + RelationGetRelationName(rel)))); + } + + ObjectAddressSet(obj, PublicationRelRelationId, prid); + performDeletion(&obj, DROP_CASCADE, 0); + } +} + + /* * Internal workhorse for changing a publication owner */ diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c index ab592ce2f15..fe4f21ec438 100644 --- a/src/backend/commands/sequence.c +++ b/src/backend/commands/sequence.c @@ -336,6 +336,85 @@ ResetSequence(Oid seq_relid) relation_close(seq_rel, NoLock); } +/* + * Reset a sequence to its initial value. + * + * The change is made transactionally, so that on failure of the current + * transaction, the sequence will be restored to its previous state. + * We do that by creating a whole new relfilenode for the sequence; so this + * works much like the rewriting forms of ALTER TABLE. + * + * Caller is assumed to have acquired AccessExclusiveLock on the sequence, + * which must not be released until end of transaction. Caller is also + * responsible for permissions checking. + */ +void +ResetSequence2(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called) +{ + Relation seq_rel; + SeqTable elm; + Form_pg_sequence_data seq; + Buffer buf; + HeapTupleData seqdatatuple; + HeapTuple tuple; + + /* + * Read the old sequence. This does a bit more work than really + * necessary, but it's simple, and we do want to double-check that it's + * indeed a sequence. + */ + init_sequence(seq_relid, &elm, &seq_rel); + (void) read_seq_tuple(seq_rel, &buf, &seqdatatuple); + + /* + * Copy the existing sequence tuple. + */ + tuple = heap_copytuple(&seqdatatuple); + + /* Now we're done with the old page */ + UnlockReleaseBuffer(buf); + + /* + * Modify the copied tuple to execute the restart (compare the RESTART + * action in AlterSequence) + */ + seq = (Form_pg_sequence_data) GETSTRUCT(tuple); + seq->last_value = last_value; + seq->is_called = is_called; + seq->log_cnt = log_cnt; + + /* + * Create a new storage file for the sequence. + */ + RelationSetNewRelfilenode(seq_rel, seq_rel->rd_rel->relpersistence); + + /* + * Ensure sequence's relfrozenxid is at 0, since it won't contain any + * unfrozen XIDs. Same with relminmxid, since a sequence will never + * contain multixacts. + */ + Assert(seq_rel->rd_rel->relfrozenxid == InvalidTransactionId); + Assert(seq_rel->rd_rel->relminmxid == InvalidMultiXactId); + + /* + * Insert the modified tuple into the new storage file. + * + * XXX Maybe this should also use created=true, just like the other places + * calling fill_seq_with_data. That's probably needed for correct cascading + * replication. + * + * XXX That'd mean all fill_seq_with_data callers use created=true, making + * the parameter unnecessary. + */ + fill_seq_with_data(seq_rel, tuple); + + /* Clear local cache so that we don't think we have cached numbers */ + /* Note that we do not change the currval() state */ + elm->cached = elm->last; + + relation_close(seq_rel, NoLock); +} + /* * Initialize a sequence's relation with the specified tuple as content */ diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 3ef6607d246..5beb67e7652 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -85,6 +85,7 @@ typedef struct SubOpts } SubOpts; static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); +static List *fetch_sequence_list(WalReceiverConn *wrconn, List *publications); static void check_duplicates_in_publist(List *publist, Datum *datums); static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname); static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); @@ -496,6 +497,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, char *err; WalReceiverConn *wrconn; List *tables; + List *sequences; ListCell *lc; char table_state; @@ -534,6 +536,26 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, InvalidXLogRecPtr); } + /* + * Get the sequence list from publisher and build local sequence + * status info. + */ + sequences = fetch_sequence_list(wrconn, publications); + foreach(lc, sequences) + { + RangeVar *rv = (RangeVar *) lfirst(lc); + Oid relid; + + relid = RangeVarGetRelid(rv, AccessShareLock, false); + + /* Check for supported relkind. */ + CheckSubscriptionRelkind(get_rel_relkind(relid), + rv->schemaname, rv->relname); + + AddSubscriptionRelState(subid, relid, table_state, + InvalidXLogRecPtr); + } + /* * If requested, create permanent slot for the subscription. We * won't use the initial snapshot for anything, so no need to @@ -706,6 +728,10 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) { Oid relid = subrel_local_oids[off]; + /* XXX ignore sequences - maybe do this in GetSubscriptionRelations? */ + if (get_rel_relkind(relid) == RELKIND_SEQUENCE) + continue; + if (!bsearch(&relid, pubrel_local_oids, list_length(pubrel_names), sizeof(Oid), oid_cmp)) { @@ -797,6 +823,183 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); } } + + /* + * XXX now do the same thing for sequences, maybe before the preceding + * block, or earlier? + */ + + /* Get the table list from publisher. */ + pubrel_names = fetch_sequence_list(wrconn, sub->publications); + + /* Get local table list. */ + subrel_states = GetSubscriptionRelations(sub->oid); + + /* + * Build qsorted array of local table oids for faster lookup. This can + * potentially contain all tables in the database so speed of lookup + * is important. + */ + subrel_local_oids = palloc(list_length(subrel_states) * sizeof(Oid)); + off = 0; + foreach(lc, subrel_states) + { + SubscriptionRelState *relstate = (SubscriptionRelState *) lfirst(lc); + + subrel_local_oids[off++] = relstate->relid; + } + qsort(subrel_local_oids, list_length(subrel_states), + sizeof(Oid), oid_cmp); + + /* + * Rels that we want to remove from subscription and drop any slots + * and origins corresponding to them. + */ + sub_remove_rels = palloc(list_length(subrel_states) * sizeof(SubRemoveRels)); + + /* + * Walk over the remote tables and try to match them to locally known + * tables. If the table is not known locally create a new state for + * it. + * + * Also builds array of local oids of remote tables for the next step. + */ + off = 0; + pubrel_local_oids = palloc(list_length(pubrel_names) * sizeof(Oid)); + + foreach(lc, pubrel_names) + { + RangeVar *rv = (RangeVar *) lfirst(lc); + Oid relid; + + relid = RangeVarGetRelid(rv, AccessShareLock, false); + + /* Check for supported relkind. */ + CheckSubscriptionRelkind(get_rel_relkind(relid), + rv->schemaname, rv->relname); + + pubrel_local_oids[off++] = relid; + + if (!bsearch(&relid, subrel_local_oids, + list_length(subrel_states), sizeof(Oid), oid_cmp)) + { + AddSubscriptionRelState(sub->oid, relid, + copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY, + InvalidXLogRecPtr); + ereport(DEBUG1, + (errmsg_internal("table \"%s.%s\" added to subscription \"%s\"", + rv->schemaname, rv->relname, sub->name))); + } + } + + /* + * Next remove state for tables we should not care about anymore using + * the data we collected above + */ + qsort(pubrel_local_oids, list_length(pubrel_names), + sizeof(Oid), oid_cmp); + + remove_rel_len = 0; + for (off = 0; off < list_length(subrel_states); off++) + { + Oid relid = subrel_local_oids[off]; + + /* XXX ignore non-sequences - maybe do this in GetSubscriptionRelations? */ + if (get_rel_relkind(relid) != RELKIND_SEQUENCE) + continue; + + if (!bsearch(&relid, pubrel_local_oids, + list_length(pubrel_names), sizeof(Oid), oid_cmp)) + { + char state; + XLogRecPtr statelsn; + + /* + * Lock pg_subscription_rel with AccessExclusiveLock to + * prevent any race conditions with the apply worker + * re-launching workers at the same time this code is trying + * to remove those tables. + * + * Even if new worker for this particular rel is restarted it + * won't be able to make any progress as we hold exclusive + * lock on subscription_rel till the transaction end. It will + * simply exit as there is no corresponding rel entry. + * + * This locking also ensures that the state of rels won't + * change till we are done with this refresh operation. + */ + if (!rel) + rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock); + + /* Last known rel state. */ + state = GetSubscriptionRelState(sub->oid, relid, &statelsn); + + sub_remove_rels[remove_rel_len].relid = relid; + sub_remove_rels[remove_rel_len++].state = state; + + RemoveSubscriptionRel(sub->oid, relid); + + logicalrep_worker_stop(sub->oid, relid); + + /* + * For READY state, we would have already dropped the + * tablesync origin. + */ + if (state != SUBREL_STATE_READY) + { + char originname[NAMEDATALEN]; + + /* + * Drop the tablesync's origin tracking if exists. + * + * It is possible that the origin is not yet created for + * tablesync worker, this can happen for the states before + * SUBREL_STATE_FINISHEDCOPY. The apply worker can also + * concurrently try to drop the origin and by this time + * the origin might be already removed. For these reasons, + * passing missing_ok = true. + */ + ReplicationOriginNameForTablesync(sub->oid, relid, originname, + sizeof(originname)); + replorigin_drop_by_name(originname, true, false); + } + + ereport(DEBUG1, + (errmsg_internal("table \"%s.%s\" removed from subscription \"%s\"", + get_namespace_name(get_rel_namespace(relid)), + get_rel_name(relid), + sub->name))); + } + } + + /* + * Drop the tablesync slots associated with removed tables. This has + * to be at the end because otherwise if there is an error while doing + * the database operations we won't be able to rollback dropped slots. + */ + for (off = 0; off < remove_rel_len; off++) + { + if (sub_remove_rels[off].state != SUBREL_STATE_READY && + sub_remove_rels[off].state != SUBREL_STATE_SYNCDONE) + { + char syncslotname[NAMEDATALEN] = {0}; + + /* + * For READY/SYNCDONE states we know the tablesync slot has + * already been dropped by the tablesync worker. + * + * For other states, there is no certainty, maybe the slot + * does not exist yet. Also, if we fail after removing some of + * the slots, next time, it will again try to drop already + * dropped slots and fail. For these reasons, we allow + * missing_ok = true for the drop. + */ + ReplicationSlotNameForTablesync(sub->oid, sub_remove_rels[off].relid, + syncslotname, sizeof(syncslotname)); + ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); + } + } + } PG_FINALLY(); { @@ -1616,6 +1819,75 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) return tablelist; } +/* + * Get the list of sequences which belong to specified publications on the + * publisher connection. + */ +static List * +fetch_sequence_list(WalReceiverConn *wrconn, List *publications) +{ + WalRcvExecResult *res; + StringInfoData cmd; + TupleTableSlot *slot; + Oid tableRow[2] = {TEXTOID, TEXTOID}; + ListCell *lc; + bool first; + List *tablelist = NIL; + + Assert(list_length(publications) > 0); + + initStringInfo(&cmd); + appendStringInfoString(&cmd, "SELECT DISTINCT s.schemaname, s.sequencename\n" + " FROM pg_catalog.pg_publication_sequences s\n" + " WHERE s.pubname IN ("); + first = true; + foreach(lc, publications) + { + char *pubname = strVal(lfirst(lc)); + + if (first) + first = false; + else + appendStringInfoString(&cmd, ", "); + + appendStringInfoString(&cmd, quote_literal_cstr(pubname)); + } + appendStringInfoChar(&cmd, ')'); + + res = walrcv_exec(wrconn, cmd.data, 2, tableRow); + pfree(cmd.data); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errmsg("could not receive list of replicated tables from the publisher: %s", + res->err))); + + /* Process tables. */ + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + { + char *nspname; + char *relname; + bool isnull; + RangeVar *rv; + + nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); + Assert(!isnull); + + rv = makeRangeVar(nspname, relname, -1); + tablelist = lappend(tablelist, rv); + + ExecClearTuple(slot); + } + ExecDropSingleTupleTableSlot(slot); + + walrcv_clear_result(res); + + return tablelist; +} + /* * This is to report the connection failure while dropping replication slots. * Here, we report the WARNING for all tablesync slots so that user can drop diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index 313c87398b2..78f14119d98 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -608,7 +608,7 @@ void CheckSubscriptionRelkind(char relkind, const char *nspname, const char *relname) { - if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE) + if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE && relkind != RELKIND_SEQUENCE) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("cannot use relation \"%s.%s\" as logical replication target", diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 6bd95bbce24..8b7e9710401 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -4852,6 +4852,7 @@ _copyCreatePublicationStmt(const CreatePublicationStmt *from) COPY_NODE_FIELD(options); COPY_NODE_FIELD(pubobjects); COPY_SCALAR_FIELD(for_all_tables); + COPY_SCALAR_FIELD(for_all_sequences); return newnode; } diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c index 4126516222b..55a7dbbddf3 100644 --- a/src/backend/nodes/equalfuncs.c +++ b/src/backend/nodes/equalfuncs.c @@ -2337,6 +2337,7 @@ _equalAlterPublicationStmt(const AlterPublicationStmt *a, COMPARE_NODE_FIELD(options); COMPARE_NODE_FIELD(pubobjects); COMPARE_SCALAR_FIELD(for_all_tables); + COMPARE_SCALAR_FIELD(for_all_sequences); COMPARE_SCALAR_FIELD(action); return true; diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index c4f32425060..bbde765cf56 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -9771,6 +9771,26 @@ PublicationObjSpec: $$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA; $$->location = @5; } + | SEQUENCE relation_expr + { + $$ = makeNode(PublicationObjSpec); + $$->pubobjtype = PUBLICATIONOBJ_SEQUENCE; + $$->pubtable = makeNode(PublicationTable); + $$->pubtable->relation = $2; + } + | ALL SEQUENCES IN_P SCHEMA ColId + { + $$ = makeNode(PublicationObjSpec); + $$->pubobjtype = PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA; + $$->name = $5; + $$->location = @5; + } + | ALL SEQUENCES IN_P SCHEMA CURRENT_SCHEMA + { + $$ = makeNode(PublicationObjSpec); + $$->pubobjtype = PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA; + $$->location = @5; + } | ColId { $$ = makeNode(PublicationObjSpec); @@ -10106,6 +10126,12 @@ UnlistenStmt: } ; +/* + * FIXME + * + * opt_publication_for_sequences and publication_for_sequences should be + * copies for sequences + */ /***************************************************************************** * @@ -10114,6 +10140,12 @@ UnlistenStmt: * BEGIN / COMMIT / ROLLBACK * (also older versions END / ABORT) * + * ALTER PUBLICATION name ADD SEQUENCE sequence [, sequence2] + * + * ALTER PUBLICATION name DROP SEQUENCE sequence [, sequence2] + * + * ALTER PUBLICATION name SET SEQUENCE sequence [, sequence2] + * *****************************************************************************/ TransactionStmt: diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 953942692ce..e8ead1387ae 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -647,6 +647,56 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, pq_sendbytes(out, message, sz); } +/* + * Write SEQUENCE to stream + */ +void +logicalrep_write_sequence(StringInfo out, Relation rel, TransactionId xid, + XLogRecPtr lsn, bool transactional, + int64 last_value, int64 log_cnt, bool is_called) +{ + uint8 flags = 0; + char *relname; + + pq_sendbyte(out, LOGICAL_REP_MSG_SEQUENCE); + + /* transaction ID (if not valid, we're not streaming) */ + if (TransactionIdIsValid(xid)) + pq_sendint32(out, xid); + + pq_sendint8(out, flags); + pq_sendint64(out, lsn); + + logicalrep_write_namespace(out, RelationGetNamespace(rel)); + relname = RelationGetRelationName(rel); + pq_sendstring(out, relname); + + pq_sendint8(out, transactional); + pq_sendint64(out, last_value); + pq_sendint64(out, log_cnt); + pq_sendint8(out, is_called); +} + +/* + * Read SEQUENCE from the stream. + */ +void +logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata) +{ + /* XXX skipping flags and lsn */ + pq_getmsgint(in, 1); + pq_getmsgint64(in); + + /* Read relation name from stream */ + seqdata->nspname = pstrdup(logicalrep_read_namespace(in)); + seqdata->seqname = pstrdup(pq_getmsgstring(in)); + + seqdata->transactional = pq_getmsgint(in, 1); + seqdata->last_value = pq_getmsgint64(in); + seqdata->log_cnt = pq_getmsgint64(in); + seqdata->is_called = pq_getmsgint(in, 1); +} + /* * Write relation description to the output stream. */ @@ -1203,6 +1253,8 @@ logicalrep_message_type(LogicalRepMsgType action) return "STREAM ABORT"; case LOGICAL_REP_MSG_STREAM_PREPARE: return "STREAM PREPARE"; + case LOGICAL_REP_MSG_SEQUENCE: + return "SEQUENCE"; } elog(ERROR, "invalid logical replication message type \"%c\"", action); diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index e596b69d466..c85867ee46b 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -100,6 +100,7 @@ #include "catalog/pg_subscription_rel.h" #include "catalog/pg_type.h" #include "commands/copy.h" +#include "commands/sequence.h" #include "miscadmin.h" #include "parser/parse_relation.h" #include "pgstat.h" @@ -359,6 +360,12 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn) * * If the synchronization position is reached (SYNCDONE), then the table can * be marked as READY and is no longer tracked. + * + * XXX This needs to handle sequences too - after AlterSubscription_refresh + * starts caring about sequences, GetSubscriptionNotReadyRelations won't + * return just tables, and we'll have to sync them here. Not sure it's worth + * creating a new "sync" worker per sequence, maybe we should just sync them + * in the current process (it's pretty light-weight). */ static void process_syncing_tables_for_apply(XLogRecPtr current_lsn) @@ -873,6 +880,99 @@ copy_table(Relation rel) logicalrep_rel_close(relmapentry, NoLock); } + + +/* + * FIXME add comment + */ +static void +fetch_sequence_data(char *nspname, char *relname, + int64 *last_value, int64 *log_cnt, bool *is_called) +{ + WalRcvExecResult *res; + StringInfoData cmd; + TupleTableSlot *slot; + Oid tableRow[3] = {INT8OID, INT8OID, BOOLOID}; + + initStringInfo(&cmd); + appendStringInfo(&cmd, "SELECT last_value, log_cnt, is_called\n" + " FROM %s", quote_qualified_identifier(nspname, relname)); + + res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 3, tableRow); + pfree(cmd.data); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errmsg("could not receive list of replicated tables from the publisher: %s", + res->err))); + + /* Process the sequence. */ + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + { + bool isnull; + + *last_value = DatumGetInt64(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + + *log_cnt = DatumGetInt64(slot_getattr(slot, 2, &isnull)); + Assert(!isnull); + + *is_called = DatumGetBool(slot_getattr(slot, 3, &isnull)); + Assert(!isnull); + + ExecClearTuple(slot); + } + ExecDropSingleTupleTableSlot(slot); + + walrcv_clear_result(res); +} + +/* + * Copy existing data of a sequence from publisher. + * + * Caller is responsible for locking the local relation. + */ +static void +copy_sequence(Relation rel) +{ + LogicalRepRelMapEntry *relmapentry; + LogicalRepRelation lrel; + StringInfoData cmd; + int64 last_value = 0, + log_cnt = 0; + bool is_called = 0; + + /* Get the publisher relation info. */ + fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)), + RelationGetRelationName(rel), &lrel); + + /* Put the relation into relmap. */ + logicalrep_relmap_update(&lrel); + + /* Map the publisher relation to local one. */ + relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock); + Assert(rel == relmapentry->localrel); + + /* Start copy on the publisher. */ + initStringInfo(&cmd); + + Assert(lrel.relkind == RELKIND_SEQUENCE); + + fetch_sequence_data(lrel.nspname, lrel.relname, &last_value, &log_cnt, &is_called); + + elog(WARNING, "sequence %s info last_value %ld log_cnt %ld is_called %d", + quote_qualified_identifier(lrel.nspname, lrel.relname), + last_value, log_cnt, is_called); + + ResetSequence2(RelationGetRelid(rel), last_value, log_cnt, is_called); + + logicalrep_rel_close(relmapentry, NoLock); +} + + + + /* * Determine the tablesync slot name. * @@ -1134,10 +1234,20 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) originname))); } - /* Now do the initial data copy */ - PushActiveSnapshot(GetTransactionSnapshot()); - copy_table(rel); - PopActiveSnapshot(); + if (get_rel_relkind(RelationGetRelid(rel)) == RELKIND_SEQUENCE) + { + /* Now do the initial sequence copy */ + PushActiveSnapshot(GetTransactionSnapshot()); + copy_sequence(rel); + PopActiveSnapshot(); + } + else + { + /* Now do the initial data copy */ + PushActiveSnapshot(GetTransactionSnapshot()); + copy_table(rel); + PopActiveSnapshot(); + } res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL); if (res->status != WALRCV_OK_COMMAND) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index d77bb32bb9e..68708d3907f 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -144,6 +144,7 @@ #include "catalog/pg_tablespace.h" #include "commands/tablecmds.h" #include "commands/tablespace.h" +#include "commands/sequence.h" #include "commands/trigger.h" #include "executor/executor.h" #include "executor/execPartition.h" @@ -1093,6 +1094,61 @@ apply_handle_origin(StringInfo s) errmsg_internal("ORIGIN message sent out of order"))); } +/* + * Handle SEQUENCE message. + */ +static void +apply_handle_sequence(StringInfo s) +{ + LogicalRepSequence seq; + Oid relid; + + if (handle_streamed_transaction(LOGICAL_REP_MSG_SEQUENCE, s)) + return; + + logicalrep_read_sequence(s, &seq); + + /* + * Non-transactional sequence updates should not be part of a remote + * transaction. There should not be any running transaction. + */ + Assert((!seq.transactional) || in_remote_transaction); + Assert(!(!seq.transactional && in_remote_transaction)); + Assert(!(!seq.transactional && IsTransactionState())); + + /* + * Make sure we're in a transaction (needed by ResetSequence2). For + * non-transactional updates we're guaranteed to start a new one, + * and we'll commit it at the end. + */ + if (!IsTransactionState()) + { + StartTransactionCommand(); + maybe_reread_subscription(); + } + + relid = RangeVarGetRelid(makeRangeVar(seq.nspname, + seq.seqname, -1), + RowExclusiveLock, false); + + /* lock the sequence in AccessExclusiveLock, as expected by ResetSequence2 */ + elog(WARNING, "locking sequence %d in exclusive mode", relid); + LockRelationOid(relid, AccessExclusiveLock); + + elog(WARNING, "applying sequence %s.%s transactional %d last_value %ld log_cnt %ld is_called %d", + seq.nspname, seq.seqname, seq.transactional, seq.last_value, seq.log_cnt, seq.is_called); + + /* apply the sequence change */ + ResetSequence2(relid, seq.last_value, seq.log_cnt, seq.is_called); + + /* + * Commit the per-stream transaction (we only do this when not in + * remote transaction, i.e. for non-transactional sequence updates. + */ + if (!in_remote_transaction) + CommitTransactionCommand(); +} + /* * Handle STREAM START message. */ @@ -2421,6 +2477,10 @@ apply_dispatch(StringInfo s) */ break; + case LOGICAL_REP_MSG_SEQUENCE: + apply_handle_sequence(s); + return; + case LOGICAL_REP_MSG_STREAM_START: apply_handle_stream_start(s); break; diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 6df705f90ff..b82c6b10305 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -49,6 +49,10 @@ static void pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message); +static void pgoutput_sequence(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr sequence_lsn, + Relation rel, bool transactional, + int64 last_value, int64 log_cnt, bool is_called); static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, RepOriginId origin_id); static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, @@ -161,6 +165,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->change_cb = pgoutput_change; cb->truncate_cb = pgoutput_truncate; cb->message_cb = pgoutput_message; + cb->sequence_cb = pgoutput_sequence; cb->commit_cb = pgoutput_commit_txn; cb->begin_prepare_cb = pgoutput_begin_prepare_txn; @@ -177,6 +182,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->stream_commit_cb = pgoutput_stream_commit; cb->stream_change_cb = pgoutput_change; cb->stream_message_cb = pgoutput_message; + cb->stream_sequence_cb = pgoutput_sequence; cb->stream_truncate_cb = pgoutput_truncate; /* transaction streaming - two-phase commit */ cb->stream_prepare_cb = pgoutput_stream_prepare_txn; @@ -190,6 +196,7 @@ parse_output_parameters(List *options, PGOutputData *data) bool publication_names_given = false; bool binary_option_given = false; bool messages_option_given = false; + bool sequences_option_given = false; bool streaming_given = false; bool two_phase_option_given = false; @@ -197,6 +204,7 @@ parse_output_parameters(List *options, PGOutputData *data) data->streaming = false; data->messages = false; data->two_phase = false; + data->sequences = true; foreach(lc, options) { @@ -262,6 +270,16 @@ parse_output_parameters(List *options, PGOutputData *data) data->messages = defGetBoolean(defel); } + else if (strcmp(defel->defname, "sequences") == 0) + { + if (sequences_option_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + sequences_option_given = true; + + data->sequences = defGetBoolean(defel); + } else if (strcmp(defel->defname, "streaming") == 0) { if (streaming_given) @@ -858,6 +876,51 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, OutputPluginWrite(ctx, true); } +static void +pgoutput_sequence(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr sequence_lsn, + Relation rel, bool transactional, + int64 last_value, int64 log_cnt, bool is_called) +{ + PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; + TransactionId xid = InvalidTransactionId; + RelationSyncEntry *relentry; + + if (!data->sequences) + return; + + if (!is_publishable_relation(rel)) + return; + + /* + * Remember the xid for the message in streaming mode. See + * pgoutput_change. + */ + if (in_streaming) + xid = txn->xid; + + relentry = get_rel_sync_entry(data, RelationGetRelid(rel)); + + /* + * First check the sequence filter. + * + * We handle just REORDER_BUFFER_CHANGE_SEQUENCE here. + */ + if (!relentry->pubactions.pubsequence) + return; + + OutputPluginPrepareWrite(ctx, true); + logicalrep_write_sequence(ctx->out, + rel, + xid, + sequence_lsn, + transactional, + last_value, + log_cnt, + is_called); + OutputPluginWrite(ctx, true); +} + /* * Currently we always forward. */ @@ -1141,7 +1204,8 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->schema_sent = false; entry->streamed_txns = NIL; entry->pubactions.pubinsert = entry->pubactions.pubupdate = - entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false; + entry->pubactions.pubdelete = entry->pubactions.pubtruncate = + entry->pubactions.pubsequence = false; entry->publish_as_relid = InvalidOid; entry->map = NULL; /* will be set by maybe_send_schema() if * needed */ @@ -1163,6 +1227,7 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) Oid publish_as_relid = relid; bool am_partition = get_rel_relispartition(relid); char relkind = get_rel_relkind(relid); + bool is_sequence = (get_rel_relkind(relid) == RELKIND_SEQUENCE); /* Reload publications if needed before use. */ if (!publications_valid) @@ -1191,6 +1256,7 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->pubactions.pubupdate = false; entry->pubactions.pubdelete = false; entry->pubactions.pubtruncate = false; + entry->pubactions.pubsequence = false; if (entry->map) { /* @@ -1213,12 +1279,23 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) Publication *pub = lfirst(lc); bool publish = false; - if (pub->alltables) + if (pub->alltables && (!is_sequence)) { publish = true; if (pub->pubviaroot && am_partition) publish_as_relid = llast_oid(get_partition_ancestors(relid)); } + else if (pub->allsequences && is_sequence) + { + publish = true; + } + + /* if a sequence, just cross-check the list of publications */ + if (!publish && is_sequence) + { + if (list_member_oid(pubids, pub->oid)) + publish = true; + } if (!publish) { @@ -1275,10 +1352,12 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->pubactions.pubupdate |= pub->pubactions.pubupdate; entry->pubactions.pubdelete |= pub->pubactions.pubdelete; entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate; + entry->pubactions.pubsequence |= pub->pubactions.pubsequence; } if (entry->pubactions.pubinsert && entry->pubactions.pubupdate && - entry->pubactions.pubdelete && entry->pubactions.pubtruncate) + entry->pubactions.pubdelete && entry->pubactions.pubtruncate && + entry->pubactions.pubsequence) break; } diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 2707fed12f4..45a8b3e490a 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -5586,6 +5586,7 @@ GetRelationPublicationActions(Relation relation) pubactions->pubupdate |= pubform->pubupdate; pubactions->pubdelete |= pubform->pubdelete; pubactions->pubtruncate |= pubform->pubtruncate; + pubactions->pubsequence |= pubform->pubsequence; ReleaseSysCache(tup); @@ -5594,7 +5595,8 @@ GetRelationPublicationActions(Relation relation) * other publications. */ if (pubactions->pubinsert && pubactions->pubupdate && - pubactions->pubdelete && pubactions->pubtruncate) + pubactions->pubdelete && pubactions->pubtruncate && + pubactions->pubsequence) break; } diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index 98882272130..4900c48ff19 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1782,20 +1782,20 @@ psql_completion(const char *text, int start, int end) COMPLETE_WITH("ADD", "DROP", "OWNER TO", "RENAME TO", "SET"); /* ALTER PUBLICATION <name> ADD */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD")) - COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE"); - else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") || - (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") && + COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE|SEQUENCE"); + else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE|SEQUENCE") || + (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE|SEQUENCE") && ends_with(prev_wd, ','))) COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables); - else if (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE")) + else if (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE|SEQUENCE")) COMPLETE_WITH(","); /* ALTER PUBLICATION <name> DROP */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP")) - COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE"); + COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE|SEQUENCE"); /* ALTER PUBLICATION <name> SET */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET")) - COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "TABLE"); - else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES", "IN", "SCHEMA")) + COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "TABLE|SEQUENCE"); + else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA")) COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas " AND nspname NOT LIKE E'pg\\\\_%'", "CURRENT_SCHEMA"); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 62f36daa981..a66eb8109d0 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11546,6 +11546,11 @@ provolatile => 's', prorettype => 'oid', proargtypes => 'text', proallargtypes => '{text,oid}', proargmodes => '{i,o}', proargnames => '{pubname,relid}', prosrc => 'pg_get_publication_tables' }, +{ oid => '8000', descr => 'get OIDs of sequences in a publication', + proname => 'pg_get_publication_sequences', prorows => '1000', proretset => 't', + provolatile => 's', prorettype => 'oid', proargtypes => 'text', + proallargtypes => '{text,oid}', proargmodes => '{i,o}', + proargnames => '{pubname,relid}', prosrc => 'pg_get_publication_sequences' }, { oid => '6121', descr => 'returns whether a relation can be part of a publication', proname => 'pg_relation_is_publishable', provolatile => 's', diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 841b9b6c253..e56286772f4 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -40,6 +40,12 @@ CATALOG(pg_publication,6104,PublicationRelationId) */ bool puballtables; + /* + * indicates that this is special publication which should encompass all + * sequences in the database (except for the unlogged and temp ones) + */ + bool puballsequences; + /* true if inserts are published */ bool pubinsert; @@ -52,6 +58,9 @@ CATALOG(pg_publication,6104,PublicationRelationId) /* true if truncates are published */ bool pubtruncate; + /* true if sequences are published */ + bool pubsequence; + /* true if partition changes are published using root schema */ bool pubviaroot; } FormData_pg_publication; @@ -72,6 +81,7 @@ typedef struct PublicationActions bool pubupdate; bool pubdelete; bool pubtruncate; + bool pubsequence; } PublicationActions; typedef struct Publication @@ -79,6 +89,7 @@ typedef struct Publication Oid oid; char *name; bool alltables; + bool allsequences; bool pubviaroot; PublicationActions pubactions; } Publication; @@ -121,6 +132,9 @@ extern List *GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt, Oid relid); +extern List *GetAllSequencesPublicationRelations(void); +extern List *GetPublicationSequenceRelations(Oid pubid); + extern bool is_publishable_relation(Relation rel); extern bool is_schema_publication(Oid pubid); extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *targetrel, diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h index 9fecc41954e..d8c255a7af5 100644 --- a/src/include/commands/sequence.h +++ b/src/include/commands/sequence.h @@ -60,6 +60,7 @@ extern ObjectAddress DefineSequence(ParseState *pstate, CreateSeqStmt *stmt); extern ObjectAddress AlterSequence(ParseState *pstate, AlterSeqStmt *stmt); extern void DeleteSequenceTuple(Oid relid); extern void ResetSequence(Oid seq_relid); +extern void ResetSequence2(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called); extern void ResetSequenceCaches(void); extern void seq_redo(XLogReaderState *rptr); diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 37fcc4c9b5a..4a990364e4a 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3656,6 +3656,10 @@ typedef enum PublicationObjSpecType PUBLICATIONOBJ_TABLES_IN_SCHEMA, /* All tables in schema */ PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA, /* All tables in first element of * search_path */ + PUBLICATIONOBJ_SEQUENCE, /* Sequence type */ + PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA, /* Sequences in schema type */ + PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA, /* Get the first element of + * search_path */ PUBLICATIONOBJ_CONTINUATION /* Continuation of previous type */ } PublicationObjSpecType; @@ -3675,6 +3679,7 @@ typedef struct CreatePublicationStmt List *options; /* List of DefElem nodes */ List *pubobjects; /* Optional list of publication objects */ bool for_all_tables; /* Special publication for all tables in db */ + bool for_all_sequences; /* Special publication for all sequences in db */ } CreatePublicationStmt; typedef enum AlterPublicationAction @@ -3698,6 +3703,7 @@ typedef struct AlterPublicationStmt */ List *pubobjects; /* Optional list of publication objects */ bool for_all_tables; /* Special publication for all tables in db */ + bool for_all_sequences; /* Special publication for all sequences in db */ AlterPublicationAction action; /* What action to perform with the given * objects */ } AlterPublicationStmt; diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 22fffaca62d..8f8c325522d 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -60,6 +60,7 @@ typedef enum LogicalRepMsgType LOGICAL_REP_MSG_RELATION = 'R', LOGICAL_REP_MSG_TYPE = 'Y', LOGICAL_REP_MSG_MESSAGE = 'M', + LOGICAL_REP_MSG_SEQUENCE = 'X', /* FIXME change */ LOGICAL_REP_MSG_BEGIN_PREPARE = 'b', LOGICAL_REP_MSG_PREPARE = 'P', LOGICAL_REP_MSG_COMMIT_PREPARED = 'K', @@ -117,6 +118,18 @@ typedef struct LogicalRepTyp char *typname; /* name of the remote type */ } LogicalRepTyp; +/* Sequence info */ +typedef struct LogicalRepSequence +{ + Oid remoteid; /* unique id of the remote sequence */ + char *nspname; /* schema name of remote sequence */ + char *seqname; /* name of the remote sequence */ + bool transactional; + int64 last_value; + int64 log_cnt; + bool is_called; +} LogicalRepSequence; + /* Transaction info */ typedef struct LogicalRepBeginData { @@ -227,6 +240,12 @@ extern List *logicalrep_read_truncate(StringInfo in, bool *cascade, bool *restart_seqs); extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, const char *message); +extern void logicalrep_write_sequence(StringInfo out, Relation rel, + TransactionId xid, XLogRecPtr lsn, + bool transactional, + int64 last_value, int64 log_cnt, + bool is_called); +extern void logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata); extern void logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel); extern LogicalRepRelation *logicalrep_read_rel(StringInfo in); diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h index 78aa9151ef5..a6f6843ada6 100644 --- a/src/include/replication/pgoutput.h +++ b/src/include/replication/pgoutput.h @@ -28,6 +28,7 @@ typedef struct PGOutputData bool streaming; bool messages; bool two_phase; + bool sequences; } PGOutputData; #endif /* PGOUTPUT_H */ diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 1420288d67b..3f50e100f8d 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1429,6 +1429,14 @@ pg_prepared_xacts| SELECT p.transaction, FROM ((pg_prepared_xact() p(transaction, gid, prepared, ownerid, dbid) LEFT JOIN pg_authid u ON ((p.ownerid = u.oid))) LEFT JOIN pg_database d ON ((p.dbid = d.oid))); +pg_publication_sequences| SELECT p.pubname, + n.nspname AS schemaname, + c.relname AS sequencename + FROM pg_publication p, + LATERAL pg_get_publication_sequences((p.pubname)::text) gpt(relid), + (pg_class c + JOIN pg_namespace n ON ((n.oid = c.relnamespace))) + WHERE (c.oid = gpt.relid); pg_publication_tables| SELECT p.pubname, n.nspname AS schemaname, c.relname AS tablename diff --git a/src/test/subscription/t/028_sequences.pl b/src/test/subscription/t/028_sequences.pl new file mode 100644 index 00000000000..58775769cdc --- /dev/null +++ b/src/test/subscription/t/028_sequences.pl @@ -0,0 +1,196 @@ + +# Copyright (c) 2021, PostgreSQL Global Development Group + +# This tests that sequences are replicated correctly by logical replication +use strict; +use warnings; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More tests => 6; + +# Initialize publisher node +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +# Create subscriber node +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init(allows_streaming => 'logical'); +$node_subscriber->start; + +# Create some preexisting content on publisher +my $ddl = qq( + CREATE SEQUENCE s; +); + +# Setup structure on the publisher +$node_publisher->safe_psql('postgres', $ddl); + +# Create some the same structure on subscriber, and an extra sequence that +# we'll create on the publisher later +$ddl = qq( + CREATE SEQUENCE s; + CREATE SEQUENCE s2; +); + +$node_subscriber->safe_psql('postgres', $ddl); + +# Setup logical replication +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION seq_pub"); + +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION seq_pub ADD SEQUENCE s"); + +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION seq_sub CONNECTION '$publisher_connstr' PUBLICATION seq_pub WITH (slot_name = seq_sub_slot)" +); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Wait for initial sync to finish as well +my $synced_query = + "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');"; +$node_subscriber->poll_query_until('postgres', $synced_query) + or die "Timed out while waiting for subscriber to synchronize data"; + +# Insert initial test data +$node_publisher->safe_psql( + 'postgres', qq( + -- generate a number of values using the sequence + SELECT nextval('s') FROM generate_series(1,100); +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +my $result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s; +)); + +is( $result, '132|0|t', + 'check replicated sequence values on subscriber'); + + +# advance the sequence in a rolled-back transaction - should not be replicated +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + SELECT nextval('s') FROM generate_series(1,100); + ROLLBACK; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s; +)); + +is( $result, '231|0|t', + 'check replicated sequence values on subscriber'); + + +# create a new sequence and roll it back - should not be replicated, due to +# the transactional behavior +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + CREATE SEQUENCE s2; + ALTER PUBLICATION seq_pub ADD SEQUENCE s2; + SELECT nextval('s2') FROM generate_series(1,100); + ROLLBACK; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s2; +)); + +is( $result, '1|0|f', + 'check replicated sequence values on subscriber'); + + +# create a new sequence, advance it in a rolled-back transaction, but commit +# the create - the advance should be replicated nevertheless +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + CREATE SEQUENCE s2; + ALTER PUBLICATION seq_pub ADD SEQUENCE s2; + SAVEPOINT sp1; + SELECT nextval('s2') FROM generate_series(1,100); + ROLLBACK TO sp1; + COMMIT; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Wait for sync of the second sequence we just added to finish +$synced_query = + "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');"; +$node_subscriber->poll_query_until('postgres', $synced_query) + or die "Timed out while waiting for subscriber to synchronize data"; + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s2; +)); + +is( $result, '132|0|t', + 'check replicated sequence values on subscriber'); + + +# advance the new sequence in a transaction, and roll it back - in this case +# it should not be replicated at commit +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + SELECT nextval('s2') FROM generate_series(1,100); + ROLLBACK; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s2; +)); + +is( $result, '231|0|t', + 'check replicated sequence values on subscriber'); + + +# advance the sequence in a subtransaction - the subtransaction gets rolled +# back, but commit the main one - the changes should still be replicated +$node_publisher->safe_psql( + 'postgres', qq( + BEGIN; + SAVEPOINT s1; + SELECT nextval('s2') FROM generate_series(1,100); + ROLLBACK TO s1; + COMMIT; +)); + +$node_publisher->wait_for_catchup('seq_sub'); + +# Check the data on subscriber +$result = $node_subscriber->safe_psql( + 'postgres', qq( + SELECT * FROM s2; +)); + +is( $result, '330|0|t', + 'check replicated sequence values on subscriber'); + + +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); -- 2.34.1 [text/x-patch] 0002-tweak-test-20220212.patch (2.6K, ../../[email protected]/3-0002-tweak-test-20220212.patch) download | inline diff: From aa91b8e9a469a4fa13a8b185dfda98e45ee4b8c3 Mon Sep 17 00:00:00 2001 From: Tomas Vondra <[email protected]> Date: Sat, 12 Feb 2022 01:24:47 +0100 Subject: [PATCH 2/2] tweak test --- src/test/subscription/t/028_sequences.pl | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/src/test/subscription/t/028_sequences.pl b/src/test/subscription/t/028_sequences.pl index 58775769cdc..58ed02462d8 100644 --- a/src/test/subscription/t/028_sequences.pl +++ b/src/test/subscription/t/028_sequences.pl @@ -20,6 +20,7 @@ $node_subscriber->start; # Create some preexisting content on publisher my $ddl = qq( + CREATE TABLE seq_test (v BIGINT); CREATE SEQUENCE s; ); @@ -29,6 +30,7 @@ $node_publisher->safe_psql('postgres', $ddl); # Create some the same structure on subscriber, and an extra sequence that # we'll create on the publisher later $ddl = qq( + CREATE TABLE seq_test (v BIGINT); CREATE SEQUENCE s; CREATE SEQUENCE s2; ); @@ -59,7 +61,7 @@ $node_subscriber->poll_query_until('postgres', $synced_query) $node_publisher->safe_psql( 'postgres', qq( -- generate a number of values using the sequence - SELECT nextval('s') FROM generate_series(1,100); + INSERT INTO seq_test SELECT nextval('s') FROM generate_series(1,100); )); $node_publisher->wait_for_catchup('seq_sub'); @@ -78,7 +80,7 @@ is( $result, '132|0|t', $node_publisher->safe_psql( 'postgres', qq( BEGIN; - SELECT nextval('s') FROM generate_series(1,100); + INSERT INTO seq_test SELECT nextval('s') FROM generate_series(1,100); ROLLBACK; )); @@ -101,7 +103,7 @@ $node_publisher->safe_psql( BEGIN; CREATE SEQUENCE s2; ALTER PUBLICATION seq_pub ADD SEQUENCE s2; - SELECT nextval('s2') FROM generate_series(1,100); + INSERT INTO seq_test SELECT nextval('s2') FROM generate_series(1,100); ROLLBACK; )); @@ -125,7 +127,7 @@ $node_publisher->safe_psql( CREATE SEQUENCE s2; ALTER PUBLICATION seq_pub ADD SEQUENCE s2; SAVEPOINT sp1; - SELECT nextval('s2') FROM generate_series(1,100); + INSERT INTO seq_test SELECT nextval('s2') FROM generate_series(1,100); ROLLBACK TO sp1; COMMIT; )); @@ -153,7 +155,7 @@ is( $result, '132|0|t', $node_publisher->safe_psql( 'postgres', qq( BEGIN; - SELECT nextval('s2') FROM generate_series(1,100); + INSERT INTO seq_test SELECT nextval('s2') FROM generate_series(1,100); ROLLBACK; )); @@ -175,7 +177,7 @@ $node_publisher->safe_psql( 'postgres', qq( BEGIN; SAVEPOINT s1; - SELECT nextval('s2') FROM generate_series(1,100); + INSERT INTO seq_test SELECT nextval('s2') FROM generate_series(1,100); ROLLBACK TO s1; COMMIT; )); -- 2.34.1 ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-02-28 11:46 Amit Kapila <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 2 replies; 17+ messages in thread From: Amit Kapila @ 2022-02-28 11:46 UTC (permalink / raw) To: Tomas Vondra <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra <[email protected]> wrote: > > On 2/10/22 19:17, Tomas Vondra wrote: > > I've polished & pushed the first part adding sequence decoding > > infrastructure etc. Attached are the two remaining parts. > > > > I plan to wait a day or two and then push the test_decoding part. The > > last part (for built-in replication) will need more work and maybe > > rethinking the grammar etc. > > > > I've pushed the second part, adding sequences to test_decoding. > The test_decoding is failing randomly in the last few days. I am not completely sure but they might be related to this work. The two of these appears to be due to the same reason: https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: "reorderbuffer.c", Line: 1173, PID: 35013) 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 Another: https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=mandrill&dt=2022-02-16%2006%3A21%3A48 --- /home/nm/farm/xlc32/HEAD/pgsql.build/contrib/test_decoding/expected/rewrite.out 2022-02-14 20:19:14.000000000 +0000 +++ /home/nm/farm/xlc32/HEAD/pgsql.build/contrib/test_decoding/results/rewrite.out 2022-02-16 07:42:18.000000000 +0000 @@ -126,6 +126,7 @@ table public.replication_example: INSERT: id[integer]:4 somedata[integer]:3 text[character varying]:null testcolumn1[integer]:null table public.replication_example: INSERT: id[integer]:5 somedata[integer]:4 text[character varying]:null testcolumn1[integer]:2 testcolumn2[integer]:1 COMMIT + sequence public.replication_example_id_seq: transactional:0 last_value: 38 log_cnt: 0 is_called:1 BEGIN table public.replication_example: INSERT: id[integer]:6 somedata[integer]:5 text[character varying]:null testcolumn1[integer]:3 testcolumn2[integer]:null COMMIT @@ -133,7 +134,7 @@ table public.replication_example: INSERT: id[integer]:7 somedata[integer]:6 text[character varying]:null testcolumn1[integer]:4 testcolumn2[integer]:null table public.replication_example: INSERT: id[integer]:8 somedata[integer]:7 text[character varying]:null testcolumn1[integer]:5 testcolumn2[integer]:null testcolumn3[integer]:1 COMMIT - (15 rows) + (16 rows) -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-01 11:53 Amit Kapila <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 2 replies; 17+ messages in thread From: Amit Kapila @ 2022-03-01 11:53 UTC (permalink / raw) To: Tomas Vondra <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On Mon, Feb 28, 2022 at 5:16 PM Amit Kapila <[email protected]> wrote: > > On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra > <[email protected]> wrote: > > > > On 2/10/22 19:17, Tomas Vondra wrote: > > > I've polished & pushed the first part adding sequence decoding > > > infrastructure etc. Attached are the two remaining parts. > > > > > > I plan to wait a day or two and then push the test_decoding part. The > > > last part (for built-in replication) will need more work and maybe > > > rethinking the grammar etc. > > > > > > > I've pushed the second part, adding sequences to test_decoding. > > > > The test_decoding is failing randomly in the last few days. I am not > completely sure but they might be related to this work. The two of > these appears to be due to the same reason: > https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 > https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 > > TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: > "reorderbuffer.c", Line: 1173, PID: 35013) > 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 > While reviewing the code for this, I noticed that in sequence_decode(), we don't call ReorderBufferProcessXid to register the first known lsn in WAL for the current xid. The similar functions logicalmsg_decode() or heap_decode() do call ReorderBufferProcessXid even if they decide not to queue or send the change. Is there a reason for not doing the same here? However, I am not able to deduce any scenario where lack of this will lead to such an Assertion failure. Any thoughts? -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-07 02:13 Peter Smith <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 0 replies; 17+ messages in thread From: Peter Smith @ 2022-03-07 02:13 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Tomas Vondra <[email protected]>; Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On Tue, Mar 1, 2022 at 10:54 PM Amit Kapila <[email protected]> wrote: > > On Mon, Feb 28, 2022 at 5:16 PM Amit Kapila <[email protected]> wrote: > > > > On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra > > <[email protected]> wrote: > > > > > > On 2/10/22 19:17, Tomas Vondra wrote: > > > > I've polished & pushed the first part adding sequence decoding > > > > infrastructure etc. Attached are the two remaining parts. > > > > > > > > I plan to wait a day or two and then push the test_decoding part. The > > > > last part (for built-in replication) will need more work and maybe > > > > rethinking the grammar etc. > > > > > > > > > > I've pushed the second part, adding sequences to test_decoding. > > > > > > > The test_decoding is failing randomly in the last few days. I am not > > completely sure but they might be related to this work. The two of > > these appears to be due to the same reason: > > https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 > > https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 > > > > TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: > > "reorderbuffer.c", Line: 1173, PID: 35013) > > 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 > > FYI, it looks like the same assertion has failed again on the same build-farm machine [1] > > While reviewing the code for this, I noticed that in > sequence_decode(), we don't call ReorderBufferProcessXid to register > the first known lsn in WAL for the current xid. The similar functions > logicalmsg_decode() or heap_decode() do call ReorderBufferProcessXid > even if they decide not to queue or send the change. Is there a reason > for not doing the same here? However, I am not able to deduce any > scenario where lack of this will lead to such an Assertion failure. > Any thoughts? > ------ [1] https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-03-03%2023%3A14%3A26 Kind Regards, Peter Smith. Fujitsu Australia ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-07 16:39 Tomas Vondra <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 1 reply; 17+ messages in thread From: Tomas Vondra @ 2022-03-07 16:39 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On 3/1/22 12:53, Amit Kapila wrote: > On Mon, Feb 28, 2022 at 5:16 PM Amit Kapila <[email protected]> wrote: >> >> On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra >> <[email protected]> wrote: >>> >>> On 2/10/22 19:17, Tomas Vondra wrote: >>>> I've polished & pushed the first part adding sequence decoding >>>> infrastructure etc. Attached are the two remaining parts. >>>> >>>> I plan to wait a day or two and then push the test_decoding part. The >>>> last part (for built-in replication) will need more work and maybe >>>> rethinking the grammar etc. >>>> >>> >>> I've pushed the second part, adding sequences to test_decoding. >>> >> >> The test_decoding is failing randomly in the last few days. I am not >> completely sure but they might be related to this work. The two of >> these appears to be due to the same reason: >> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 >> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 >> >> TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: >> "reorderbuffer.c", Line: 1173, PID: 35013) >> 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 >> > > While reviewing the code for this, I noticed that in > sequence_decode(), we don't call ReorderBufferProcessXid to register > the first known lsn in WAL for the current xid. The similar functions > logicalmsg_decode() or heap_decode() do call ReorderBufferProcessXid > even if they decide not to queue or send the change. Is there a reason > for not doing the same here? However, I am not able to deduce any > scenario where lack of this will lead to such an Assertion failure. > Any thoughts? > Thanks, that seems like an omission. Will fix. regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-07 16:53 Tomas Vondra <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 1 reply; 17+ messages in thread From: Tomas Vondra @ 2022-03-07 16:53 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On 2/28/22 12:46, Amit Kapila wrote: > On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra > <[email protected]> wrote: >> >> On 2/10/22 19:17, Tomas Vondra wrote: >>> I've polished & pushed the first part adding sequence decoding >>> infrastructure etc. Attached are the two remaining parts. >>> >>> I plan to wait a day or two and then push the test_decoding part. The >>> last part (for built-in replication) will need more work and maybe >>> rethinking the grammar etc. >>> >> >> I've pushed the second part, adding sequences to test_decoding. >> > > The test_decoding is failing randomly in the last few days. I am not > completely sure but they might be related to this work. The two of > these appears to be due to the same reason: > https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 > https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 > > TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: > "reorderbuffer.c", Line: 1173, PID: 35013) > 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 > This might be related to the issue reported by Amit, i.e. that sequence_decode does not call ReorderBufferProcessXid(). If this keeps failing, we'll have to add some extra debug info (logging LSN etc.), at least temporarily. It'd be valuable to inspect the WAL too. > Another: > https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=mandrill&dt=2022-02-16%2006%3A21%3A48 > > --- /home/nm/farm/xlc32/HEAD/pgsql.build/contrib/test_decoding/expected/rewrite.out > 2022-02-14 20:19:14.000000000 +0000 > +++ /home/nm/farm/xlc32/HEAD/pgsql.build/contrib/test_decoding/results/rewrite.out > 2022-02-16 07:42:18.000000000 +0000 > @@ -126,6 +126,7 @@ > table public.replication_example: INSERT: id[integer]:4 > somedata[integer]:3 text[character varying]:null > testcolumn1[integer]:null > table public.replication_example: INSERT: id[integer]:5 > somedata[integer]:4 text[character varying]:null > testcolumn1[integer]:2 testcolumn2[integer]:1 > COMMIT > + sequence public.replication_example_id_seq: transactional:0 > last_value: 38 log_cnt: 0 is_called:1 > BEGIN > table public.replication_example: INSERT: id[integer]:6 > somedata[integer]:5 text[character varying]:null > testcolumn1[integer]:3 testcolumn2[integer]:null > COMMIT > @@ -133,7 +134,7 @@ > table public.replication_example: INSERT: id[integer]:7 > somedata[integer]:6 text[character varying]:null > testcolumn1[integer]:4 testcolumn2[integer]:null > table public.replication_example: INSERT: id[integer]:8 > somedata[integer]:7 text[character varying]:null > testcolumn1[integer]:5 testcolumn2[integer]:null > testcolumn3[integer]:1 > COMMIT > - (15 rows) > + (16 rows) > Interesting. I can think of one reason that might cause this - we log the first sequence increment after a checkpoint. So if a checkpoint happens in an unfortunate place, there'll be an extra WAL record. On slow / busy machines that's quite possible, I guess. I wonder if these two issues might be related ... regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-07 21:11 Tomas Vondra <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 1 reply; 17+ messages in thread From: Tomas Vondra @ 2022-03-07 21:11 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On 3/7/22 17:39, Tomas Vondra wrote: > > > On 3/1/22 12:53, Amit Kapila wrote: >> On Mon, Feb 28, 2022 at 5:16 PM Amit Kapila <[email protected]> wrote: >>> >>> On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra >>> <[email protected]> wrote: >>>> >>>> On 2/10/22 19:17, Tomas Vondra wrote: >>>>> I've polished & pushed the first part adding sequence decoding >>>>> infrastructure etc. Attached are the two remaining parts. >>>>> >>>>> I plan to wait a day or two and then push the test_decoding part. The >>>>> last part (for built-in replication) will need more work and maybe >>>>> rethinking the grammar etc. >>>>> >>>> >>>> I've pushed the second part, adding sequences to test_decoding. >>>> >>> >>> The test_decoding is failing randomly in the last few days. I am not >>> completely sure but they might be related to this work. The two of >>> these appears to be due to the same reason: >>> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 >>> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 >>> >>> TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: >>> "reorderbuffer.c", Line: 1173, PID: 35013) >>> 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 >>> >> >> While reviewing the code for this, I noticed that in >> sequence_decode(), we don't call ReorderBufferProcessXid to register >> the first known lsn in WAL for the current xid. The similar functions >> logicalmsg_decode() or heap_decode() do call ReorderBufferProcessXid >> even if they decide not to queue or send the change. Is there a reason >> for not doing the same here? However, I am not able to deduce any >> scenario where lack of this will lead to such an Assertion failure. >> Any thoughts? >> > > Thanks, that seems like an omission. Will fix. > I've pushed this simple fix. Not sure it'll fix the assert failures on skink/locust, though. Given the lack of information it'll be difficult to verify. So let's wait a bit. regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-07 21:25 Tomas Vondra <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 1 reply; 17+ messages in thread From: Tomas Vondra @ 2022-03-07 21:25 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On 3/7/22 17:53, Tomas Vondra wrote: > On 2/28/22 12:46, Amit Kapila wrote: >> On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra >> <[email protected]> wrote: >>> >>> On 2/10/22 19:17, Tomas Vondra wrote: >>>> I've polished & pushed the first part adding sequence decoding >>>> infrastructure etc. Attached are the two remaining parts. >>>> >>>> I plan to wait a day or two and then push the test_decoding part. The >>>> last part (for built-in replication) will need more work and maybe >>>> rethinking the grammar etc. >>>> >>> >>> I've pushed the second part, adding sequences to test_decoding. >>> >> >> The test_decoding is failing randomly in the last few days. I am not >> completely sure but they might be related to this work. The two of >> these appears to be due to the same reason: >> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 >> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 >> >> TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: >> "reorderbuffer.c", Line: 1173, PID: 35013) >> 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 >> > > This might be related to the issue reported by Amit, i.e. that > sequence_decode does not call ReorderBufferProcessXid(). If this keeps > failing, we'll have to add some extra debug info (logging LSN etc.), at > least temporarily. It'd be valuable to inspect the WAL too. > >> Another: >> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=mandrill&dt=2022-02-16%2006%3A21%3A48 >> >> --- /home/nm/farm/xlc32/HEAD/pgsql.build/contrib/test_decoding/expected/rewrite.out >> 2022-02-14 20:19:14.000000000 +0000 >> +++ /home/nm/farm/xlc32/HEAD/pgsql.build/contrib/test_decoding/results/rewrite.out >> 2022-02-16 07:42:18.000000000 +0000 >> @@ -126,6 +126,7 @@ >> table public.replication_example: INSERT: id[integer]:4 >> somedata[integer]:3 text[character varying]:null >> testcolumn1[integer]:null >> table public.replication_example: INSERT: id[integer]:5 >> somedata[integer]:4 text[character varying]:null >> testcolumn1[integer]:2 testcolumn2[integer]:1 >> COMMIT >> + sequence public.replication_example_id_seq: transactional:0 >> last_value: 38 log_cnt: 0 is_called:1 >> BEGIN >> table public.replication_example: INSERT: id[integer]:6 >> somedata[integer]:5 text[character varying]:null >> testcolumn1[integer]:3 testcolumn2[integer]:null >> COMMIT >> @@ -133,7 +134,7 @@ >> table public.replication_example: INSERT: id[integer]:7 >> somedata[integer]:6 text[character varying]:null >> testcolumn1[integer]:4 testcolumn2[integer]:null >> table public.replication_example: INSERT: id[integer]:8 >> somedata[integer]:7 text[character varying]:null >> testcolumn1[integer]:5 testcolumn2[integer]:null >> testcolumn3[integer]:1 >> COMMIT >> - (15 rows) >> + (16 rows) >> > > Interesting. I can think of one reason that might cause this - we log > the first sequence increment after a checkpoint. So if a checkpoint > happens in an unfortunate place, there'll be an extra WAL record. On > slow / busy machines that's quite possible, I guess. > I've tweaked the checkpoint_interval to make checkpoints more aggressive (set it to 1s), and it seems my hunch was correct - it produces failures exactly like this one. The best fix probably is to just disable decoding of sequences in those tests that are not aimed at testing sequence decoding. regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-08 18:29 Tomas Vondra <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 1 reply; 17+ messages in thread From: Tomas Vondra @ 2022-03-08 18:29 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On 3/7/22 22:25, Tomas Vondra wrote: > > > On 3/7/22 17:53, Tomas Vondra wrote: >> On 2/28/22 12:46, Amit Kapila wrote: >>> On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra >>> <[email protected]> wrote: >>>> >>>> On 2/10/22 19:17, Tomas Vondra wrote: >>>>> I've polished & pushed the first part adding sequence decoding >>>>> infrastructure etc. Attached are the two remaining parts. >>>>> >>>>> I plan to wait a day or two and then push the test_decoding part. The >>>>> last part (for built-in replication) will need more work and maybe >>>>> rethinking the grammar etc. >>>>> >>>> >>>> I've pushed the second part, adding sequences to test_decoding. >>>> >>> >>> The test_decoding is failing randomly in the last few days. I am not >>> completely sure but they might be related to this work. The two of >>> these appears to be due to the same reason: >>> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 >>> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 >>> >>> TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: >>> "reorderbuffer.c", Line: 1173, PID: 35013) >>> 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 >>> >> >> This might be related to the issue reported by Amit, i.e. that >> sequence_decode does not call ReorderBufferProcessXid(). If this keeps >> failing, we'll have to add some extra debug info (logging LSN etc.), at >> least temporarily. It'd be valuable to inspect the WAL too. >> >>> Another: >>> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=mandrill&dt=2022-02-16%2006%3A21%3A48 >>> >>> --- /home/nm/farm/xlc32/HEAD/pgsql.build/contrib/test_decoding/expected/rewrite.out >>> 2022-02-14 20:19:14.000000000 +0000 >>> +++ /home/nm/farm/xlc32/HEAD/pgsql.build/contrib/test_decoding/results/rewrite.out >>> 2022-02-16 07:42:18.000000000 +0000 >>> @@ -126,6 +126,7 @@ >>> table public.replication_example: INSERT: id[integer]:4 >>> somedata[integer]:3 text[character varying]:null >>> testcolumn1[integer]:null >>> table public.replication_example: INSERT: id[integer]:5 >>> somedata[integer]:4 text[character varying]:null >>> testcolumn1[integer]:2 testcolumn2[integer]:1 >>> COMMIT >>> + sequence public.replication_example_id_seq: transactional:0 >>> last_value: 38 log_cnt: 0 is_called:1 >>> BEGIN >>> table public.replication_example: INSERT: id[integer]:6 >>> somedata[integer]:5 text[character varying]:null >>> testcolumn1[integer]:3 testcolumn2[integer]:null >>> COMMIT >>> @@ -133,7 +134,7 @@ >>> table public.replication_example: INSERT: id[integer]:7 >>> somedata[integer]:6 text[character varying]:null >>> testcolumn1[integer]:4 testcolumn2[integer]:null >>> table public.replication_example: INSERT: id[integer]:8 >>> somedata[integer]:7 text[character varying]:null >>> testcolumn1[integer]:5 testcolumn2[integer]:null >>> testcolumn3[integer]:1 >>> COMMIT >>> - (15 rows) >>> + (16 rows) >>> >> >> Interesting. I can think of one reason that might cause this - we log >> the first sequence increment after a checkpoint. So if a checkpoint >> happens in an unfortunate place, there'll be an extra WAL record. On >> slow / busy machines that's quite possible, I guess. >> > > I've tweaked the checkpoint_interval to make checkpoints more aggressive > (set it to 1s), and it seems my hunch was correct - it produces failures > exactly like this one. The best fix probably is to just disable decoding > of sequences in those tests that are not aimed at testing sequence decoding. > I've pushed a fix for this, adding "include-sequences=0" to a couple test_decoding tests, which were failing with concurrent checkpoints. Unfortunately, I realized we have a similar issue in the "sequences" tests too :-( Imagine you do a series of sequence increments, e.g. SELECT nextval('s') FROM generate_sequences(1,100); If there's a concurrent checkpoint, this may add an extra WAL record, affecting the decoded output (and also the data stored in the sequence relation itself). Not sure what to do about this ... regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-08 22:44 Tomas Vondra <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 1 reply; 17+ messages in thread From: Tomas Vondra @ 2022-03-08 22:44 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On 3/7/22 22:11, Tomas Vondra wrote: > > > On 3/7/22 17:39, Tomas Vondra wrote: >> >> >> On 3/1/22 12:53, Amit Kapila wrote: >>> On Mon, Feb 28, 2022 at 5:16 PM Amit Kapila <[email protected]> wrote: >>>> >>>> On Sat, Feb 12, 2022 at 6:04 AM Tomas Vondra >>>> <[email protected]> wrote: >>>>> >>>>> On 2/10/22 19:17, Tomas Vondra wrote: >>>>>> I've polished & pushed the first part adding sequence decoding >>>>>> infrastructure etc. Attached are the two remaining parts. >>>>>> >>>>>> I plan to wait a day or two and then push the test_decoding part. The >>>>>> last part (for built-in replication) will need more work and maybe >>>>>> rethinking the grammar etc. >>>>>> >>>>> >>>>> I've pushed the second part, adding sequences to test_decoding. >>>>> >>>> >>>> The test_decoding is failing randomly in the last few days. I am not >>>> completely sure but they might be related to this work. The two of >>>> these appears to be due to the same reason: >>>> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2022-02-25%2018%3A50%3A09 >>>> https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=locust&dt=2022-02-17%2015%3A17%3A07 >>>> >>>> TRAP: FailedAssertion("prev_first_lsn < cur_txn->first_lsn", File: >>>> "reorderbuffer.c", Line: 1173, PID: 35013) >>>> 0 postgres 0x00593de0 ExceptionalCondition + 160\\0 >>>> >>> >>> While reviewing the code for this, I noticed that in >>> sequence_decode(), we don't call ReorderBufferProcessXid to register >>> the first known lsn in WAL for the current xid. The similar functions >>> logicalmsg_decode() or heap_decode() do call ReorderBufferProcessXid >>> even if they decide not to queue or send the change. Is there a reason >>> for not doing the same here? However, I am not able to deduce any >>> scenario where lack of this will lead to such an Assertion failure. >>> Any thoughts? >>> >> >> Thanks, that seems like an omission. Will fix. >> > > I've pushed this simple fix. Not sure it'll fix the assert failures on > skink/locust, though. Given the lack of information it'll be difficult > to verify. So let's wait a bit. > I've done about 5000 runs of 'make check' in test_decoding, on two rpi machines (one armv7, one aarch64). Not a single assert failure :-( How come skink/locust hit that in just a couple runs? regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-09 11:41 Amit Kapila <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 1 reply; 17+ messages in thread From: Amit Kapila @ 2022-03-09 11:41 UTC (permalink / raw) To: Tomas Vondra <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On Wed, Mar 9, 2022 at 4:14 AM Tomas Vondra <[email protected]> wrote: > > On 3/7/22 22:11, Tomas Vondra wrote: > > > > I've pushed this simple fix. Not sure it'll fix the assert failures on > > skink/locust, though. Given the lack of information it'll be difficult > > to verify. So let's wait a bit. > > > > I've done about 5000 runs of 'make check' in test_decoding, on two rpi > machines (one armv7, one aarch64). Not a single assert failure :-( > > How come skink/locust hit that in just a couple runs? > Is it failed after you pushed a fix? I don't think so or am I missing something? I feel even if doesn't occur again it would have been better if we had some theory on how it occurred in the first place because that would make us feel more confident that we won't have any related problem left. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-09 13:18 Tomas Vondra <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 0 replies; 17+ messages in thread From: Tomas Vondra @ 2022-03-09 13:18 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On 3/9/22 12:41, Amit Kapila wrote: > On Wed, Mar 9, 2022 at 4:14 AM Tomas Vondra > <[email protected]> wrote: >> >> On 3/7/22 22:11, Tomas Vondra wrote: >>> >>> I've pushed this simple fix. Not sure it'll fix the assert failures on >>> skink/locust, though. Given the lack of information it'll be difficult >>> to verify. So let's wait a bit. >>> >> >> I've done about 5000 runs of 'make check' in test_decoding, on two rpi >> machines (one armv7, one aarch64). Not a single assert failure :-( >> >> How come skink/locust hit that in just a couple runs? >> > > Is it failed after you pushed a fix? I don't think so or am I missing > something? I feel even if doesn't occur again it would have been > better if we had some theory on how it occurred in the first place > because that would make us feel more confident that we won't have any > related problem left. > I don't think it failed yet - we have to wait a bit longer to make any conclusions, though. On skink it failed only twice over 1 month. I agree it'd be nice to have some theory, but I really don't have one. regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-11 11:34 Amit Kapila <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 2 replies; 17+ messages in thread From: Amit Kapila @ 2022-03-11 11:34 UTC (permalink / raw) To: Tomas Vondra <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On Tue, Mar 8, 2022 at 11:59 PM Tomas Vondra <[email protected]> wrote: > > On 3/7/22 22:25, Tomas Vondra wrote: > >> > >> Interesting. I can think of one reason that might cause this - we log > >> the first sequence increment after a checkpoint. So if a checkpoint > >> happens in an unfortunate place, there'll be an extra WAL record. On > >> slow / busy machines that's quite possible, I guess. > >> > > > > I've tweaked the checkpoint_interval to make checkpoints more aggressive > > (set it to 1s), and it seems my hunch was correct - it produces failures > > exactly like this one. The best fix probably is to just disable decoding > > of sequences in those tests that are not aimed at testing sequence decoding. > > > > I've pushed a fix for this, adding "include-sequences=0" to a couple > test_decoding tests, which were failing with concurrent checkpoints. > > Unfortunately, I realized we have a similar issue in the "sequences" > tests too :-( Imagine you do a series of sequence increments, e.g. > > SELECT nextval('s') FROM generate_sequences(1,100); > > If there's a concurrent checkpoint, this may add an extra WAL record, > affecting the decoded output (and also the data stored in the sequence > relation itself). Not sure what to do about this ... > I am also not sure what to do for it but maybe if in some way we can increase checkpoint timeout or other parameters for these tests then it would reduce the chances of such failures. The other idea could be to perform checkpoint before the start of tests to reduce the possibility of another checkpoint. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-11 11:38 Amit Kapila <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 0 replies; 17+ messages in thread From: Amit Kapila @ 2022-03-11 11:38 UTC (permalink / raw) To: Tomas Vondra <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On Fri, Mar 11, 2022 at 5:04 PM Amit Kapila <[email protected]> wrote: > > On Tue, Mar 8, 2022 at 11:59 PM Tomas Vondra > <[email protected]> wrote: > > > > On 3/7/22 22:25, Tomas Vondra wrote: > > >> > > >> Interesting. I can think of one reason that might cause this - we log > > >> the first sequence increment after a checkpoint. So if a checkpoint > > >> happens in an unfortunate place, there'll be an extra WAL record. On > > >> slow / busy machines that's quite possible, I guess. > > >> > > > > > > I've tweaked the checkpoint_interval to make checkpoints more aggressive > > > (set it to 1s), and it seems my hunch was correct - it produces failures > > > exactly like this one. The best fix probably is to just disable decoding > > > of sequences in those tests that are not aimed at testing sequence decoding. > > > > > > > I've pushed a fix for this, adding "include-sequences=0" to a couple > > test_decoding tests, which were failing with concurrent checkpoints. > > > > Unfortunately, I realized we have a similar issue in the "sequences" > > tests too :-( Imagine you do a series of sequence increments, e.g. > > > > SELECT nextval('s') FROM generate_sequences(1,100); > > > > If there's a concurrent checkpoint, this may add an extra WAL record, > > affecting the decoded output (and also the data stored in the sequence > > relation itself). Not sure what to do about this ... > > > > I am also not sure what to do for it but maybe if in some way we can > increase checkpoint timeout or other parameters for these tests then > it would reduce the chances of such failures. The other idea could be > to perform checkpoint before the start of tests to reduce the > possibility of another checkpoint. > One more thing, I notice while checking the commit for this feature is that the below include seems to be out of order: --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -42,6 +42,7 @@ #include "replication/reorderbuffer.h" #include "replication/snapbuild.h" #include "storage/standby.h" +#include "commands/sequence.h" -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 17+ messages in thread
* Re: logical decoding and replication of sequences @ 2022-03-11 12:53 Tomas Vondra <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 0 replies; 17+ messages in thread From: Tomas Vondra @ 2022-03-11 12:53 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Peter Eisentraut <[email protected]>; Petr Jelinek <[email protected]>; PostgreSQL Hackers <[email protected]> On 3/11/22 12:34, Amit Kapila wrote: > On Tue, Mar 8, 2022 at 11:59 PM Tomas Vondra > <[email protected]> wrote: >> >> On 3/7/22 22:25, Tomas Vondra wrote: >>>> >>>> Interesting. I can think of one reason that might cause this - we log >>>> the first sequence increment after a checkpoint. So if a checkpoint >>>> happens in an unfortunate place, there'll be an extra WAL record. On >>>> slow / busy machines that's quite possible, I guess. >>>> >>> >>> I've tweaked the checkpoint_interval to make checkpoints more aggressive >>> (set it to 1s), and it seems my hunch was correct - it produces failures >>> exactly like this one. The best fix probably is to just disable decoding >>> of sequences in those tests that are not aimed at testing sequence decoding. >>> >> >> I've pushed a fix for this, adding "include-sequences=0" to a couple >> test_decoding tests, which were failing with concurrent checkpoints. >> >> Unfortunately, I realized we have a similar issue in the "sequences" >> tests too :-( Imagine you do a series of sequence increments, e.g. >> >> SELECT nextval('s') FROM generate_sequences(1,100); >> >> If there's a concurrent checkpoint, this may add an extra WAL record, >> affecting the decoded output (and also the data stored in the sequence >> relation itself). Not sure what to do about this ... >> > > I am also not sure what to do for it but maybe if in some way we can > increase checkpoint timeout or other parameters for these tests then > it would reduce the chances of such failures. The other idea could be > to perform checkpoint before the start of tests to reduce the > possibility of another checkpoint. > Yeah, I had the same ideas, but I'm not sure I like any of them. I doubt we want to make checkpoints extremely rare, and even if we do that it'll still fail on slow machines (e.g. with valgrind, clobber cache etc.). regards -- Tomas Vondra EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company ^ permalink raw reply [nested|flat] 17+ messages in thread
end of thread, other threads:[~2022-03-11 12:53 UTC | newest] Thread overview: 17+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2019-09-10 08:28 [PATCH v8 3/4] Remove globals readSegNo, readOff, readLen Kyotaro Horiguchi <[email protected]> 2022-02-10 18:17 Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-02-12 00:34 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-02-28 11:46 ` Re: logical decoding and replication of sequences Amit Kapila <[email protected]> 2022-03-01 11:53 ` Re: logical decoding and replication of sequences Amit Kapila <[email protected]> 2022-03-07 02:13 ` Re: logical decoding and replication of sequences Peter Smith <[email protected]> 2022-03-07 16:39 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-03-07 21:11 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-03-08 22:44 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-03-09 11:41 ` Re: logical decoding and replication of sequences Amit Kapila <[email protected]> 2022-03-09 13:18 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-03-07 16:53 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-03-07 21:25 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-03-08 18:29 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]> 2022-03-11 11:34 ` Re: logical decoding and replication of sequences Amit Kapila <[email protected]> 2022-03-11 11:38 ` Re: logical decoding and replication of sequences Amit Kapila <[email protected]> 2022-03-11 12:53 ` Re: logical decoding and replication of sequences Tomas Vondra <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox