agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH 2/4] Write WAL for empty nbtree index build 14+ messages / 5 participants [nested] [flat]
* [PATCH 2/4] Write WAL for empty nbtree index build @ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw) After relation truncation indexes are also rebuild. It doesn't emit WAL in minimal mode and if truncation happened within its creation transaction, crash recovery leaves an empty index heap, which is considered broken. This patch forces to emit WAL when an index_build turns into empty nbtree index. --- src/backend/access/nbtree/nbtsort.c | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index 16f5755777..2c2647b530 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -610,8 +610,14 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno) /* Ensure rd_smgr is open (could have been closed by relcache flush!) */ RelationOpenSmgr(wstate->index); - /* XLOG stuff */ - if (wstate->btws_use_wal) + /* XLOG stuff + * + * Even if minimal mode, WAL is required here if truncation happened after + * being created in the same transaction. It is not needed otherwise but + * we don't bother identifying the case precisely. + */ + if (wstate->btws_use_wal || + (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)) { /* We use the heap NEWPAGE record type for this */ log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true); @@ -1055,6 +1061,11 @@ _bt_uppershutdown(BTWriteState *wstate, BTPageState *state) * set to point to "P_NONE"). This changes the index to the "valid" state * by filling in a valid magic number in the metapage. */ + /* + * If no tuple was inserted, it's possible that we are truncating a + * relation. We need to emit WAL for the metapage in the case. However it + * is not required elsewise, + */ metapage = (Page) palloc(BLCKSZ); _bt_initmetapage(metapage, rootblkno, rootlevel); _bt_blwritepage(wstate, metapage, BTREE_METAPAGE); -- 2.16.3 ----Next_Part(Thu_Oct_11_17_04_53_2018_748)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v3-0001-TAP-test-for-copy-truncation-optimization.patch" ^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH 2/4] Write WAL for empty nbtree index build @ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw) After relation truncation indexes are also rebuild. It doesn't emit WAL in minimal mode and if truncation happened within its creation transaction, crash recovery leaves an empty index heap, which is considered broken. This patch forces to emit WAL when an index_build turns into empty nbtree index. --- src/backend/access/nbtree/nbtsort.c | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index 16f5755777..2c2647b530 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -610,8 +610,14 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno) /* Ensure rd_smgr is open (could have been closed by relcache flush!) */ RelationOpenSmgr(wstate->index); - /* XLOG stuff */ - if (wstate->btws_use_wal) + /* XLOG stuff + * + * Even if minimal mode, WAL is required here if truncation happened after + * being created in the same transaction. It is not needed otherwise but + * we don't bother identifying the case precisely. + */ + if (wstate->btws_use_wal || + (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)) { /* We use the heap NEWPAGE record type for this */ log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true); @@ -1055,6 +1061,11 @@ _bt_uppershutdown(BTWriteState *wstate, BTPageState *state) * set to point to "P_NONE"). This changes the index to the "valid" state * by filling in a valid magic number in the metapage. */ + /* + * If no tuple was inserted, it's possible that we are truncating a + * relation. We need to emit WAL for the metapage in the case. However it + * is not required elsewise, + */ metapage = (Page) palloc(BLCKSZ); _bt_initmetapage(metapage, rootblkno, rootlevel); _bt_blwritepage(wstate, metapage, BTREE_METAPAGE); -- 2.16.3 ----Next_Part(Thu_Dec_20_17_32_25_2018_311)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v5-0003-Add-infrastructure-to-WAL-logging-skip-feature.patch" ^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH 2/4] Write WAL for empty nbtree index build @ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw) After relation truncation indexes are also rebuild. It doesn't emit WAL in minimal mode and if truncation happened within its creation transaction, crash recovery leaves an empty index heap, which is considered broken. This patch forces to emit WAL when an index_build turns into empty nbtree index. --- src/backend/access/nbtree/nbtsort.c | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index 16f5755777..2c2647b530 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -610,8 +610,14 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno) /* Ensure rd_smgr is open (could have been closed by relcache flush!) */ RelationOpenSmgr(wstate->index); - /* XLOG stuff */ - if (wstate->btws_use_wal) + /* XLOG stuff + * + * Even if minimal mode, WAL is required here if truncation happened after + * being created in the same transaction. It is not needed otherwise but + * we don't bother identifying the case precisely. + */ + if (wstate->btws_use_wal || + (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)) { /* We use the heap NEWPAGE record type for this */ log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true); @@ -1055,6 +1061,11 @@ _bt_uppershutdown(BTWriteState *wstate, BTPageState *state) * set to point to "P_NONE"). This changes the index to the "valid" state * by filling in a valid magic number in the metapage. */ + /* + * If no tuple was inserted, it's possible that we are truncating a + * relation. We need to emit WAL for the metapage in the case. However it + * is not required elsewise, + */ metapage = (Page) palloc(BLCKSZ); _bt_initmetapage(metapage, rootblkno, rootlevel); _bt_blwritepage(wstate, metapage, BTREE_METAPAGE); -- 2.16.3 ----Next_Part(Wed_Nov_14_12_47_36_2018_817)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v4-0001-TAP-test-for-copy-truncation-optimization.patch" ^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH 2/4] Write WAL for empty nbtree index build @ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw) After relation truncation indexes are also rebuild. It doesn't emit WAL in minimal mode and if truncation happened within its creation transaction, crash recovery leaves an empty index heap, which is considered broken. This patch forces to emit WAL when an index_build turns into empty nbtree index. --- src/backend/access/nbtree/nbtsort.c | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index dc398e1186..70d4380533 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -611,8 +611,14 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno) /* Ensure rd_smgr is open (could have been closed by relcache flush!) */ RelationOpenSmgr(wstate->index); - /* XLOG stuff */ - if (wstate->btws_use_wal) + /* XLOG stuff + * + * Even if minimal mode, WAL is required here if truncation happened after + * being created in the same transaction. It is not needed otherwise but + * we don't bother identifying the case precisely. + */ + if (wstate->btws_use_wal || + (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)) { /* We use the heap NEWPAGE record type for this */ log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true); @@ -1056,6 +1062,11 @@ _bt_uppershutdown(BTWriteState *wstate, BTPageState *state) * set to point to "P_NONE"). This changes the index to the "valid" state * by filling in a valid magic number in the metapage. */ + /* + * If no tuple was inserted, it's possible that we are truncating a + * relation. We need to emit WAL for the metapage in the case. However it + * is not required elsewise, + */ metapage = (Page) palloc(BLCKSZ); _bt_initmetapage(metapage, rootblkno, rootlevel); _bt_blwritepage(wstate, metapage, BTREE_METAPAGE); -- 2.16.3 ----Next_Part(Wed_Jan_30_10_26_34_2019_453)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v6-0003-Add-infrastructure-to-WAL-logging-skip-feature.patch" ^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH 3/3] Write WAL for empty nbtree index build @ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw) After relation truncation indexes are also rebuild. It doesn't emit WAL in minimal mode and if truncation happened within its creation transaction, crash recovery leaves an empty index heap, which is considered broken. This patch forces to emit WAL when an index_build turns into empty nbtree index. --- src/backend/access/nbtree/nbtsort.c | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index 16f5755777..2c2647b530 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -610,8 +610,14 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno) /* Ensure rd_smgr is open (could have been closed by relcache flush!) */ RelationOpenSmgr(wstate->index); - /* XLOG stuff */ - if (wstate->btws_use_wal) + /* XLOG stuff + * + * Even if minimal mode, WAL is required here if truncation happened after + * being created in the same transaction. It is not needed otherwise but + * we don't bother identifying the case precisely. + */ + if (wstate->btws_use_wal || + (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)) { /* We use the heap NEWPAGE record type for this */ log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true); @@ -1055,6 +1061,11 @@ _bt_uppershutdown(BTWriteState *wstate, BTPageState *state) * set to point to "P_NONE"). This changes the index to the "valid" state * by filling in a valid magic number in the metapage. */ + /* + * If no tuple was inserted, it's possible that we are truncating a + * relation. We need to emit WAL for the metapage in the case. However it + * is not required elsewise, + */ metapage = (Page) palloc(BLCKSZ); _bt_initmetapage(metapage, rootblkno, rootlevel); _bt_blwritepage(wstate, metapage, BTREE_METAPAGE); -- 2.16.3 ----Next_Part(Thu_Oct_11_13_42_35_2018_969)---- ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: wake up logical workers after ALTER SUBSCRIPTION @ 2023-01-04 05:27 Amit Kapila <[email protected]> 2023-01-04 18:12 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Amit Kapila @ 2023-01-04 05:27 UTC (permalink / raw) To: Nathan Bossart <[email protected]>; +Cc: Tom Lane <[email protected]>; Melih Mutlu <[email protected]>; Thomas Munro <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; pgsql-hackers On Tue, Jan 3, 2023 at 11:40 PM Nathan Bossart <[email protected]> wrote: > > On Tue, Jan 03, 2023 at 11:03:32AM +0530, Amit Kapila wrote: > > On Thu, Dec 15, 2022 at 4:47 AM Nathan Bossart <[email protected]> wrote: > >> On Wed, Dec 14, 2022 at 02:02:58PM -0500, Tom Lane wrote: > >> > Maybe we could have workers that are exiting for that reason set a > >> > flag saying "please restart me without delay"? > >> > >> That helps a bit, but there are still delays when starting workers for new > >> subscriptions. I think we'd need to create a new array in shared memory > >> for subscription OIDs that need their workers started immediately. > > > > That would be tricky because the list of subscription OIDs can be > > longer than the workers. Can't we set a boolean variable > > (check_immediate or something like that) in LogicalRepCtxStruct and > > use that to traverse the subscriptions? So, when any worker will > > restart because of a parameter change, we can set the variable and > > send a signal to the launcher. The launcher can then check this > > variable to decide whether to start the missing workers for enabled > > subscriptions. > > My approach was to add a variable to LogicalRepWorker that indicated > whether a worker needed to be restarted immediately. While this is a > little weird because the workers array is treated as slots, it worked > nicely for ALTER SUBSCRIPTION. > So, are you planning to keep its in_use and subid flag as it is in logicalrep_worker_cleanup()? Otherwise, without that it could be reused for some other subscription. > However, this doesn't help at all for > CREATE SUBSCRIPTION. > What if we maintain a hash table similar to 'last_start_times' maintained in tablesync.c? It won't have entries for new subscriptions, so for those we may not need to wait till wal_retrieve_retry_interval. > IIUC you are suggesting just one variable that would bypass > wal_retrieve_retry_interval for all subscriptions, not just those newly > altered or created. This definitely seems like it would prevent delays, > but it would also cause wal_retrieve_retry_interval to be incorrectly > bypassed for the other workers in some cases. > Right, but I guess it would be rare in practical cases that someone Altered/Created a subscription, and also some workers are restarted due to errors/crashes as only in those cases launcher can restart the worker when it shouldn't. However, in that case, also, it won't restart the apply worker again and again unless there are concurrent Create/Alter Subscription operations going on. IIUC, currently also it can always first time restart the worker immediately after ERROR/CRASH because we don't maintain last_start_time for each worker. I think this is probably okay as we want to avoid repeated restarts after the ERROR. BTW, now users also have a subscription option 'disable_on_error' which could also be used to avoid repeated restarts due to ERRORS. > Is this acceptable? > To me, this sounds acceptable but if you and others don't think so then we can try to develop some solution like per-worker-flag and a hash table as discussed in the earlier part of the email. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: wake up logical workers after ALTER SUBSCRIPTION 2023-01-04 05:27 Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> @ 2023-01-04 18:12 ` Nathan Bossart <[email protected]> 2023-01-05 00:49 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Nathan Bossart @ 2023-01-04 18:12 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Tom Lane <[email protected]>; Melih Mutlu <[email protected]>; Thomas Munro <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; pgsql-hackers On Wed, Jan 04, 2023 at 10:57:43AM +0530, Amit Kapila wrote: > On Tue, Jan 3, 2023 at 11:40 PM Nathan Bossart <[email protected]> wrote: >> My approach was to add a variable to LogicalRepWorker that indicated >> whether a worker needed to be restarted immediately. While this is a >> little weird because the workers array is treated as slots, it worked >> nicely for ALTER SUBSCRIPTION. > > So, are you planning to keep its in_use and subid flag as it is in > logicalrep_worker_cleanup()? Otherwise, without that it could be > reused for some other subscription. I believe I did something like this in my proof-of-concept. I might have used the new flag as another indicator that the slot was still "in use". In any case, you are right that we need to prevent the slot from being reused. > What if we maintain a hash table similar to 'last_start_times' > maintained in tablesync.c? It won't have entries for new > subscriptions, so for those we may not need to wait till > wal_retrieve_retry_interval. I proposed this upthread [0]. I still think it is a worthwhile change. Right now, if a worker needs to be restarted but another unrelated worker was restarted less than wal_retrieve_retry_interval milliseconds ago, the launcher waits to restart it. I think it makes more sense for each worker to have its own restart interval tracked. >> IIUC you are suggesting just one variable that would bypass >> wal_retrieve_retry_interval for all subscriptions, not just those newly >> altered or created. This definitely seems like it would prevent delays, >> but it would also cause wal_retrieve_retry_interval to be incorrectly >> bypassed for the other workers in some cases. > > Right, but I guess it would be rare in practical cases that someone > Altered/Created a subscription, and also some workers are restarted > due to errors/crashes as only in those cases launcher can restart the > worker when it shouldn't. However, in that case, also, it won't > restart the apply worker again and again unless there are concurrent > Create/Alter Subscription operations going on. IIUC, currently also it > can always first time restart the worker immediately after ERROR/CRASH > because we don't maintain last_start_time for each worker. I think > this is probably okay as we want to avoid repeated restarts after the > ERROR. This line of thinking is why I felt that lowering wal_retrieve_retry_interval for the tests might be sufficient. Besides the fact that it revealed multiple bugs, I don't see the point in adding much more complexity here. In practice, workers will usually start right away, unless of course there are other worker starts happening around the same time. This consistently causes testing delays because the tests stress these code paths, but I don't think what the tests are doing is a typical use-case. From the discussion thus far, it sounds like the alternatives are to 1) add a global flag that causes wal_retrieve_retry_interval to be bypassed for all workers or to 2) add a hash map in the launcher and a restart_immediately flag in each worker slot. I'll go ahead and create a patch for 2 since it seems like the most complete solution, and we can evaluate whether the complexity seems appropriate. [0] https://postgr.es/m/20221214171023.GA689106%40nathanxps13 -- Nathan Bossart Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: wake up logical workers after ALTER SUBSCRIPTION 2023-01-04 05:27 Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> 2023-01-04 18:12 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> @ 2023-01-05 00:49 ` Nathan Bossart <[email protected]> 2023-01-05 05:27 ` Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Nathan Bossart @ 2023-01-05 00:49 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Tom Lane <[email protected]>; Melih Mutlu <[email protected]>; Thomas Munro <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; pgsql-hackers On Wed, Jan 04, 2023 at 10:12:19AM -0800, Nathan Bossart wrote: > From the discussion thus far, it sounds like the alternatives are to 1) add > a global flag that causes wal_retrieve_retry_interval to be bypassed for > all workers or to 2) add a hash map in the launcher and a > restart_immediately flag in each worker slot. I'll go ahead and create a > patch for 2 since it seems like the most complete solution, and we can > evaluate whether the complexity seems appropriate. Here is a first attempt at adding a hash table to the launcher and a restart_immediately flag in each worker slot. This provides a similar speedup to lowering wal_retrieve_retry_interval to 1ms. I've noted a couple of possible race conditions in comments, but none of them seemed particularly egregious. Ideally, we'd put the hash table in shared memory so that other backends could adjust it directly, but IIUC that requires it to be a fixed size, and the number of subscriptions is virtually unbounded. There might still be problems with the patch, but I'm hoping it at least helps further the discussion about which approach to take. -- Nathan Bossart Amazon Web Services: https://aws.amazon.com Attachments: [text/x-diff] v11-0001-wake-up-logical-workers-as-needed-instead-of-rel.patch (6.4K, ../../20230105004906.GA187083@nathanxps13/2-v11-0001-wake-up-logical-workers-as-needed-instead-of-rel.patch) download | inline diff: From 2c4c3c552ca83eeb8a4a7cca724ce455ba98ac0f Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Mon, 21 Nov 2022 16:01:01 -0800 Subject: [PATCH v11 1/2] wake up logical workers as needed instead of relying on periodic wakeups --- src/backend/access/transam/xact.c | 3 ++ src/backend/commands/alter.c | 7 ++++ src/backend/commands/subscriptioncmds.c | 4 ++ src/backend/replication/logical/tablesync.c | 10 +++++ src/backend/replication/logical/worker.c | 46 +++++++++++++++++++++ src/include/replication/logicalworker.h | 3 ++ 6 files changed, 73 insertions(+) diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 24221542e7..54145bf805 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -47,6 +47,7 @@ #include "pgstat.h" #include "replication/logical.h" #include "replication/logicallauncher.h" +#include "replication/logicalworker.h" #include "replication/origin.h" #include "replication/snapbuild.h" #include "replication/syncrep.h" @@ -2360,6 +2361,7 @@ CommitTransaction(void) AtEOXact_PgStat(true, is_parallel_worker); AtEOXact_Snapshot(true, false); AtEOXact_ApplyLauncher(true); + AtEOXact_LogicalRepWorkers(true); pgstat_report_xact_timestamp(0); CurrentResourceOwner = NULL; @@ -2860,6 +2862,7 @@ AbortTransaction(void) AtEOXact_HashTables(false); AtEOXact_PgStat(false, is_parallel_worker); AtEOXact_ApplyLauncher(false); + AtEOXact_LogicalRepWorkers(false); pgstat_report_xact_timestamp(0); } diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c index 70d359eb6a..4e8102b59f 100644 --- a/src/backend/commands/alter.c +++ b/src/backend/commands/alter.c @@ -59,6 +59,7 @@ #include "commands/user.h" #include "miscadmin.h" #include "parser/parse_func.h" +#include "replication/logicalworker.h" #include "rewrite/rewriteDefine.h" #include "tcop/utility.h" #include "utils/builtins.h" @@ -279,6 +280,12 @@ AlterObjectRename_internal(Relation rel, Oid objectId, const char *new_name) if (strncmp(new_name, "regress_", 8) != 0) elog(WARNING, "subscriptions created by regression test cases should have names starting with \"regress_\""); #endif + + /* + * Wake up the logical replication workers to handle this change + * quickly. + */ + LogicalRepWorkersWakeupAtCommit(objectId); } else if (nameCacheId >= 0) { diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index b9c5df796f..b9bbb2cf4e 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -34,6 +34,7 @@ #include "nodes/makefuncs.h" #include "pgstat.h" #include "replication/logicallauncher.h" +#include "replication/logicalworker.h" #include "replication/origin.h" #include "replication/slot.h" #include "replication/walreceiver.h" @@ -1362,6 +1363,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, InvokeObjectPostAlterHook(SubscriptionRelationId, subid, 0); + /* Wake up the logical replication workers to handle this change quickly. */ + LogicalRepWorkersWakeupAtCommit(subid); + return myself; } diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 2fdfeb5b4c..fcadc1e98e 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -105,6 +105,7 @@ #include "pgstat.h" #include "replication/logicallauncher.h" #include "replication/logicalrelation.h" +#include "replication/logicalworker.h" #include "replication/walreceiver.h" #include "replication/worker_internal.h" #include "replication/slot.h" @@ -619,6 +620,15 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) if (started_tx) { + /* + * If we are ready to enable two_phase mode, wake up the logical + * replication workers to handle this change quickly. + */ + CommandCounterIncrement(); + if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING && + AllTablesyncsReady()) + LogicalRepWorkersWakeupAtCommit(MyLogicalRepWorker->subid); + CommitTransactionCommand(); pgstat_report_stat(true); } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 446f84fa97..3e2ea32e1e 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -254,6 +254,8 @@ WalReceiverConn *LogRepWorkerWalRcvConn = NULL; Subscription *MySubscription = NULL; static bool MySubscriptionValid = false; +static List *on_commit_wakeup_workers_subids = NIL; + bool in_remote_transaction = false; static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr; @@ -4097,3 +4099,47 @@ reset_apply_error_context_info(void) apply_error_callback_arg.remote_attnum = -1; set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr); } + +/* + * Wakeup the stored subscriptions' workers on commit if requested. + */ +void +AtEOXact_LogicalRepWorkers(bool isCommit) +{ + if (isCommit && on_commit_wakeup_workers_subids != NIL) + { + ListCell *subid; + + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + foreach(subid, on_commit_wakeup_workers_subids) + { + List *workers; + ListCell *worker; + + workers = logicalrep_workers_find(lfirst_oid(subid), true); + foreach(worker, workers) + logicalrep_worker_wakeup_ptr((LogicalRepWorker *) lfirst(worker)); + } + LWLockRelease(LogicalRepWorkerLock); + } + + on_commit_wakeup_workers_subids = NIL; +} + +/* + * Request wakeup of the workers for the given subscription ID on commit of the + * transaction. + * + * This is used to ensure that the workers process assorted changes as soon as + * possible. + */ +void +LogicalRepWorkersWakeupAtCommit(Oid subid) +{ + MemoryContext oldcxt; + + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + on_commit_wakeup_workers_subids = list_append_unique_oid(on_commit_wakeup_workers_subids, + subid); + MemoryContextSwitchTo(oldcxt); +} diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h index f1e7e8a348..3b2084f81f 100644 --- a/src/include/replication/logicalworker.h +++ b/src/include/replication/logicalworker.h @@ -16,4 +16,7 @@ extern void ApplyWorkerMain(Datum main_arg); extern bool IsLogicalWorker(void); +extern void LogicalRepWorkersWakeupAtCommit(Oid subid); +extern void AtEOXact_LogicalRepWorkers(bool isCommit); + #endif /* LOGICALWORKER_H */ -- 2.25.1 [text/x-diff] v11-0002-bypass-wal_retrieve_retry_interval-for-logical-w.patch (10.0K, ../../20230105004906.GA187083@nathanxps13/3-v11-0002-bypass-wal_retrieve_retry_interval-for-logical-w.patch) download | inline diff: From 126e41be1c0af2260fa84f6a3bf6b738442df58f Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Wed, 4 Jan 2023 11:38:20 -0800 Subject: [PATCH v11 2/2] bypass wal_retrieve_retry_interval for logical workers as appropriate --- src/backend/commands/subscriptioncmds.c | 9 +- src/backend/replication/logical/launcher.c | 155 +++++++++++++++----- src/backend/replication/logical/tablesync.c | 2 +- src/backend/replication/logical/worker.c | 2 +- src/include/replication/worker_internal.h | 3 + 5 files changed, 128 insertions(+), 43 deletions(-) diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index b9bbb2cf4e..3293ec2043 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1137,8 +1137,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, BoolGetDatum(opts.enabled); replaces[Anum_pg_subscription_subenabled - 1] = true; - if (opts.enabled) - ApplyLauncherWakeupAtCommit(); + /* + * Even if the subscription is disabled, we wake up the + * launcher so that it clears its last start time. This + * ensures the workers will be able to start up right away when + * the subscription is enabled again. + */ + ApplyLauncherWakeupAtCommit(); update_tuple = true; break; diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index a69e371c05..544e73dd62 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -297,7 +297,7 @@ retry: { LogicalRepWorker *w = &LogicalRepCtx->workers[i]; - if (!w->in_use) + if (!w->in_use && !w->restart_immediately) { worker = w; slot = i; @@ -620,8 +620,14 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker) worker->proc = NULL; worker->dbid = InvalidOid; worker->userid = InvalidOid; - worker->subid = InvalidOid; worker->relid = InvalidOid; + + /* + * If restart_immediately was set, retain the subscription ID so that the + * launcher knows which worker it should restart right away. + */ + if (!worker->restart_immediately) + worker->subid = InvalidOid; } /* @@ -801,7 +807,13 @@ ApplyLauncherWakeup(void) void ApplyLauncherMain(Datum main_arg) { - TimestampTz last_start_time = 0; + struct launcher_start_time_mapping + { + Oid subid; + TimestampTz last_start_time; + }; + HTAB *last_start_times = NULL; + HASHCTL ctl; ereport(DEBUG1, (errmsg_internal("logical replication launcher started"))); @@ -822,6 +834,18 @@ ApplyLauncherMain(Datum main_arg) */ BackgroundWorkerInitializeConnection(NULL, NULL, 0); + /* + * Prepare a hash table for tracking last start times of workers, to avoid + * immediate restarts. Ideally, this hash table would be created in shared + * memory so that other backends could adjust it directly to avoid race + * conditions, but it would need to be a fixed size, and the number of + * subscriptions is virtually unbounded. + */ + ctl.keysize = sizeof(Oid); + ctl.entrysize = sizeof(struct launcher_start_time_mapping); + last_start_times = hash_create("Logical replication apply worker start times", + 256, &ctl, HASH_ELEM | HASH_BLOBS); + /* Enter main loop */ for (;;) { @@ -832,63 +856,116 @@ ApplyLauncherMain(Datum main_arg) MemoryContext oldctx; TimestampTz now; long wait_time = DEFAULT_NAPTIME_PER_CYCLE; + HASH_SEQ_STATUS status; + struct launcher_start_time_mapping *hentry; CHECK_FOR_INTERRUPTS(); now = GetCurrentTimestamp(); - /* Limit the start retry to once a wal_retrieve_retry_interval */ - if (TimestampDifferenceExceeds(last_start_time, now, - wal_retrieve_retry_interval)) + /* Use temporary context for the database list and worker info. */ + subctx = AllocSetContextCreate(TopMemoryContext, + "Logical Replication Launcher sublist", + ALLOCSET_DEFAULT_SIZES); + oldctx = MemoryContextSwitchTo(subctx); + + sublist = get_subscription_list(); + foreach(lc, sublist) { - /* Use temporary context for the database list and worker info. */ - subctx = AllocSetContextCreate(TopMemoryContext, - "Logical Replication Launcher sublist", - ALLOCSET_DEFAULT_SIZES); - oldctx = MemoryContextSwitchTo(subctx); + Subscription *sub = (Subscription *) lfirst(lc); + bool bypass_retry_interval = false; + LogicalRepWorker *w; - /* search for subscriptions to start or stop. */ - sublist = get_subscription_list(); + LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE); - /* Start the missing workers for enabled subscriptions. */ - foreach(lc, sublist) + /* + * Bypass wal_retrieve_retry_interval if the worker set + * restart_immediately. We do this before checking if the + * subscription is enabled so that the slot can be freed + * regardless. + */ + for (int i = 0; i < max_logical_replication_workers; i++) { - Subscription *sub = (Subscription *) lfirst(lc); - LogicalRepWorker *w; + w = &LogicalRepCtx->workers[i]; - if (!sub->enabled) - continue; + if (!w->in_use && !w->proc && w->subid == sub->oid && + w->relid == InvalidOid && w->restart_immediately) + { + w->restart_immediately = false; + logicalrep_worker_cleanup(w); + bypass_retry_interval = true; + break; + } + } - LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - w = logicalrep_worker_find(sub->oid, InvalidOid, false); - LWLockRelease(LogicalRepWorkerLock); + w = logicalrep_worker_find(sub->oid, InvalidOid, false); + LWLockRelease(LogicalRepWorkerLock); - if (w == NULL) - { - last_start_time = now; - wait_time = wal_retrieve_retry_interval; + /* + * If the subscription isn't enabled, clear its entry in + * last_start_times so that its apply worker is restarted right + * away when it is enabled again. There is a chance that the + * subscription could be enabled again before we've had a chance to + * clear its entry, in which case we'll wait a little bit before + * starting the worker. + */ + if (!sub->enabled) + { + hash_search(last_start_times, &sub->oid, HASH_REMOVE, NULL); + continue; + } + + /* + * If its okay to start the worker now, do so. Otherwise, adjust + * wait_time so that we wake up when we can start it. + */ + if (w == NULL) + { + bool found; + hentry = hash_search(last_start_times, &sub->oid, + HASH_ENTER, &found); + + if (bypass_retry_interval || !found || + TimestampDifferenceExceeds(hentry->last_start_time, now, + wal_retrieve_retry_interval)) + { + hentry->last_start_time = now; logicalrep_worker_launch(sub->dbid, sub->oid, sub->name, sub->owner, InvalidOid); } - } + else + { + long elapsed; - /* Switch back to original memory context. */ - MemoryContextSwitchTo(oldctx); - /* Clean the temporary memory. */ - MemoryContextDelete(subctx); + elapsed = TimestampDifferenceMilliseconds(hentry->last_start_time, now); + wait_time = Min(wait_time, wal_retrieve_retry_interval - elapsed); + } + } } - else + + /* + * Do garbage collection on the last_start_times hash table. In + * theory, a subscription OID could be reused before its entry is + * removed, but the risk of that seems low, and at worst the launcher + * will wait a bit longer before starting the new subscription's apply + * worker. This risk could be reduced by removing entries for + * subscriptions that aren't in sublist, but it doesn't seem worth the + * trouble. + */ + hash_seq_init(&status, last_start_times); + while ((hentry = (struct launcher_start_time_mapping *) hash_seq_search(&status)) != NULL) { - /* - * The wait in previous cycle was interrupted in less than - * wal_retrieve_retry_interval since last worker was started, this - * usually means crash of the worker, so we should retry in - * wal_retrieve_retry_interval again. - */ - wait_time = wal_retrieve_retry_interval; + if (TimestampDifferenceExceeds(hentry->last_start_time, now, + wal_retrieve_retry_interval)) + hash_search(last_start_times, &hentry->subid, HASH_REMOVE, NULL); } + /* Switch back to original memory context. */ + MemoryContextSwitchTo(oldctx); + /* Clean the temporary memory. */ + MemoryContextDelete(subctx); + /* Wait for more work. */ rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index fcadc1e98e..3fa6472c88 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -465,7 +465,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) ereport(LOG, (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled", MySubscription->name))); - + MyLogicalRepWorker->restart_immediately = true; proc_exit(0); } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 3e2ea32e1e..0521459898 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -3130,7 +3130,7 @@ maybe_reread_subscription(void) ereport(LOG, (errmsg("logical replication apply worker for subscription \"%s\" will restart because of a parameter change", MySubscription->name))); - + MyLogicalRepWorker->restart_immediately = true; proc_exit(0); } diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 2a3ec5c2d8..68a56c175b 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -66,6 +66,9 @@ typedef struct LogicalRepWorker TimestampTz last_recv_time; XLogRecPtr reply_lsn; TimestampTz reply_time; + + /* Should the launcher restart the worker immediately? */ + bool restart_immediately; } LogicalRepWorker; /* Main memory context for apply worker. Permanent during worker lifetime. */ -- 2.25.1 ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: wake up logical workers after ALTER SUBSCRIPTION 2023-01-04 05:27 Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> 2023-01-04 18:12 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-01-05 00:49 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> @ 2023-01-05 05:27 ` Amit Kapila <[email protected]> 2023-01-05 17:29 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Amit Kapila @ 2023-01-05 05:27 UTC (permalink / raw) To: Nathan Bossart <[email protected]>; +Cc: Tom Lane <[email protected]>; Melih Mutlu <[email protected]>; Thomas Munro <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; pgsql-hackers On Thu, Jan 5, 2023 at 6:19 AM Nathan Bossart <[email protected]> wrote: > > On Wed, Jan 04, 2023 at 10:12:19AM -0800, Nathan Bossart wrote: > > From the discussion thus far, it sounds like the alternatives are to 1) add > > a global flag that causes wal_retrieve_retry_interval to be bypassed for > > all workers or to 2) add a hash map in the launcher and a > > restart_immediately flag in each worker slot. I'll go ahead and create a > > patch for 2 since it seems like the most complete solution, and we can > > evaluate whether the complexity seems appropriate. > > Here is a first attempt at adding a hash table to the launcher and a > restart_immediately flag in each worker slot. This provides a similar > speedup to lowering wal_retrieve_retry_interval to 1ms. I've noted a > couple of possible race conditions in comments, but none of them seemed > particularly egregious. Ideally, we'd put the hash table in shared memory > so that other backends could adjust it directly, but IIUC that requires it > to be a fixed size, and the number of subscriptions is virtually unbounded. > True, if we want we can use dshash for this. The garbage collection mechanism used in the patch seems odd to me as that will remove/add entries to the hash table even when the corresponding subscription is never dropped. Also, adding this garbage collection each time seems like an overhead, especially for small values of wal_retrieve_retry_interval and a large number of subscriptions. Another point is immediately after cleaning the worker info, trying to find it again seems of no use. In logicalrep_worker_launch(), using both in_use and restart_immediately to find an unused slot doesn't look neat to me, we could probably keep the in_use flag intact if we want to reuse the worker. But again after freeing the worker, keeping its associated slot allocated sounds odd to me. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: wake up logical workers after ALTER SUBSCRIPTION 2023-01-04 05:27 Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> 2023-01-04 18:12 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-01-05 00:49 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-01-05 05:27 ` Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> @ 2023-01-05 17:29 ` Nathan Bossart <[email protected]> 2023-01-06 00:00 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Nathan Bossart @ 2023-01-05 17:29 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Tom Lane <[email protected]>; Melih Mutlu <[email protected]>; Thomas Munro <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; pgsql-hackers On Thu, Jan 05, 2023 at 10:57:58AM +0530, Amit Kapila wrote: > True, if we want we can use dshash for this. I'll look into this. > The garbage collection > mechanism used in the patch seems odd to me as that will remove/add > entries to the hash table even when the corresponding subscription is > never dropped. Yeah, I think this deserves a comment. We can remove anything beyond wal_retrieve_retry_interval because the lack of a hash table entry is taken to mean that we can start the worker immediately. There might be a corner case when wal_retrieve_retry_interval is concurrently updated, in which case we'll effectively use the previous value for the worker. That doesn't seem too terrible to me. It might be possible to remove this garbage collection completely if we use dshash, but I haven't thought through that approach completely yet. > Also, adding this garbage collection each time seems > like an overhead, especially for small values of > wal_retrieve_retry_interval and a large number of subscriptions. Right. > Another point is immediately after cleaning the worker info, trying to > find it again seems of no use. In logicalrep_worker_launch(), using > both in_use and restart_immediately to find an unused slot doesn't > look neat to me, we could probably keep the in_use flag intact if we > want to reuse the worker. But again after freeing the worker, keeping > its associated slot allocated sounds odd to me. Yeah, this flag certainly feels hacky. With a shared hash table, we could just have backends remove the last-start-time entry directly, and we wouldn't need the flag. -- Nathan Bossart Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: wake up logical workers after ALTER SUBSCRIPTION 2023-01-04 05:27 Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> 2023-01-04 18:12 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-01-05 00:49 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-01-05 05:27 ` Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> 2023-01-05 17:29 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> @ 2023-01-06 00:00 ` Nathan Bossart <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Nathan Bossart @ 2023-01-06 00:00 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Tom Lane <[email protected]>; Melih Mutlu <[email protected]>; Thomas Munro <[email protected]>; Hayato Kuroda (Fujitsu) <[email protected]>; pgsql-hackers On Thu, Jan 05, 2023 at 09:29:24AM -0800, Nathan Bossart wrote: > On Thu, Jan 05, 2023 at 10:57:58AM +0530, Amit Kapila wrote: >> True, if we want we can use dshash for this. > > I'll look into this. Here is an attempt at using dshash. This is quite a bit cleaner since we don't need garbage collection or the flag in the worker slots. There is some extra work required to set up the table, but it doesn't seem too bad. -- Nathan Bossart Amazon Web Services: https://aws.amazon.com Attachments: [text/x-diff] v13-0001-wake-up-logical-workers-as-needed-instead-of-rel.patch (8.2K, ../../20230106000027.GA581414@nathanxps13/2-v13-0001-wake-up-logical-workers-as-needed-instead-of-rel.patch) download | inline diff: From 8853500e9b238b0d07871940d4e7db9e43802b8b Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Mon, 21 Nov 2022 16:01:01 -0800 Subject: [PATCH v13 1/2] wake up logical workers as needed instead of relying on periodic wakeups --- src/backend/access/transam/xact.c | 3 ++ src/backend/commands/alter.c | 7 +++ src/backend/commands/subscriptioncmds.c | 7 +++ src/backend/replication/logical/tablesync.c | 49 ++++++++++++--------- src/backend/replication/logical/worker.c | 46 +++++++++++++++++++ src/include/replication/logicalworker.h | 3 ++ 6 files changed, 93 insertions(+), 22 deletions(-) diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 24221542e7..54145bf805 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -47,6 +47,7 @@ #include "pgstat.h" #include "replication/logical.h" #include "replication/logicallauncher.h" +#include "replication/logicalworker.h" #include "replication/origin.h" #include "replication/snapbuild.h" #include "replication/syncrep.h" @@ -2360,6 +2361,7 @@ CommitTransaction(void) AtEOXact_PgStat(true, is_parallel_worker); AtEOXact_Snapshot(true, false); AtEOXact_ApplyLauncher(true); + AtEOXact_LogicalRepWorkers(true); pgstat_report_xact_timestamp(0); CurrentResourceOwner = NULL; @@ -2860,6 +2862,7 @@ AbortTransaction(void) AtEOXact_HashTables(false); AtEOXact_PgStat(false, is_parallel_worker); AtEOXact_ApplyLauncher(false); + AtEOXact_LogicalRepWorkers(false); pgstat_report_xact_timestamp(0); } diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c index 70d359eb6a..4e8102b59f 100644 --- a/src/backend/commands/alter.c +++ b/src/backend/commands/alter.c @@ -59,6 +59,7 @@ #include "commands/user.h" #include "miscadmin.h" #include "parser/parse_func.h" +#include "replication/logicalworker.h" #include "rewrite/rewriteDefine.h" #include "tcop/utility.h" #include "utils/builtins.h" @@ -279,6 +280,12 @@ AlterObjectRename_internal(Relation rel, Oid objectId, const char *new_name) if (strncmp(new_name, "regress_", 8) != 0) elog(WARNING, "subscriptions created by regression test cases should have names starting with \"regress_\""); #endif + + /* + * Wake up the logical replication workers to handle this change + * quickly. + */ + LogicalRepWorkersWakeupAtCommit(objectId); } else if (nameCacheId >= 0) { diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index b9c5df796f..c0fd82cdf2 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -34,6 +34,7 @@ #include "nodes/makefuncs.h" #include "pgstat.h" #include "replication/logicallauncher.h" +#include "replication/logicalworker.h" #include "replication/origin.h" #include "replication/slot.h" #include "replication/walreceiver.h" @@ -1362,6 +1363,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, InvokeObjectPostAlterHook(SubscriptionRelationId, subid, 0); + /* Wake up the logical replication workers to handle this change quickly. */ + LogicalRepWorkersWakeupAtCommit(subid); + return myself; } @@ -1733,6 +1737,9 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId) form->oid, 0); ApplyLauncherWakeupAtCommit(); + + /* Wake up the logical replication workers to handle this change quickly. */ + LogicalRepWorkersWakeupAtCommit(form->oid); } /* diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 2fdfeb5b4c..8c8f27ebcf 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -415,6 +415,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) static HTAB *last_start_times = NULL; ListCell *lc; bool started_tx = false; + bool should_exit = false; Assert(!IsTransactionState()); @@ -446,28 +447,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) last_start_times = NULL; } - /* - * Even when the two_phase mode is requested by the user, it remains as - * 'pending' until all tablesyncs have reached READY state. - * - * When this happens, we restart the apply worker and (if the conditions - * are still ok) then the two_phase tri-state will become 'enabled' at - * that time. - * - * Note: If the subscription has no tables then leave the state as - * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to - * work. - */ - if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING && - AllTablesyncsReady()) - { - ereport(LOG, - (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled", - MySubscription->name))); - - proc_exit(0); - } - /* * Process all tables that are being synchronized. */ @@ -619,9 +598,35 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) if (started_tx) { + /* + * Even when the two_phase mode is requested by the user, it remains as + * 'pending' until all tablesyncs have reached READY state. + * + * When this happens, we restart the apply worker and (if the conditions + * are still ok) then the two_phase tri-state will become 'enabled' at + * that time. + * + * Note: If the subscription has no tables then leave the state as + * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to + * work. + */ + CommandCounterIncrement(); /* make updates visible */ + if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING && + AllTablesyncsReady()) + { + ereport(LOG, + (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled", + MySubscription->name))); + + should_exit = true; + } + CommitTransactionCommand(); pgstat_report_stat(true); } + + if (should_exit) + proc_exit(0); } /* diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index f8e8cf71eb..cf38ed821e 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -254,6 +254,8 @@ WalReceiverConn *LogRepWorkerWalRcvConn = NULL; Subscription *MySubscription = NULL; static bool MySubscriptionValid = false; +static List *on_commit_wakeup_workers_subids = NIL; + bool in_remote_transaction = false; static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr; @@ -4092,3 +4094,47 @@ reset_apply_error_context_info(void) apply_error_callback_arg.remote_attnum = -1; set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr); } + +/* + * Wakeup the stored subscriptions' workers on commit if requested. + */ +void +AtEOXact_LogicalRepWorkers(bool isCommit) +{ + if (isCommit && on_commit_wakeup_workers_subids != NIL) + { + ListCell *subid; + + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + foreach(subid, on_commit_wakeup_workers_subids) + { + List *workers; + ListCell *worker; + + workers = logicalrep_workers_find(lfirst_oid(subid), true); + foreach(worker, workers) + logicalrep_worker_wakeup_ptr((LogicalRepWorker *) lfirst(worker)); + } + LWLockRelease(LogicalRepWorkerLock); + } + + on_commit_wakeup_workers_subids = NIL; +} + +/* + * Request wakeup of the workers for the given subscription ID on commit of the + * transaction. + * + * This is used to ensure that the workers process assorted changes as soon as + * possible. + */ +void +LogicalRepWorkersWakeupAtCommit(Oid subid) +{ + MemoryContext oldcxt; + + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + on_commit_wakeup_workers_subids = list_append_unique_oid(on_commit_wakeup_workers_subids, + subid); + MemoryContextSwitchTo(oldcxt); +} diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h index f1e7e8a348..3b2084f81f 100644 --- a/src/include/replication/logicalworker.h +++ b/src/include/replication/logicalworker.h @@ -16,4 +16,7 @@ extern void ApplyWorkerMain(Datum main_arg); extern bool IsLogicalWorker(void); +extern void LogicalRepWorkersWakeupAtCommit(Oid subid); +extern void AtEOXact_LogicalRepWorkers(bool isCommit); + #endif /* LOGICALWORKER_H */ -- 2.25.1 [text/x-diff] v13-0002-bypass-wal_retrieve_retry_interval-for-logical-w.patch (14.1K, ../../20230106000027.GA581414@nathanxps13/3-v13-0002-bypass-wal_retrieve_retry_interval-for-logical-w.patch) download | inline diff: From 8aae426e8a10df698e299ed4ff7a13e5b8444efc Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Thu, 5 Jan 2023 14:22:53 -0800 Subject: [PATCH v13 2/2] bypass wal_retrieve_retry_interval for logical workers as appropriate --- doc/src/sgml/monitoring.sgml | 10 + src/backend/commands/subscriptioncmds.c | 20 ++ src/backend/replication/logical/launcher.c | 201 +++++++++++++++----- src/backend/replication/logical/tablesync.c | 6 + src/backend/replication/logical/worker.c | 8 + src/backend/storage/lmgr/lwlock.c | 4 + src/include/replication/logicallauncher.h | 2 + src/include/storage/lwlock.h | 2 + 8 files changed, 208 insertions(+), 45 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 5bcba0fdec..8f06e234c8 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -1991,6 +1991,16 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser <entry>Waiting to read or update information about <quote>heavyweight</quote> locks.</entry> </row> + <row> + <entry><literal>LogicalRepLauncherDSA</literal></entry> + <entry>Waiting for logical replication launcher dynamic shared memory + allocator access</entry> + </row> + <row> + <entry><literal>LogicalRepLauncherHash</literal></entry> + <entry>Waiting for logical replication launcher shared memory hash table + access</entry> + </row> <row> <entry><literal>LogicalRepWorker</literal></entry> <entry>Waiting to read or update the state of logical replication diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index c0fd82cdf2..bab576e966 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1139,6 +1139,18 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (opts.enabled) ApplyLauncherWakeupAtCommit(); + else + { + /* + * Clear the last-start time for the apply worker so the + * worker is restarted right away when the subscription is + * enabled again. If this transaction rolls back, the + * launcher might restart the apply worker before + * wal_retrieve_retry_interval milliseconds have elapsed, + * but that's probably okay. + */ + logicalrep_launcher_delete_last_start_time(sub->oid); + } update_tuple = true; break; @@ -1504,6 +1516,14 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) } list_free(subworkers); + /* + * Clear the last-start time for the apply worker to free up space. If + * this transaction rolls back, the launcher might restart the apply worker + * before wal_retrieve_retry_interval milliseconds have elapsed, but that's + * probably okay. + */ + logicalrep_launcher_delete_last_start_time(subid); + /* * Cleanup of tablesync replication origins. * diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index a69e371c05..dfe49db64f 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -25,6 +25,7 @@ #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" #include "funcapi.h" +#include "lib/dshash.h" #include "libpq/pqsignal.h" #include "miscadmin.h" #include "pgstat.h" @@ -56,6 +57,25 @@ int max_logical_replication_workers = 4; int max_sync_workers_per_subscription = 2; +/* an entry in the last-start times hash table */ +typedef struct LauncherLastStartTimesEntry +{ + Oid subid; + TimestampTz last_start_time; +} LauncherLastStartTimesEntry; + +/* parameters for the last-start times hash table */ +static const dshash_parameters dsh_params = { + sizeof(Oid), + sizeof(LauncherLastStartTimesEntry), + dshash_memcmp, + dshash_memhash, + LWTRANCHE_LAUNCHER_HASH +}; + +static dsa_area *last_start_times_dsa = NULL; +static dshash_table *last_start_times = NULL; + LogicalRepWorker *MyLogicalRepWorker = NULL; typedef struct LogicalRepCtxStruct @@ -63,6 +83,10 @@ typedef struct LogicalRepCtxStruct /* Supervisor process. */ pid_t launcher_pid; + /* hash table for last-start times */ + dsa_handle last_start_dsa; + dshash_table_handle last_start_dsh; + /* Background workers. */ LogicalRepWorker workers[FLEXIBLE_ARRAY_MEMBER]; } LogicalRepCtxStruct; @@ -74,6 +98,9 @@ static void logicalrep_launcher_onexit(int code, Datum arg); static void logicalrep_worker_onexit(int code, Datum arg); static void logicalrep_worker_detach(void); static void logicalrep_worker_cleanup(LogicalRepWorker *worker); +static void logicalrep_launcher_attach_dshmem(void); +static void logicalrep_launcher_set_last_start_time(Oid subid, TimestampTz start_time); +static TimestampTz logicalrep_launcher_get_last_start_time(Oid subid); static bool on_commit_launcher_wakeup = false; @@ -756,6 +783,9 @@ ApplyLauncherShmemInit(void) memset(worker, 0, sizeof(LogicalRepWorker)); SpinLockInit(&worker->relmutex); } + + LogicalRepCtx->last_start_dsa = DSM_HANDLE_INVALID; + LogicalRepCtx->last_start_dsh = DSM_HANDLE_INVALID; } } @@ -801,8 +831,6 @@ ApplyLauncherWakeup(void) void ApplyLauncherMain(Datum main_arg) { - TimestampTz last_start_time = 0; - ereport(DEBUG1, (errmsg_internal("logical replication launcher started"))); @@ -837,58 +865,55 @@ ApplyLauncherMain(Datum main_arg) now = GetCurrentTimestamp(); - /* Limit the start retry to once a wal_retrieve_retry_interval */ - if (TimestampDifferenceExceeds(last_start_time, now, - wal_retrieve_retry_interval)) - { - /* Use temporary context for the database list and worker info. */ - subctx = AllocSetContextCreate(TopMemoryContext, - "Logical Replication Launcher sublist", - ALLOCSET_DEFAULT_SIZES); - oldctx = MemoryContextSwitchTo(subctx); - - /* search for subscriptions to start or stop. */ - sublist = get_subscription_list(); - - /* Start the missing workers for enabled subscriptions. */ - foreach(lc, sublist) - { - Subscription *sub = (Subscription *) lfirst(lc); - LogicalRepWorker *w; + /* Use temporary context for the database list and worker info. */ + subctx = AllocSetContextCreate(TopMemoryContext, + "Logical Replication Launcher sublist", + ALLOCSET_DEFAULT_SIZES); + oldctx = MemoryContextSwitchTo(subctx); - if (!sub->enabled) - continue; + sublist = get_subscription_list(); + foreach(lc, sublist) + { + Subscription *sub = (Subscription *) lfirst(lc); + LogicalRepWorker *w; + TimestampTz last_start; - LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - w = logicalrep_worker_find(sub->oid, InvalidOid, false); - LWLockRelease(LogicalRepWorkerLock); + if (!sub->enabled) + continue; - if (w == NULL) - { - last_start_time = now; - wait_time = wal_retrieve_retry_interval; + LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE); + w = logicalrep_worker_find(sub->oid, InvalidOid, false); + LWLockRelease(LogicalRepWorkerLock); - logicalrep_worker_launch(sub->dbid, sub->oid, sub->name, - sub->owner, InvalidOid); - } - } + if (w != NULL) + continue; - /* Switch back to original memory context. */ - MemoryContextSwitchTo(oldctx); - /* Clean the temporary memory. */ - MemoryContextDelete(subctx); - } - else - { /* - * The wait in previous cycle was interrupted in less than - * wal_retrieve_retry_interval since last worker was started, this - * usually means crash of the worker, so we should retry in - * wal_retrieve_retry_interval again. + * If the worker is eligible to start now, launch it. Otherwise, + * adjust wait_time so that we wake up when it can be started. */ - wait_time = wal_retrieve_retry_interval; + last_start = logicalrep_launcher_get_last_start_time(sub->oid); + if (TimestampDifferenceExceeds(last_start, now, + wal_retrieve_retry_interval)) + { + logicalrep_launcher_set_last_start_time(sub->oid, now); + logicalrep_worker_launch(sub->dbid, sub->oid, sub->name, + sub->owner, InvalidOid); + } + else + { + long elapsed; + + elapsed = TimestampDifferenceMilliseconds(last_start, now); + wait_time = Min(wait_time, wal_retrieve_retry_interval - elapsed); + } } + /* Switch back to original memory context. */ + MemoryContextSwitchTo(oldctx); + /* Clean the temporary memory. */ + MemoryContextDelete(subctx); + /* Wait for more work. */ rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, @@ -996,3 +1021,89 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS) return (Datum) 0; } + +/* + * Initialize or attach to the dynamic shared hash table that stores the + * last-start times, if not already done. This must be called before using the + * table. + */ +static void +logicalrep_launcher_attach_dshmem(void) +{ + MemoryContext oldcontext; + + oldcontext = MemoryContextSwitchTo(TopMemoryContext); + LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE); + + if (LogicalRepCtx->last_start_dsh == DSM_HANDLE_INVALID) + { + /* Initialize dynamic shared hash table for last-start times. */ + last_start_times_dsa = dsa_create(LWTRANCHE_LAUNCHER_DSA); + dsa_pin(last_start_times_dsa); + dsa_pin_mapping(last_start_times_dsa); + last_start_times = dshash_create(last_start_times_dsa, &dsh_params, 0); + + /* Store handles in shared memory for other backends to use. */ + LogicalRepCtx->last_start_dsa = dsa_get_handle(last_start_times_dsa); + LogicalRepCtx->last_start_dsh = dshash_get_hash_table_handle(last_start_times); + } + else if (!last_start_times) + { + /* Attach to existing dynamic shared hash table. */ + last_start_times_dsa = dsa_attach(LogicalRepCtx->last_start_dsa); + dsa_pin_mapping(last_start_times_dsa); + last_start_times = dshash_attach(last_start_times_dsa, &dsh_params, + LogicalRepCtx->last_start_dsh, 0); + } + + LWLockRelease(LogicalRepWorkerLock); + MemoryContextSwitchTo(oldcontext); +} + +/* + * Set the last-start time for the subscription. + */ +static void +logicalrep_launcher_set_last_start_time(Oid subid, TimestampTz start_time) +{ + LauncherLastStartTimesEntry *entry; + bool found; + + logicalrep_launcher_attach_dshmem(); + + entry = dshash_find_or_insert(last_start_times, &subid, &found); + entry->last_start_time = start_time; + dshash_release_lock(last_start_times, entry); +} + +/* + * Return the last-start time for the subscription, or 0 if there isn't one. + */ +static TimestampTz +logicalrep_launcher_get_last_start_time(Oid subid) +{ + LauncherLastStartTimesEntry *entry; + TimestampTz ret; + + logicalrep_launcher_attach_dshmem(); + + entry = dshash_find(last_start_times, &subid, false); + if (entry == NULL) + return 0; + + ret = entry->last_start_time; + dshash_release_lock(last_start_times, entry); + + return ret; +} + +/* + * Remove the last-start time for the subscription, if one exists. + */ +void +logicalrep_launcher_delete_last_start_time(Oid subid) +{ + logicalrep_launcher_attach_dshmem(); + + (void) dshash_delete_key(last_start_times, &subid); +} diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 8c8f27ebcf..adc1043211 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -618,6 +618,12 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled", MySubscription->name))); + /* + * Clear the last-start time for the worker so that the launcher + * restarts it immediately, bypassing wal_retrieve_retry_interval. + */ + logicalrep_launcher_delete_last_start_time(MySubscription->oid); + should_exit = true; } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index cf38ed821e..b319990bcc 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -164,6 +164,7 @@ #include "postmaster/walwriter.h" #include "replication/decode.h" #include "replication/logical.h" +#include "replication/logicallauncher.h" #include "replication/logicalproto.h" #include "replication/logicalrelation.h" #include "replication/logicalworker.h" @@ -3126,6 +3127,13 @@ maybe_reread_subscription(void) (errmsg("logical replication apply worker for subscription \"%s\" will restart because of a parameter change", MySubscription->name))); + /* + * Clear the last-start time for the apply worker so that the launcher + * will restart it immediately, bypassing wal_retreive_retry_interval. + */ + if (!am_tablesync_worker()) + logicalrep_launcher_delete_last_start_time(MySubscription->oid); + proc_exit(0); } diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c index 196bece0a3..d2ec396045 100644 --- a/src/backend/storage/lmgr/lwlock.c +++ b/src/backend/storage/lmgr/lwlock.c @@ -186,6 +186,10 @@ static const char *const BuiltinTrancheNames[] = { "PgStatsHash", /* LWTRANCHE_PGSTATS_DATA: */ "PgStatsData", + /* LWTRANCHE_LAUNCHER_DSA: */ + "LogicalRepLauncherDSA", + /* LWTRANCHE_LAUNCHER_HASH: */ + "LogicalRepLauncherHash", }; StaticAssertDecl(lengthof(BuiltinTrancheNames) == diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h index e1661b6c91..00f6f89d72 100644 --- a/src/include/replication/logicallauncher.h +++ b/src/include/replication/logicallauncher.h @@ -26,4 +26,6 @@ extern void AtEOXact_ApplyLauncher(bool isCommit); extern bool IsLogicalLauncher(void); +extern void logicalrep_launcher_delete_last_start_time(Oid subid); + #endif /* LOGICALLAUNCHER_H */ diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h index e4162db613..d2c7afb8f4 100644 --- a/src/include/storage/lwlock.h +++ b/src/include/storage/lwlock.h @@ -204,6 +204,8 @@ typedef enum BuiltinTrancheIds LWTRANCHE_PGSTATS_DSA, LWTRANCHE_PGSTATS_HASH, LWTRANCHE_PGSTATS_DATA, + LWTRANCHE_LAUNCHER_DSA, + LWTRANCHE_LAUNCHER_HASH, LWTRANCHE_FIRST_USER_DEFINED } BuiltinTrancheIds; -- 2.25.1 ^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH v1 6/7] Row pattern recognition patch (tests). @ 2023-06-25 11:48 Tatsuo Ishii <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Tatsuo Ishii @ 2023-06-25 11:48 UTC (permalink / raw) --- src/test/regress/expected/rpr.out | 273 +++++++++++++++++++++++++++++ src/test/regress/parallel_schedule | 2 +- src/test/regress/sql/rpr.sql | 150 ++++++++++++++++ 3 files changed, 424 insertions(+), 1 deletion(-) create mode 100644 src/test/regress/expected/rpr.out create mode 100644 src/test/regress/sql/rpr.sql diff --git a/src/test/regress/expected/rpr.out b/src/test/regress/expected/rpr.out new file mode 100644 index 0000000000..9ede60ba39 --- /dev/null +++ b/src/test/regress/expected/rpr.out @@ -0,0 +1,273 @@ +-- +-- Test for row pattern definition clause +-- +CREATE TEMP TABLE stock ( + company TEXT, + tdate DATE, + price INTEGER + ); +INSERT INTO stock VALUES ('company1', '2023-07-01', 100); +INSERT INTO stock VALUES ('company1', '2023-07-02', 200); +INSERT INTO stock VALUES ('company1', '2023-07-03', 150); +INSERT INTO stock VALUES ('company1', '2023-07-04', 140); +INSERT INTO stock VALUES ('company1', '2023-07-05', 150); +INSERT INTO stock VALUES ('company1', '2023-07-06', 90); +INSERT INTO stock VALUES ('company1', '2023-07-07', 110); +INSERT INTO stock VALUES ('company1', '2023-07-08', 130); +INSERT INTO stock VALUES ('company1', '2023-07-09', 120); +INSERT INTO stock VALUES ('company1', '2023-07-10', 130); +INSERT INTO stock VALUES ('company2', '2023-07-01', 50); +INSERT INTO stock VALUES ('company2', '2023-07-02', 2000); +INSERT INTO stock VALUES ('company2', '2023-07-03', 1500); +INSERT INTO stock VALUES ('company2', '2023-07-04', 1400); +INSERT INTO stock VALUES ('company2', '2023-07-05', 1500); +INSERT INTO stock VALUES ('company2', '2023-07-06', 60); +INSERT INTO stock VALUES ('company2', '2023-07-07', 1100); +INSERT INTO stock VALUES ('company2', '2023-07-08', 1300); +INSERT INTO stock VALUES ('company2', '2023-07-09', 1200); +INSERT INTO stock VALUES ('company2', '2023-07-10', 1300); +SELECT * FROM stock; + company | tdate | price +----------+------------+------- + company1 | 07-01-2023 | 100 + company1 | 07-02-2023 | 200 + company1 | 07-03-2023 | 150 + company1 | 07-04-2023 | 140 + company1 | 07-05-2023 | 150 + company1 | 07-06-2023 | 90 + company1 | 07-07-2023 | 110 + company1 | 07-08-2023 | 130 + company1 | 07-09-2023 | 120 + company1 | 07-10-2023 | 130 + company2 | 07-01-2023 | 50 + company2 | 07-02-2023 | 2000 + company2 | 07-03-2023 | 1500 + company2 | 07-04-2023 | 1400 + company2 | 07-05-2023 | 1500 + company2 | 07-06-2023 | 60 + company2 | 07-07-2023 | 1100 + company2 | 07-08-2023 | 1300 + company2 | 07-09-2023 | 1200 + company2 | 07-10-2023 | 1300 +(20 rows) + +-- basic test using PREV +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | rpr +----------+------------+-------+------ + company1 | 07-01-2023 | 100 | 100 + company1 | 07-02-2023 | 200 | + company1 | 07-03-2023 | 150 | + company1 | 07-04-2023 | 140 | 140 + company1 | 07-05-2023 | 150 | + company1 | 07-06-2023 | 90 | 90 + company1 | 07-07-2023 | 110 | 110 + company1 | 07-08-2023 | 130 | + company1 | 07-09-2023 | 120 | + company1 | 07-10-2023 | 130 | + company2 | 07-01-2023 | 50 | 50 + company2 | 07-02-2023 | 2000 | + company2 | 07-03-2023 | 1500 | + company2 | 07-04-2023 | 1400 | 1400 + company2 | 07-05-2023 | 1500 | + company2 | 07-06-2023 | 60 | 60 + company2 | 07-07-2023 | 1100 | 1100 + company2 | 07-08-2023 | 1300 | + company2 | 07-09-2023 | 1200 | + company2 | 07-10-2023 | 1300 | +(20 rows) + +-- the first row start with less than or equal to 100 +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (LOWPRICE UP+ DOWN+) + DEFINE + LOWPRICE AS price <= 100, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | rpr +----------+------------+-------+----- + company1 | 07-01-2023 | 100 | 100 + company1 | 07-02-2023 | 200 | + company1 | 07-03-2023 | 150 | + company1 | 07-04-2023 | 140 | + company1 | 07-05-2023 | 150 | + company1 | 07-06-2023 | 90 | 90 + company1 | 07-07-2023 | 110 | + company1 | 07-08-2023 | 130 | + company1 | 07-09-2023 | 120 | + company1 | 07-10-2023 | 130 | + company2 | 07-01-2023 | 50 | 50 + company2 | 07-02-2023 | 2000 | + company2 | 07-03-2023 | 1500 | + company2 | 07-04-2023 | 1400 | + company2 | 07-05-2023 | 1500 | + company2 | 07-06-2023 | 60 | 60 + company2 | 07-07-2023 | 1100 | + company2 | 07-08-2023 | 1300 | + company2 | 07-09-2023 | 1200 | + company2 | 07-10-2023 | 1300 | +(20 rows) + +-- second row raises 120% +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (LOWPRICE UP+ DOWN+) + DEFINE + LOWPRICE AS price <= 100, + UP AS price > PREV(price) * 1.2, + DOWN AS price < PREV(price) +); + company | tdate | price | rpr +----------+------------+-------+----- + company1 | 07-01-2023 | 100 | 100 + company1 | 07-02-2023 | 200 | + company1 | 07-03-2023 | 150 | + company1 | 07-04-2023 | 140 | + company1 | 07-05-2023 | 150 | + company1 | 07-06-2023 | 90 | + company1 | 07-07-2023 | 110 | + company1 | 07-08-2023 | 130 | + company1 | 07-09-2023 | 120 | + company1 | 07-10-2023 | 130 | + company2 | 07-01-2023 | 50 | 50 + company2 | 07-02-2023 | 2000 | + company2 | 07-03-2023 | 1500 | + company2 | 07-04-2023 | 1400 | + company2 | 07-05-2023 | 1500 | + company2 | 07-06-2023 | 60 | + company2 | 07-07-2023 | 1100 | + company2 | 07-08-2023 | 1300 | + company2 | 07-09-2023 | 1200 | + company2 | 07-10-2023 | 1300 | +(20 rows) + +-- using NEXT +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UPDOWN) + DEFINE + START AS TRUE, + UPDOWN AS price > PREV(price) AND price > NEXT(price) +); + company | tdate | price | rpr +----------+------------+-------+------ + company1 | 07-01-2023 | 100 | 100 + company1 | 07-02-2023 | 200 | + company1 | 07-03-2023 | 150 | + company1 | 07-04-2023 | 140 | 140 + company1 | 07-05-2023 | 150 | + company1 | 07-06-2023 | 90 | + company1 | 07-07-2023 | 110 | 110 + company1 | 07-08-2023 | 130 | + company1 | 07-09-2023 | 120 | + company1 | 07-10-2023 | 130 | + company2 | 07-01-2023 | 50 | 50 + company2 | 07-02-2023 | 2000 | + company2 | 07-03-2023 | 1500 | + company2 | 07-04-2023 | 1400 | 1400 + company2 | 07-05-2023 | 1500 | + company2 | 07-06-2023 | 60 | + company2 | 07-07-2023 | 1100 | 1100 + company2 | 07-08-2023 | 1300 | + company2 | 07-09-2023 | 1200 | + company2 | 07-10-2023 | 1300 | +(20 rows) + +-- row pattern definition variable name must not appear more than once +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price), + UP AS price > PREV(price) +); +ERROR: row pattern definition variable name "up" appears more than once in DEFINE clause +LINE 9: UP AS price > PREV(price), + ^ +-- pattern variable name must appear in DEFINE +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+ END) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); +ERROR: syntax error at or near "END" +LINE 6: PATTERN (START UP+ DOWN+ END) + ^ +-- FRAME must start at current row when row patttern recognition is used +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); +ERROR: FRAME must start at current row when row patttern recognition is used +-- AFTER MATCH SKIP TO PAST LAST ROW is not supported +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO PAST LAST ROW + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); +ERROR: syntax error at or near "PAST" +LINE 5: AFTER MATCH SKIP TO PAST LAST ROW + ^ +-- SEEK is not supported +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO NEXT ROW + SEEK + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); +ERROR: SEEK is not supported +LINE 6: SEEK + ^ +HINT: Use INITIAL. diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule index cf46fa3359..ebb741318a 100644 --- a/src/test/regress/parallel_schedule +++ b/src/test/regress/parallel_schedule @@ -98,7 +98,7 @@ test: publication subscription # Another group of parallel tests # select_views depends on create_view # ---------- -test: select_views portals_p2 foreign_key cluster dependency guc bitmapops combocid tsearch tsdicts foreign_data window xmlmap functional_deps advisory_lock indirect_toast equivclass +test: select_views portals_p2 foreign_key cluster dependency guc bitmapops combocid tsearch tsdicts foreign_data window xmlmap functional_deps advisory_lock indirect_toast equivclass rpr # ---------- # Another group of parallel tests (JSON related) diff --git a/src/test/regress/sql/rpr.sql b/src/test/regress/sql/rpr.sql new file mode 100644 index 0000000000..921a9fcdfa --- /dev/null +++ b/src/test/regress/sql/rpr.sql @@ -0,0 +1,150 @@ +-- +-- Test for row pattern definition clause +-- + +CREATE TEMP TABLE stock ( + company TEXT, + tdate DATE, + price INTEGER + ); +INSERT INTO stock VALUES ('company1', '2023-07-01', 100); +INSERT INTO stock VALUES ('company1', '2023-07-02', 200); +INSERT INTO stock VALUES ('company1', '2023-07-03', 150); +INSERT INTO stock VALUES ('company1', '2023-07-04', 140); +INSERT INTO stock VALUES ('company1', '2023-07-05', 150); +INSERT INTO stock VALUES ('company1', '2023-07-06', 90); +INSERT INTO stock VALUES ('company1', '2023-07-07', 110); +INSERT INTO stock VALUES ('company1', '2023-07-08', 130); +INSERT INTO stock VALUES ('company1', '2023-07-09', 120); +INSERT INTO stock VALUES ('company1', '2023-07-10', 130); +INSERT INTO stock VALUES ('company2', '2023-07-01', 50); +INSERT INTO stock VALUES ('company2', '2023-07-02', 2000); +INSERT INTO stock VALUES ('company2', '2023-07-03', 1500); +INSERT INTO stock VALUES ('company2', '2023-07-04', 1400); +INSERT INTO stock VALUES ('company2', '2023-07-05', 1500); +INSERT INTO stock VALUES ('company2', '2023-07-06', 60); +INSERT INTO stock VALUES ('company2', '2023-07-07', 1100); +INSERT INTO stock VALUES ('company2', '2023-07-08', 1300); +INSERT INTO stock VALUES ('company2', '2023-07-09', 1200); +INSERT INTO stock VALUES ('company2', '2023-07-10', 1300); + +SELECT * FROM stock; + +-- basic test using PREV +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- the first row start with less than or equal to 100 +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (LOWPRICE UP+ DOWN+) + DEFINE + LOWPRICE AS price <= 100, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- second row raises 120% +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (LOWPRICE UP+ DOWN+) + DEFINE + LOWPRICE AS price <= 100, + UP AS price > PREV(price) * 1.2, + DOWN AS price < PREV(price) +); + +-- using NEXT +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UPDOWN) + DEFINE + START AS TRUE, + UPDOWN AS price > PREV(price) AND price > NEXT(price) +); + +-- row pattern definition variable name must not appear more than once +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price), + UP AS price > PREV(price) +); + +-- pattern variable name must appear in DEFINE +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+ END) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- FRAME must start at current row when row patttern recognition is used +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- AFTER MATCH SKIP TO PAST LAST ROW is not supported +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO PAST LAST ROW + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- SEEK is not supported +SELECT company, tdate, price, rpr(price) OVER w FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO NEXT ROW + SEEK + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); -- 2.25.1 ----Next_Part(Sun_Jun_25_21_05_09_2023_126)---- ^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH v4 1/1] Adjust extra lines generated by psql to be valid SQL comments. @ 2023-07-26 21:35 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Nathan Bossart @ 2023-07-26 21:35 UTC (permalink / raw) psql's --echo-hidden, --log-file, and --single-step options generate extra lines to clearly separate queries from other output. Presently, these extra lines are not valid SQL comments, which makes them a hazard for anyone trying to copy/paste these decorated queries into a client. This commit replaces the starting and ending asterisks in these extra lines with forward slashes so that they are valid SQL comments that can be copy/pasted without incident. Author: Kirk Wolak Reviewed-by: Pavel Stehule, Laurenz Albe, Tom Lane, �lvaro Herrera, Andrey Borodin Discussion: https://postgr.es/m/CACLU5mTFJRJYtbvmZ26txGgmXWQo0hkGhH2o3hEquUPmSbGtBw%40mail.gmail.com --- src/bin/psql/command.c | 8 ++++---- src/bin/psql/common.c | 16 ++++++++-------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c index 6733f008fd..1300869d79 100644 --- a/src/bin/psql/command.c +++ b/src/bin/psql/command.c @@ -5395,16 +5395,16 @@ echo_hidden_command(const char *query) { if (pset.echo_hidden != PSQL_ECHO_HIDDEN_OFF) { - printf(_("********* QUERY **********\n" + printf(_("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(stdout); if (pset.logfile) { fprintf(pset.logfile, - _("********* QUERY **********\n" + _("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(pset.logfile); } diff --git a/src/bin/psql/common.c b/src/bin/psql/common.c index 5973df2e39..10ad1f2538 100644 --- a/src/bin/psql/common.c +++ b/src/bin/psql/common.c @@ -589,16 +589,16 @@ PSQLexec(const char *query) if (pset.echo_hidden != PSQL_ECHO_HIDDEN_OFF) { - printf(_("********* QUERY **********\n" + printf(_("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(stdout); if (pset.logfile) { fprintf(pset.logfile, - _("********* QUERY **********\n" + _("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(pset.logfile); } @@ -1060,9 +1060,9 @@ SendQuery(const char *query) char buf[3]; fflush(stderr); - printf(_("***(Single step mode: verify command)*******************************************\n" + printf(_("/**(Single step mode: verify command)******************************************/\n" "%s\n" - "***(press return to proceed or enter x and return to cancel)********************\n"), + "/**(press return to proceed or enter x and return to cancel)*******************/\n"), query); fflush(stdout); if (fgets(buf, sizeof(buf), stdin) != NULL) @@ -1080,9 +1080,9 @@ SendQuery(const char *query) if (pset.logfile) { fprintf(pset.logfile, - _("********* QUERY **********\n" + _("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(pset.logfile); } -- 2.25.1 --UlVJffcvxoiEqYs2-- ^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH v4 1/1] Adjust extra lines generated by psql to be valid SQL comments. @ 2023-07-26 21:35 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Nathan Bossart @ 2023-07-26 21:35 UTC (permalink / raw) psql's --echo-hidden, --log-file, and --single-step options generate extra lines to clearly separate queries from other output. Presently, these extra lines are not valid SQL comments, which makes them a hazard for anyone trying to copy/paste these decorated queries into a client. This commit replaces the starting and ending asterisks in these extra lines with forward slashes so that they are valid SQL comments that can be copy/pasted without incident. Author: Kirk Wolak Reviewed-by: Pavel Stehule, Laurenz Albe, Tom Lane, �lvaro Herrera, Andrey Borodin Discussion: https://postgr.es/m/CACLU5mTFJRJYtbvmZ26txGgmXWQo0hkGhH2o3hEquUPmSbGtBw%40mail.gmail.com --- src/bin/psql/command.c | 8 ++++---- src/bin/psql/common.c | 16 ++++++++-------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c index 6733f008fd..1300869d79 100644 --- a/src/bin/psql/command.c +++ b/src/bin/psql/command.c @@ -5395,16 +5395,16 @@ echo_hidden_command(const char *query) { if (pset.echo_hidden != PSQL_ECHO_HIDDEN_OFF) { - printf(_("********* QUERY **********\n" + printf(_("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(stdout); if (pset.logfile) { fprintf(pset.logfile, - _("********* QUERY **********\n" + _("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(pset.logfile); } diff --git a/src/bin/psql/common.c b/src/bin/psql/common.c index 5973df2e39..10ad1f2538 100644 --- a/src/bin/psql/common.c +++ b/src/bin/psql/common.c @@ -589,16 +589,16 @@ PSQLexec(const char *query) if (pset.echo_hidden != PSQL_ECHO_HIDDEN_OFF) { - printf(_("********* QUERY **********\n" + printf(_("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(stdout); if (pset.logfile) { fprintf(pset.logfile, - _("********* QUERY **********\n" + _("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(pset.logfile); } @@ -1060,9 +1060,9 @@ SendQuery(const char *query) char buf[3]; fflush(stderr); - printf(_("***(Single step mode: verify command)*******************************************\n" + printf(_("/**(Single step mode: verify command)******************************************/\n" "%s\n" - "***(press return to proceed or enter x and return to cancel)********************\n"), + "/**(press return to proceed or enter x and return to cancel)*******************/\n"), query); fflush(stdout); if (fgets(buf, sizeof(buf), stdin) != NULL) @@ -1080,9 +1080,9 @@ SendQuery(const char *query) if (pset.logfile) { fprintf(pset.logfile, - _("********* QUERY **********\n" + _("/******** QUERY *********/\n" "%s\n" - "**************************\n\n"), query); + "/************************/\n\n"), query); fflush(pset.logfile); } -- 2.25.1 --UlVJffcvxoiEqYs2-- ^ permalink raw reply [nested|flat] 14+ messages in thread
end of thread, other threads:[~2023-07-26 21:35 UTC | newest] Thread overview: 14+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2018-10-11 01:03 [PATCH 2/4] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]> 2018-10-11 01:03 [PATCH 2/4] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]> 2018-10-11 01:03 [PATCH 2/4] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]> 2018-10-11 01:03 [PATCH 2/4] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]> 2018-10-11 01:03 [PATCH 3/3] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]> 2023-01-04 05:27 Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> 2023-01-04 18:12 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-01-05 00:49 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-01-05 05:27 ` Re: wake up logical workers after ALTER SUBSCRIPTION Amit Kapila <[email protected]> 2023-01-05 17:29 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-01-06 00:00 ` Re: wake up logical workers after ALTER SUBSCRIPTION Nathan Bossart <[email protected]> 2023-06-25 11:48 [PATCH v1 6/7] Row pattern recognition patch (tests). Tatsuo Ishii <[email protected]> 2023-07-26 21:35 [PATCH v4 1/1] Adjust extra lines generated by psql to be valid SQL comments. Nathan Bossart <[email protected]> 2023-07-26 21:35 [PATCH v4 1/1] Adjust extra lines generated by psql to be valid SQL comments. Nathan Bossart <[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