public inbox for [email protected]help / color / mirror / Atom feed
RE: Logical replication timeout problem 14+ messages / 6 participants [nested] [flat]
* RE: Logical replication timeout problem @ 2022-04-28 10:01 [email protected] <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: [email protected] @ 2022-04-28 10:01 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; +Cc: Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Wednesday, April 20, 2022 3:21 PM Masahiko Sawada <[email protected]> wrote: > > BTW the changes in > REL_14_v1-0001-Fix-the-logical-replication-timeout-during-large-.patch, > adding end_xact to LogicalDecodingContext, seems good to me and it > might be better than the approach of v17 patch from plugin developers’ > perspective? This is because they won’t need to pass true/false to > end_xact of OutputPluginUpdateProgress(). Furthermore, if we do what > we do in update_replication_progress() in > OutputPluginUpdateProgress(), what plugins need to do will be just to > call OutputPluginUpdate() in every callback and they don't need to > have the CHANGES_THRESHOLD logic. What do you think? Hi Sawada-san, Wang I was looking at the patch and noticed that we moved some logic from update_replication_progress() to OutputPluginUpdateProgress() like your suggestion. I have a question about this change. In the patch we added some restriction in this function OutputPluginUpdateProgress() like below: + /* + * If we are at the end of transaction LSN, update progress tracking. + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) + { + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, + skipped_xact); + changes_count = 0; + } After the patch, we won't be able to always invoke the update_progress() if the caller are at the middle of transaction(e.g. end_xact = false). The bebavior of the public function OutputPluginUpdateProgress() is changed. I am thinking is it ok to change this at back-branches ? Because OutputPluginUpdateProgress() is a public function for the extension developer, just a little concerned about the behavior change here. Besides, the check of 'end_xact' and the 'CHANGES_THRESHOLD' seems specified to the pgoutput. I am not very sure that if plugin author also needs these logic(they might want to change the strategy), so I'd like to confirm it with you. Best regards, Hou zj ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-02 02:03 Masahiko Sawada <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Masahiko Sawada @ 2022-05-02 02:03 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: [email protected] <[email protected]>; Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Thu, Apr 28, 2022 at 7:01 PM [email protected] <[email protected]> wrote: > > On Wednesday, April 20, 2022 3:21 PM Masahiko Sawada <[email protected]> wrote: > > > > BTW the changes in > > REL_14_v1-0001-Fix-the-logical-replication-timeout-during-large-.patch, > > adding end_xact to LogicalDecodingContext, seems good to me and it > > might be better than the approach of v17 patch from plugin developers’ > > perspective? This is because they won’t need to pass true/false to > > end_xact of OutputPluginUpdateProgress(). Furthermore, if we do what > > we do in update_replication_progress() in > > OutputPluginUpdateProgress(), what plugins need to do will be just to > > call OutputPluginUpdate() in every callback and they don't need to > > have the CHANGES_THRESHOLD logic. What do you think? > > Hi Sawada-san, Wang > > I was looking at the patch and noticed that we moved some logic from > update_replication_progress() to OutputPluginUpdateProgress() like > your suggestion. > > I have a question about this change. In the patch we added some > restriction in this function OutputPluginUpdateProgress() like below: > > + /* > + * If we are at the end of transaction LSN, update progress tracking. > + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we > + * try to send a keepalive message if required. > + */ > + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) > + { > + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, > + skipped_xact); > + changes_count = 0; > + } > > After the patch, we won't be able to always invoke the update_progress() if the > caller are at the middle of transaction(e.g. end_xact = false). The bebavior of the > public function OutputPluginUpdateProgress() is changed. I am thinking is it ok to > change this at back-branches ? > > Because OutputPluginUpdateProgress() is a public function for the extension > developer, just a little concerned about the behavior change here. Good point. As you pointed out, it would not be good if we change the behavior of OutputPluginUpdateProgress() in back branches. Also, after more thought, it is not a good idea even for HEAD since there might be background workers that use logical decoding and the timeout checking might not be relevant at all with them. BTW, I think you're concerned about the plugins that call OutputPluginUpdateProgress() at the middle of the transaction (i.e., end_xact = false). We have the following change that makes the walsender not update the progress at the middle of the transaction. Do you think it is okay since it's not common usage to call OutputPluginUpdateProgress() at the middle of the transaction by the plugin that is used by the walsender? #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, + if (end_xact && TimestampDifferenceExceeds(sendTime, now, WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + { + LagTrackerWrite(lsn, now); + sendTime = now; + } Regards, -- Masahiko Sawada EDB: https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-02 02:32 Amit Kapila <[email protected]> parent: Masahiko Sawada <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Amit Kapila @ 2022-05-02 02:32 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, May 2, 2022 at 7:33 AM Masahiko Sawada <[email protected]> wrote: > On Thu, Apr 28, 2022 at 7:01 PM [email protected] > <[email protected]> wrote: > > > > Hi Sawada-san, Wang > > > > I was looking at the patch and noticed that we moved some logic from > > update_replication_progress() to OutputPluginUpdateProgress() like > > your suggestion. > > > > I have a question about this change. In the patch we added some > > restriction in this function OutputPluginUpdateProgress() like below: > > > > + /* > > + * If we are at the end of transaction LSN, update progress tracking. > > + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we > > + * try to send a keepalive message if required. > > + */ > > + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) > > + { > > + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, > > + skipped_xact); > > + changes_count = 0; > > + } > > > > After the patch, we won't be able to always invoke the update_progress() if the > > caller are at the middle of transaction(e.g. end_xact = false). The bebavior of the > > public function OutputPluginUpdateProgress() is changed. I am thinking is it ok to > > change this at back-branches ? > > > > Because OutputPluginUpdateProgress() is a public function for the extension > > developer, just a little concerned about the behavior change here. > > Good point. > > As you pointed out, it would not be good if we change the behavior of > OutputPluginUpdateProgress() in back branches. Also, after more > thought, it is not a good idea even for HEAD since there might be > background workers that use logical decoding and the timeout checking > might not be relevant at all with them. > So, shall we go back to the previous approach of using a separate function update_replication_progress? > BTW, I think you're concerned about the plugins that call > OutputPluginUpdateProgress() at the middle of the transaction (i.e., > end_xact = false). We have the following change that makes the > walsender not update the progress at the middle of the transaction. Do > you think it is okay since it's not common usage to call > OutputPluginUpdateProgress() at the middle of the transaction by the > plugin that is used by the walsender? > We have done that purposefully as otherwise, the lag tracker shows incorrect information. See email [1]. The reason is that we always get ack from subscribers for transaction end. Also, prior to this patch we never call the lag tracker recording apart from the transaction end, so as a bug fix we shouldn't try to change it. [1] - https://www.postgresql.org/message-id/OS3PR01MB62755D216245199554DDC8DB9EEA9%40OS3PR01MB6275.jpnprd0... -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-02 02:36 Masahiko Sawada <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Masahiko Sawada @ 2022-05-02 02:36 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, May 2, 2022 at 11:32 AM Amit Kapila <[email protected]> wrote: > > On Mon, May 2, 2022 at 7:33 AM Masahiko Sawada <[email protected]> wrote: > > On Thu, Apr 28, 2022 at 7:01 PM [email protected] > > <[email protected]> wrote: > > > > > > Hi Sawada-san, Wang > > > > > > I was looking at the patch and noticed that we moved some logic from > > > update_replication_progress() to OutputPluginUpdateProgress() like > > > your suggestion. > > > > > > I have a question about this change. In the patch we added some > > > restriction in this function OutputPluginUpdateProgress() like below: > > > > > > + /* > > > + * If we are at the end of transaction LSN, update progress tracking. > > > + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we > > > + * try to send a keepalive message if required. > > > + */ > > > + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) > > > + { > > > + ctx->update_progress(ctx, ctx->write_location, ctx->write_xid, > > > + skipped_xact); > > > + changes_count = 0; > > > + } > > > > > > After the patch, we won't be able to always invoke the update_progress() if the > > > caller are at the middle of transaction(e.g. end_xact = false). The bebavior of the > > > public function OutputPluginUpdateProgress() is changed. I am thinking is it ok to > > > change this at back-branches ? > > > > > > Because OutputPluginUpdateProgress() is a public function for the extension > > > developer, just a little concerned about the behavior change here. > > > > Good point. > > > > As you pointed out, it would not be good if we change the behavior of > > OutputPluginUpdateProgress() in back branches. Also, after more > > thought, it is not a good idea even for HEAD since there might be > > background workers that use logical decoding and the timeout checking > > might not be relevant at all with them. > > > > So, shall we go back to the previous approach of using a separate > function update_replication_progress? Ok, agreed. > > > BTW, I think you're concerned about the plugins that call > > OutputPluginUpdateProgress() at the middle of the transaction (i.e., > > end_xact = false). We have the following change that makes the > > walsender not update the progress at the middle of the transaction. Do > > you think it is okay since it's not common usage to call > > OutputPluginUpdateProgress() at the middle of the transaction by the > > plugin that is used by the walsender? > > > > We have done that purposefully as otherwise, the lag tracker shows > incorrect information. See email [1]. The reason is that we always get > ack from subscribers for transaction end. Also, prior to this patch we > never call the lag tracker recording apart from the transaction end, > so as a bug fix we shouldn't try to change it. Make sense. Regards, -- Masahiko Sawada EDB: https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-04 10:18 Amit Kapila <[email protected]> parent: Masahiko Sawada <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Amit Kapila @ 2022-05-04 10:18 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, May 2, 2022 at 8:07 AM Masahiko Sawada <[email protected]> wrote: > > On Mon, May 2, 2022 at 11:32 AM Amit Kapila <[email protected]> wrote: > > > > > > So, shall we go back to the previous approach of using a separate > > function update_replication_progress? > > Ok, agreed. > Attached, please find the updated patch accordingly. Currently, I have prepared it for HEAD, if you don't see any problem with this, we can prepare the back-branch patches based on this. -- With Regards, Amit Kapila. Attachments: [application/octet-stream] v20-0001-Fix-the-logical-replication-timeout-during-large.patch (14.5K, ../../CAA4eK1JuhAuG-SzxCXi5YMhpKvGT_-boJT61e7Z14qfp6AwG6g@mail.gmail.com/2-v20-0001-Fix-the-logical-replication-timeout-during-large.patch) download | inline diff: From 2014eb81689d16baa170b0aa1cde965d6f1f089d Mon Sep 17 00:00:00 2001 From: Amit Kapila <[email protected]> Date: Wed, 4 May 2022 11:41:59 +0530 Subject: [PATCH v20] Fix the logical replication timeout during large transactions. The problem is that we don't send keep-alive messages for a long time while processing large transactions during logical replication where we don't send any data of such transactions. This can happen when the table modified in the transaction is not published or because all the changes got filtered. We do try to send the keep_alive if necessary at the end of the transaction (via WalSndWriteData()) but by that time the subscriber-side can timeout and exit. To fix this we try to send the keepalive message if required after processing certain threshold of changes. --- src/backend/replication/logical/logical.c | 27 +++++++++++ src/backend/replication/pgoutput/pgoutput.c | 54 ++++++++++++++++++--- src/backend/replication/walsender.c | 24 +++++++-- src/include/replication/logical.h | 2 + 4 files changed, 98 insertions(+), 9 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 788769dd73..625a7f4273 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -746,6 +746,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.startup_cb(ctx, opt, is_init); @@ -773,6 +774,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.shutdown_cb(ctx); @@ -808,6 +810,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.begin_cb(ctx, txn); @@ -839,6 +842,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* do the actual work: call callback */ ctx->callbacks.commit_cb(ctx, txn, commit_lsn); @@ -879,6 +883,7 @@ begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* * If the plugin supports two-phase commits then begin prepare callback is @@ -923,6 +928,7 @@ prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin supports two-phase commits then prepare callback is @@ -967,6 +973,7 @@ commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin support two-phase commits then commit prepared callback @@ -1012,6 +1019,7 @@ rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin support two-phase commits then rollback prepared callback @@ -1062,6 +1070,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.change_cb(ctx, txn, relation, change); /* Pop the error context stack */ @@ -1102,6 +1112,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ @@ -1129,6 +1141,7 @@ filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid); @@ -1159,6 +1172,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id); @@ -1196,6 +1210,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, @@ -1239,6 +1254,8 @@ stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = first_lsn; + ctx->end_xact = false; + /* in streaming mode, stream_start_cb is required */ if (ctx->callbacks.stream_start_cb == NULL) ereport(ERROR, @@ -1286,6 +1303,8 @@ stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = last_lsn; + ctx->end_xact = false; + /* in streaming mode, stream_stop_cb is required */ if (ctx->callbacks.stream_stop_cb == NULL) ereport(ERROR, @@ -1325,6 +1344,7 @@ stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = abort_lsn; + ctx->end_xact = true; /* in streaming mode, stream_abort_cb is required */ if (ctx->callbacks.stream_abort_cb == NULL) @@ -1369,6 +1389,7 @@ stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; + ctx->end_xact = true; /* in streaming mode with two-phase commits, stream_prepare_cb is required */ if (ctx->callbacks.stream_prepare_cb == NULL) @@ -1409,6 +1430,7 @@ stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; + ctx->end_xact = true; /* in streaming mode, stream_commit_cb is required */ if (ctx->callbacks.stream_commit_cb == NULL) @@ -1457,6 +1479,8 @@ stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + /* in streaming mode, stream_change_cb is required */ if (ctx->callbacks.stream_change_cb == NULL) ereport(ERROR, @@ -1501,6 +1525,7 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix, @@ -1549,6 +1574,8 @@ stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index b197bfd565..edd50b0713 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -91,6 +91,8 @@ static void send_relation_and_attrs(Relation relation, TransactionId xid, static void send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, XLogRecPtr origin_lsn, bool send_origin); +static void update_replication_progress(LogicalDecodingContext* ctx, + bool skipped_xact); /* * Only 3 publication actions are used for row filtering ("insert", "update", @@ -558,7 +560,7 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, * from this transaction has been sent to the downstream. */ sent_begin_txn = txndata->sent_begin_txn; - OutputPluginUpdateProgress(ctx, !sent_begin_txn); + update_replication_progress(ctx, !sent_begin_txn); pfree(txndata); txn->output_plugin_private = NULL; @@ -597,7 +599,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -611,7 +613,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -627,7 +629,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1360,6 +1362,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, TupleTableSlot *old_slot = NULL; TupleTableSlot *new_slot = NULL; + update_replication_progress(ctx, false); + if (!is_publishable_relation(relation)) return; @@ -1592,6 +1596,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Oid *relids; TransactionId xid = InvalidTransactionId; + update_replication_progress(ctx, false); + /* Remember the xid for the change in streaming mode. See pgoutput_change. */ if (in_streaming) xid = change->txn->xid; @@ -1655,6 +1661,8 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; TransactionId xid = InvalidTransactionId; + update_replication_progress(ctx, false); + if (!data->messages) return; @@ -1847,7 +1855,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1868,7 +1876,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); @@ -2361,3 +2369,37 @@ send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, } } } + +/* + * Try to update progress and send a keepalive message if too many changes were + * processed. + * + * For a large transaction, if we don't send any change to the downstream for a + * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. + * This can happen when all or most of the changes are either not published or + * got filtered out. + */ +static void +update_replication_progress(LogicalDecodingContext *ctx, bool skipped_xact) +{ + static int changes_count = 0; + + /* + * We don't want to try sending a keepalive message after processing each + * change as that can have overhead. Tests revealed that there is no + * noticeable overhead in doing it after continuously processing 100 or so + * changes. + */ +#define CHANGES_THRESHOLD 100 + + /* + * If we are at the end of transaction LSN, update progress tracking. + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx, skipped_xact); + changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 63a818140b..79fffe849b 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1482,14 +1482,20 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); + bool pending_writes = false; + bool end_xact = ctx->end_xact; /* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently. + * + * We don't have a mechanism to get the ack for any LSN other than end xact + * LSN from the downstream. So, we track lag only for end of transaction + * LSN. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + if (end_xact && TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) { LagTrackerWrite(lsn, now); sendTime = now; @@ -1515,8 +1521,20 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId /* If we have pending write here, make sure it's actually flushed */ if (pq_is_send_pending()) - ProcessPendingWrites(); + pending_writes = true; } + + /* + * Process pending writes if any or try to send a keepalive if required. We + * don't need to try sending keep alive messages at the transaction end as + * that will be done at a later point in time. This is required only for + * large transactions where we don't send any changes to the downstream and + * the receiver can timeout due to that. + */ + if (pending_writes || (!end_xact && + now >= TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2))) + ProcessPendingWrites(); } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index a6ef16ad5b..edadacd589 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -107,6 +107,8 @@ typedef struct LogicalDecodingContext bool prepared_write; XLogRecPtr write_location; TransactionId write_xid; + /* Are we processing the end LSN of a transaction? */ + bool end_xact; } LogicalDecodingContext; -- 2.28.0.windows.1 ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-06 01:53 Masahiko Sawada <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Masahiko Sawada @ 2022-05-06 01:53 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Wed, May 4, 2022 at 7:18 PM Amit Kapila <[email protected]> wrote: > > On Mon, May 2, 2022 at 8:07 AM Masahiko Sawada <[email protected]> wrote: > > > > On Mon, May 2, 2022 at 11:32 AM Amit Kapila <[email protected]> wrote: > > > > > > > > > So, shall we go back to the previous approach of using a separate > > > function update_replication_progress? > > > > Ok, agreed. > > > > Attached, please find the updated patch accordingly. Currently, I have > prepared it for HEAD, if you don't see any problem with this, we can > prepare the back-branch patches based on this. Thank you for updating the patch. Looks good to me. Regards, -- Masahiko Sawada EDB: https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 14+ messages in thread
* RE: Logical replication timeout problem @ 2022-05-06 07:11 [email protected] <[email protected]> parent: Masahiko Sawada <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: [email protected] @ 2022-05-06 07:11 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; Amit Kapila <[email protected]>; +Cc: [email protected] <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, May 6, 2022 at 9:54 AM Masahiko Sawada <[email protected]> wrote: > On Wed, May 4, 2022 at 7:18 PM Amit Kapila <[email protected]> wrote: > > > > On Mon, May 2, 2022 at 8:07 AM Masahiko Sawada <[email protected]> > wrote: > > > > > > On Mon, May 2, 2022 at 11:32 AM Amit Kapila <[email protected]> > wrote: > > > > > > > > > > > > So, shall we go back to the previous approach of using a separate > > > > function update_replication_progress? > > > > > > Ok, agreed. > > > > > > > Attached, please find the updated patch accordingly. Currently, I have > > prepared it for HEAD, if you don't see any problem with this, we can > > prepare the back-branch patches based on this. > > Thank you for updating the patch. Looks good to me. Thanks for your review. Improve the back-branch patches according to the discussion. Move the CHANGES_THRESHOLD logic from function OutputPluginUpdateProgress to new funcion update_replication_progress. In addition, improve all patches formatting with pgindent. Attach the patches. Regards, Wang wei Attachments: [application/octet-stream] HEAD_v21-0001-Fix-the-logical-replication-timeout-during-large.patch (14.5K, ../../OS3PR01MB62754AA446A2FB5B72816C529EC59@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-HEAD_v21-0001-Fix-the-logical-replication-timeout-during-large.patch) download | inline diff: From a4b30e8e30bcebef289489d98a311b623a7007bc Mon Sep 17 00:00:00 2001 From: Amit Kapila <[email protected]> Date: Wed, 4 May 2022 11:41:59 +0530 Subject: [PATCH v21] Fix the logical replication timeout during large transactions. The problem is that we don't send keep-alive messages for a long time while processing large transactions during logical replication where we don't send any data of such transactions. This can happen when the table modified in the transaction is not published or because all the changes got filtered. We do try to send the keep_alive if necessary at the end of the transaction (via WalSndWriteData()) but by that time the subscriber-side can timeout and exit. To fix this we try to send the keepalive message if required after processing certain threshold of changes. --- src/backend/replication/logical/logical.c | 27 +++++++++++ src/backend/replication/pgoutput/pgoutput.c | 54 ++++++++++++++++++--- src/backend/replication/walsender.c | 24 +++++++-- src/include/replication/logical.h | 2 + 4 files changed, 98 insertions(+), 9 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 788769dd73..625a7f4273 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -746,6 +746,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.startup_cb(ctx, opt, is_init); @@ -773,6 +774,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.shutdown_cb(ctx); @@ -808,6 +810,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.begin_cb(ctx, txn); @@ -839,6 +842,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* do the actual work: call callback */ ctx->callbacks.commit_cb(ctx, txn, commit_lsn); @@ -879,6 +883,7 @@ begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* * If the plugin supports two-phase commits then begin prepare callback is @@ -923,6 +928,7 @@ prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin supports two-phase commits then prepare callback is @@ -967,6 +973,7 @@ commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin support two-phase commits then commit prepared callback @@ -1012,6 +1019,7 @@ rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin support two-phase commits then rollback prepared callback @@ -1062,6 +1070,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.change_cb(ctx, txn, relation, change); /* Pop the error context stack */ @@ -1102,6 +1112,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ @@ -1129,6 +1141,7 @@ filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid); @@ -1159,6 +1172,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id); @@ -1196,6 +1210,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, @@ -1239,6 +1254,8 @@ stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = first_lsn; + ctx->end_xact = false; + /* in streaming mode, stream_start_cb is required */ if (ctx->callbacks.stream_start_cb == NULL) ereport(ERROR, @@ -1286,6 +1303,8 @@ stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = last_lsn; + ctx->end_xact = false; + /* in streaming mode, stream_stop_cb is required */ if (ctx->callbacks.stream_stop_cb == NULL) ereport(ERROR, @@ -1325,6 +1344,7 @@ stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = abort_lsn; + ctx->end_xact = true; /* in streaming mode, stream_abort_cb is required */ if (ctx->callbacks.stream_abort_cb == NULL) @@ -1369,6 +1389,7 @@ stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; + ctx->end_xact = true; /* in streaming mode with two-phase commits, stream_prepare_cb is required */ if (ctx->callbacks.stream_prepare_cb == NULL) @@ -1409,6 +1430,7 @@ stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; + ctx->end_xact = true; /* in streaming mode, stream_commit_cb is required */ if (ctx->callbacks.stream_commit_cb == NULL) @@ -1457,6 +1479,8 @@ stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + /* in streaming mode, stream_change_cb is required */ if (ctx->callbacks.stream_change_cb == NULL) ereport(ERROR, @@ -1501,6 +1525,7 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix, @@ -1549,6 +1574,8 @@ stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index b197bfd565..406ad84e1d 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -91,6 +91,8 @@ static void send_relation_and_attrs(Relation relation, TransactionId xid, static void send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, XLogRecPtr origin_lsn, bool send_origin); +static void update_replication_progress(LogicalDecodingContext *ctx, + bool skipped_xact); /* * Only 3 publication actions are used for row filtering ("insert", "update", @@ -558,7 +560,7 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, * from this transaction has been sent to the downstream. */ sent_begin_txn = txndata->sent_begin_txn; - OutputPluginUpdateProgress(ctx, !sent_begin_txn); + update_replication_progress(ctx, !sent_begin_txn); pfree(txndata); txn->output_plugin_private = NULL; @@ -597,7 +599,7 @@ static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn) { - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_prepare(ctx->out, txn, prepare_lsn); @@ -611,7 +613,7 @@ static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); @@ -627,7 +629,7 @@ pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time) { - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, @@ -1360,6 +1362,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, TupleTableSlot *old_slot = NULL; TupleTableSlot *new_slot = NULL; + update_replication_progress(ctx, false); + if (!is_publishable_relation(relation)) return; @@ -1592,6 +1596,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Oid *relids; TransactionId xid = InvalidTransactionId; + update_replication_progress(ctx, false); + /* Remember the xid for the change in streaming mode. See pgoutput_change. */ if (in_streaming) xid = change->txn->xid; @@ -1655,6 +1661,8 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; TransactionId xid = InvalidTransactionId; + update_replication_progress(ctx, false); + if (!data->messages) return; @@ -1847,7 +1855,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1868,7 +1876,7 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, { Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx, false); + update_replication_progress(ctx, false); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn); OutputPluginWrite(ctx, true); @@ -2361,3 +2369,37 @@ send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, } } } + +/* + * Try to update progress and send a keepalive message if too many changes were + * processed. + * + * For a large transaction, if we don't send any change to the downstream for a + * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. + * This can happen when all or most of the changes are either not published or + * got filtered out. + */ +static void +update_replication_progress(LogicalDecodingContext *ctx, bool skipped_xact) +{ + static int changes_count = 0; + + /* + * We don't want to try sending a keepalive message after processing each + * change as that can have overhead. Tests revealed that there is no + * noticeable overhead in doing it after continuously processing 100 or so + * changes. + */ +#define CHANGES_THRESHOLD 100 + + /* + * If we are at the end of transaction LSN, update progress tracking. + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx, skipped_xact); + changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 63a818140b..c6c196b2fa 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1482,14 +1482,20 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); + bool pending_writes = false; + bool end_xact = ctx->end_xact; /* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently. + * + * We don't have a mechanism to get the ack for any LSN other than end + * xact LSN from the downstream. So, we track lag only for end of + * transaction LSN. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + if (end_xact && TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) { LagTrackerWrite(lsn, now); sendTime = now; @@ -1515,8 +1521,20 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId /* If we have pending write here, make sure it's actually flushed */ if (pq_is_send_pending()) - ProcessPendingWrites(); + pending_writes = true; } + + /* + * Process pending writes if any or try to send a keepalive if required. + * We don't need to try sending keep alive messages at the transaction end + * as that will be done at a later point in time. This is required only + * for large transactions where we don't send any changes to the + * downstream and the receiver can timeout due to that. + */ + if (pending_writes || (!end_xact && + now >= TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2))) + ProcessPendingWrites(); } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index a6ef16ad5b..edadacd589 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -107,6 +107,8 @@ typedef struct LogicalDecodingContext bool prepared_write; XLogRecPtr write_location; TransactionId write_xid; + /* Are we processing the end LSN of a transaction? */ + bool end_xact; } LogicalDecodingContext; -- 2.23.0.windows.1 [application/octet-stream] REL14_v4-0001-Fix-the-logical-replication-timeout-during-large-.patch (13.4K, ../../OS3PR01MB62754AA446A2FB5B72816C529EC59@OS3PR01MB6275.jpnprd01.prod.outlook.com/3-REL14_v4-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From 12c26ad9dd9957279ccad5cfa74c2acaf7c7d931 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Thu, 21 Apr 2022 09:20:52 +0800 Subject: [PATCH v4] Fix the logical replication timeout during large transactions. The problem is that we don't send keep-alive messages for a long time while processing large transactions during logical replication where we don't send any data of such transactions. This can happen when the table modified in the transaction is not published. We do try to send the keep_alive if necessary at the end of the transaction (via WalSndWriteData()) but by that time the subscriber-side can timeout and exit. To fix this we try to send the keepalive message if required after processing certain threshold of changes. --- src/backend/replication/logical/logical.c | 27 +++++++++++++ src/backend/replication/pgoutput/pgoutput.c | 44 ++++++++++++++++++++- src/backend/replication/walsender.c | 38 +++++++++++++++--- src/include/replication/logical.h | 3 ++ 4 files changed, 105 insertions(+), 7 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index d536a5f3ba..f7d1491907 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -724,6 +724,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.startup_cb(ctx, opt, is_init); @@ -751,6 +752,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.shutdown_cb(ctx); @@ -786,6 +788,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.begin_cb(ctx, txn); @@ -817,6 +820,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* do the actual work: call callback */ ctx->callbacks.commit_cb(ctx, txn, commit_lsn); @@ -857,6 +861,7 @@ begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* * If the plugin supports two-phase commits then begin prepare callback is @@ -901,6 +906,7 @@ prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin supports two-phase commits then prepare callback is @@ -945,6 +951,7 @@ commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin support two-phase commits then commit prepared callback @@ -990,6 +997,7 @@ rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* * If the plugin support two-phase commits then rollback prepared callback @@ -1040,6 +1048,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.change_cb(ctx, txn, relation, change); /* Pop the error context stack */ @@ -1080,6 +1090,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ @@ -1107,6 +1119,7 @@ filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid); @@ -1137,6 +1150,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id); @@ -1174,6 +1188,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, @@ -1217,6 +1232,8 @@ stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = first_lsn; + ctx->end_xact = false; + /* in streaming mode, stream_start_cb is required */ if (ctx->callbacks.stream_start_cb == NULL) ereport(ERROR, @@ -1264,6 +1281,8 @@ stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = last_lsn; + ctx->end_xact = false; + /* in streaming mode, stream_stop_cb is required */ if (ctx->callbacks.stream_stop_cb == NULL) ereport(ERROR, @@ -1303,6 +1322,7 @@ stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = abort_lsn; + ctx->end_xact = true; /* in streaming mode, stream_abort_cb is required */ if (ctx->callbacks.stream_abort_cb == NULL) @@ -1347,6 +1367,7 @@ stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; + ctx->end_xact = true; /* in streaming mode with two-phase commits, stream_prepare_cb is required */ if (ctx->callbacks.stream_prepare_cb == NULL) @@ -1387,6 +1408,7 @@ stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; + ctx->end_xact = true; /* in streaming mode, stream_commit_cb is required */ if (ctx->callbacks.stream_commit_cb == NULL) @@ -1435,6 +1457,8 @@ stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + /* in streaming mode, stream_change_cb is required */ if (ctx->callbacks.stream_change_cb == NULL) ereport(ERROR, @@ -1479,6 +1503,7 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix, @@ -1527,6 +1552,8 @@ stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 2ea540ce4d..ff9cf5d406 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -70,6 +70,7 @@ static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); static void send_relation_and_attrs(Relation relation, TransactionId xid, LogicalDecodingContext *ctx); +static void update_replication_progress(LogicalDecodingContext *ctx); /* * Entry in the map used to remember which relation schemas we sent. @@ -381,7 +382,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + update_replication_progress(ctx); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -535,6 +536,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, TransactionId xid = InvalidTransactionId; Relation ancestor = NULL; + update_replication_progress(ctx); + if (!is_publishable_relation(relation)) return; @@ -677,6 +680,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Oid *relids; TransactionId xid = InvalidTransactionId; + update_replication_progress(ctx); + /* Remember the xid for the change in streaming mode. See pgoutput_change. */ if (in_streaming) xid = change->txn->xid; @@ -735,6 +740,8 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; TransactionId xid = InvalidTransactionId; + update_replication_progress(ctx); + if (!data->messages) return; @@ -921,7 +928,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx, Assert(!in_streaming); Assert(rbtxn_is_streamed(txn)); - OutputPluginUpdateProgress(ctx); + update_replication_progress(ctx); OutputPluginPrepareWrite(ctx, true); logicalrep_write_stream_commit(ctx->out, txn, commit_lsn); @@ -1304,3 +1311,36 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue) entry->pubactions.pubtruncate = false; } } + +/* + * Try to update progress and send a keepalive message if too many changes were + * processed. + * + * For a large transaction, if we don't send any change to the downstream for a + * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. + * This can happen when all or most of the changes are not published. + */ +static void +update_replication_progress(LogicalDecodingContext *ctx) +{ + static int changes_count = 0; + + /* + * We don't want to try sending a keepalive message after processing each + * change as that can have overhead. Tests revealed that there is no + * noticeable overhead in doing it after continuously processing 100 or so + * changes. + */ +#define CHANGES_THRESHOLD 100 + + /* + * If we are at the end of transaction LSN, update progress tracking. + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx); + changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 3b245c619f..28f0a29473 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -240,6 +240,7 @@ static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); static void ProcessRepliesIfAny(void); +static void ProcessPendingWrites(void); static void WalSndKeepalive(bool requestReply); static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); @@ -1288,6 +1289,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, } /* If we have pending write here, go to slow path */ + ProcessPendingWrites(); +} + +/* + * Wait until there is no pending write. Also process replies from the other + * side and check timeouts during that. + */ +static void +ProcessPendingWrites(void) +{ for (;;) { long sleeptime; @@ -1342,18 +1353,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); + bool end_xact = ctx->end_xact; /* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently. + * + * We don't have a mechanism to get the ack for any LSN other than end + * xact LSN from the downstream. So, we track lag only for end of + * transaction LSN. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (end_xact && TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* + * Try to send a keepalive if required. We don't need to try sending keep + * alive messages at the transaction end as that will be done at a later + * point in time. This is required only for large transactions where we + * don't send any changes to the downstream and the receiver can timeout + * due to that. + */ + if (!end_xact && + now >= TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + ProcessPendingWrites(); } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index af551d6f4e..25490f221f 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -49,6 +49,9 @@ typedef struct LogicalDecodingContext */ bool fast_forward; + /* Are we processing the end LSN of a transaction? */ + bool end_xact; + OutputPluginCallbacks callbacks; OutputPluginOptions options; -- 2.18.4 [application/octet-stream] REL13_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch (8.9K, ../../OS3PR01MB62754AA446A2FB5B72816C529EC59@OS3PR01MB6275.jpnprd01.prod.outlook.com/4-REL13_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From bd7e63bce532d226f4f40de2143ca8d286c7cdb1 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Thu, 21 Apr 2022 15:01:02 +0800 Subject: [PATCH v3] Fix the logical replication timeout during large transactions. The problem is that we don't send keep-alive messages for a long time while processing large transactions during logical replication where we don't send any data of such transactions. This can happen when the table modified in the transaction is not published. We do try to send the keep_alive if necessary at the end of the transaction (via WalSndWriteData()) but by that time the subscriber-side can timeout and exit. To fix this we try to send the keepalive message if required after processing certain threshold of changes. --- src/backend/replication/logical/logical.c | 10 ++++++ src/backend/replication/pgoutput/pgoutput.c | 40 ++++++++++++++++++++- src/backend/replication/walsender.c | 38 +++++++++++++++++--- src/include/replication/logical.h | 3 ++ 4 files changed, 85 insertions(+), 6 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 61902be3b0..6b9f24e52d 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -622,6 +622,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.startup_cb(ctx, opt, is_init); @@ -649,6 +650,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.shutdown_cb(ctx); @@ -684,6 +686,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.begin_cb(ctx, txn); @@ -715,6 +718,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* do the actual work: call callback */ ctx->callbacks.commit_cb(ctx, txn, commit_lsn); @@ -754,6 +758,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.change_cb(ctx, txn, relation, change); /* Pop the error context stack */ @@ -794,6 +800,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ @@ -820,6 +828,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id); @@ -857,6 +866,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 79735da21a..799242128a 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -53,6 +53,7 @@ static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); static void send_relation_and_attrs(Relation relation, LogicalDecodingContext *ctx); +static void update_replication_progress(LogicalDecodingContext *ctx); /* * Entry in the map used to remember which relation schemas we sent. @@ -277,7 +278,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + update_replication_progress(ctx); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -385,6 +386,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, RelationSyncEntry *relentry; Relation ancestor = NULL; + update_replication_progress(ctx); + if (!is_publishable_relation(relation)) return; @@ -514,6 +517,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelids; Oid *relids; + update_replication_progress(ctx); + old = MemoryContextSwitchTo(data->context); relids = palloc0(nrelations * sizeof(Oid)); @@ -912,3 +917,36 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue) while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL) entry->replicate_valid = false; } + +/* + * Try to update progress and send a keepalive message if too many changes were + * processed. + * + * For a large transaction, if we don't send any change to the downstream for a + * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. + * This can happen when all or most of the changes are not published. + */ +static void +update_replication_progress(LogicalDecodingContext *ctx) +{ + static int changes_count = 0; + + /* + * We don't want to try sending a keepalive message after processing each + * change as that can have overhead. Tests revealed that there is no + * noticeable overhead in doing it after continuously processing 100 or so + * changes. + */ +#define CHANGES_THRESHOLD 100 + + /* + * If we are at the end of transaction LSN, update progress tracking. + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx); + changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index fe0d2ee2ed..3e02404429 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -240,6 +240,7 @@ static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); static void ProcessRepliesIfAny(void); +static void ProcessPendingWrites(void); static void WalSndKeepalive(bool requestReply); static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); @@ -1295,6 +1296,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, } /* If we have pending write here, go to slow path */ + ProcessPendingWrites(); +} + +/* + * Wait until there is no pending write. Also process replies from the other + * side and check timeouts during that. + */ +static void +ProcessPendingWrites(void) +{ for (;;) { int wakeEvents; @@ -1354,18 +1365,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); + bool end_xact = ctx->end_xact; /* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently. + * + * We don't have a mechanism to get the ack for any LSN other than end + * xact LSN from the downstream. So, we track lag only for end of + * transaction LSN. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (end_xact && TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* + * Try to send a keepalive if required. We don't need to try sending keep + * alive messages at the transaction end as that will be done at a later + * point in time. This is required only for large transactions where we + * don't send any changes to the downstream and the receiver can timeout + * due to that. + */ + if (!end_xact && + now >= TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + ProcessPendingWrites(); } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index c2f2475e5d..14d71fd136 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -49,6 +49,9 @@ typedef struct LogicalDecodingContext */ bool fast_forward; + /* Are we processing the end LSN of a transaction? */ + bool end_xact; + OutputPluginCallbacks callbacks; OutputPluginOptions options; -- 2.23.0.windows.1 [application/octet-stream] REL12_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch (8.9K, ../../OS3PR01MB62754AA446A2FB5B72816C529EC59@OS3PR01MB6275.jpnprd01.prod.outlook.com/5-REL12_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From d8e3e7cc6894b1f8dea728f12d9e623353d6c6a1 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Thu, 21 Apr 2022 15:06:07 +0800 Subject: [PATCH v3] Fix the logical replication timeout during large transactions. The problem is that we don't send keep-alive messages for a long time while processing large transactions during logical replication where we don't send any data of such transactions. This can happen when the table modified in the transaction is not published. We do try to send the keep_alive if necessary at the end of the transaction (via WalSndWriteData()) but by that time the subscriber-side can timeout and exit. To fix this we try to send the keepalive message if required after processing certain threshold of changes. --- src/backend/replication/logical/logical.c | 10 ++++++ src/backend/replication/pgoutput/pgoutput.c | 40 ++++++++++++++++++++- src/backend/replication/walsender.c | 38 +++++++++++++++++--- src/include/replication/logical.h | 3 ++ 4 files changed, 85 insertions(+), 6 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index a0fc480646..8b79d67001 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -624,6 +624,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.startup_cb(ctx, opt, is_init); @@ -651,6 +652,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.shutdown_cb(ctx); @@ -686,6 +688,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.begin_cb(ctx, txn); @@ -717,6 +720,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* do the actual work: call callback */ ctx->callbacks.commit_cb(ctx, txn, commit_lsn); @@ -756,6 +760,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.change_cb(ctx, txn, relation, change); /* Pop the error context stack */ @@ -796,6 +802,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ @@ -822,6 +830,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id); @@ -859,6 +868,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index d317fd7006..6710f983ea 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -50,6 +50,7 @@ static bool publications_valid; static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); +static void update_replication_progress(LogicalDecodingContext *ctx); /* Entry in the map used to remember which relation schemas we sent. */ typedef struct RelationSyncEntry @@ -247,7 +248,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + update_replication_progress(ctx); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -309,6 +310,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, MemoryContext old; RelationSyncEntry *relentry; + update_replication_progress(ctx); + if (!is_publishable_relation(relation)) return; @@ -389,6 +392,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelids; Oid *relids; + update_replication_progress(ctx); + old = MemoryContextSwitchTo(data->context); relids = palloc0(nrelations * sizeof(Oid)); @@ -660,3 +665,36 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue) while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL) entry->replicate_valid = false; } + +/* + * Try to update progress and send a keepalive message if too many changes were + * processed. + * + * For a large transaction, if we don't send any change to the downstream for a + * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. + * This can happen when all or most of the changes are not published. + */ +static void +update_replication_progress(LogicalDecodingContext *ctx) +{ + static int changes_count = 0; + + /* + * We don't want to try sending a keepalive message after processing each + * change as that can have overhead. Tests revealed that there is no + * noticeable overhead in doing it after continuously processing 100 or so + * changes. + */ +#define CHANGES_THRESHOLD 100 + + /* + * If we are at the end of transaction LSN, update progress tracking. + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx); + changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 3698135e49..c2c0c7df43 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -244,6 +244,7 @@ static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); static void ProcessRepliesIfAny(void); +static void ProcessPendingWrites(void); static void WalSndKeepalive(bool requestReply); static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); @@ -1214,6 +1215,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, } /* If we have pending write here, go to slow path */ + ProcessPendingWrites(); +} + +/* + * Wait until there is no pending write. Also process replies from the other + * side and check timeouts during that. + */ +static void +ProcessPendingWrites(void) +{ for (;;) { int wakeEvents; @@ -1273,18 +1284,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); + bool end_xact = ctx->end_xact; /* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently. + * + * We don't have a mechanism to get the ack for any LSN other than end + * xact LSN from the downstream. So, we track lag only for end of + * transaction LSN. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (end_xact && TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* + * Try to send a keepalive if required. We don't need to try sending keep + * alive messages at the transaction end as that will be done at a later + * point in time. This is required only for large transactions where we + * don't send any changes to the downstream and the receiver can timeout + * due to that. + */ + if (!end_xact && + now >= TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + ProcessPendingWrites(); } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 31c796b765..718080d54a 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -50,6 +50,9 @@ typedef struct LogicalDecodingContext */ bool fast_forward; + /* Are we processing the end LSN of a transaction? */ + bool end_xact; + OutputPluginCallbacks callbacks; OutputPluginOptions options; -- 2.23.0.windows.1 [application/octet-stream] REL11_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch (8.9K, ../../OS3PR01MB62754AA446A2FB5B72816C529EC59@OS3PR01MB6275.jpnprd01.prod.outlook.com/6-REL11_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From 25ca55f819c58001df18d79af7f18a0be1067033 Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Fri, 6 May 2022 11:35:12 +0800 Subject: [PATCH v3] Fix the logical replication timeout during large transactions. The problem is that we don't send keep-alive messages for a long time while processing large transactions during logical replication where we don't send any data of such transactions. This can happen when the table modified in the transaction is not published. We do try to send the keep_alive if necessary at the end of the transaction (via WalSndWriteData()) but by that time the subscriber-side can timeout and exit. To fix this we try to send the keepalive message if required after processing certain threshold of changes. --- src/backend/replication/logical/logical.c | 10 ++++++ src/backend/replication/pgoutput/pgoutput.c | 40 ++++++++++++++++++++- src/backend/replication/walsender.c | 38 +++++++++++++++++--- src/include/replication/logical.h | 3 ++ 4 files changed, 85 insertions(+), 6 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 9b7b4b9cb8..6b5ac2d15a 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -612,6 +612,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.startup_cb(ctx, opt, is_init); @@ -639,6 +640,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.shutdown_cb(ctx); @@ -674,6 +676,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.begin_cb(ctx, txn); @@ -705,6 +708,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* do the actual work: call callback */ ctx->callbacks.commit_cb(ctx, txn, commit_lsn); @@ -744,6 +748,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.change_cb(ctx, txn, relation, change); /* Pop the error context stack */ @@ -784,6 +790,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change); /* Pop the error context stack */ @@ -810,6 +818,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id); @@ -847,6 +856,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index dc07420fa1..c8e5c6fb61 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -50,6 +50,7 @@ static bool publications_valid; static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); +static void update_replication_progress(LogicalDecodingContext *ctx); /* Entry in the map used to remember which relation schemas we sent. */ typedef struct RelationSyncEntry @@ -247,7 +248,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + update_replication_progress(ctx); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -309,6 +310,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, MemoryContext old; RelationSyncEntry *relentry; + update_replication_progress(ctx); + if (!is_publishable_relation(relation)) return; @@ -389,6 +392,8 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelids; Oid *relids; + update_replication_progress(ctx); + old = MemoryContextSwitchTo(data->context); relids = palloc0(nrelations * sizeof(Oid)); @@ -660,3 +665,36 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue) while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL) entry->replicate_valid = false; } + +/* + * Try to update progress and send a keepalive message if too many changes were + * processed. + * + * For a large transaction, if we don't send any change to the downstream for a + * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. + * This can happen when all or most of the changes are not published. + */ +static void +update_replication_progress(LogicalDecodingContext *ctx) +{ + static int changes_count = 0; + + /* + * We don't want to try sending a keepalive message after processing each + * change as that can have overhead. Tests revealed that there is no + * noticeable overhead in doing it after continuously processing 100 or so + * changes. + */ +#define CHANGES_THRESHOLD 100 + + /* + * If we are at the end of transaction LSN, update progress tracking. + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx); + changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 8ffd6c6140..8c785afe0d 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -243,6 +243,7 @@ static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); static void ProcessRepliesIfAny(void); +static void ProcessPendingWrites(void); static void WalSndKeepalive(bool requestReply); static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); @@ -1194,6 +1195,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, } /* If we have pending write here, go to slow path */ + ProcessPendingWrites(); +} + +/* + * Wait until there is no pending write. Also process replies from the other + * side and check timeouts during that. + */ +static void +ProcessPendingWrites(void) +{ for (;;) { int wakeEvents; @@ -1260,18 +1271,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); + bool end_xact = ctx->end_xact; /* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently. + * + * We don't have a mechanism to get the ack for any LSN other than end + * xact LSN from the downstream. So, we track lag only for end of + * transaction LSN. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (end_xact && TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* + * Try to send a keepalive if required. We don't need to try sending keep + * alive messages at the transaction end as that will be done at a later + * point in time. This is required only for large transactions where we + * don't send any changes to the downstream and the receiver can timeout + * due to that. + */ + if (!end_xact && + now >= TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + ProcessPendingWrites(); } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index c25ac1fa85..1d0eb58a64 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -52,6 +52,9 @@ typedef struct LogicalDecodingContext */ bool fast_forward; + /* Are we processing the end LSN of a transaction? */ + bool end_xact; + OutputPluginCallbacks callbacks; OutputPluginOptions options; -- 2.23.0.windows.1 [application/octet-stream] REL10_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch (8.4K, ../../OS3PR01MB62754AA446A2FB5B72816C529EC59@OS3PR01MB6275.jpnprd01.prod.outlook.com/7-REL10_v3-0001-Fix-the-logical-replication-timeout-during-large-.patch) download | inline diff: From 909b0e9da4ffc644d973384b732afc082c4e526f Mon Sep 17 00:00:00 2001 From: wangw <[email protected]> Date: Thu, 21 Apr 2022 15:53:29 +0800 Subject: [PATCH v3] Fix the logical replication timeout during large transactions. The problem is that we don't send keep-alive messages for a long time while processing large transactions during logical replication where we don't send any data of such transactions. This can happen when the table modified in the transaction is not published. We do try to send the keep_alive if necessary at the end of the transaction (via WalSndWriteData()) but by that time the subscriber-side can timeout and exit. To fix this we try to send the keepalive message if required after processing certain threshold of changes. --- src/backend/replication/logical/logical.c | 8 +++++ src/backend/replication/pgoutput/pgoutput.c | 38 ++++++++++++++++++++- src/backend/replication/walsender.c | 38 ++++++++++++++++++--- src/include/replication/logical.h | 1 + 4 files changed, 79 insertions(+), 6 deletions(-) diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 1851a626f9..59d133af01 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -589,6 +589,7 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.startup_cb(ctx, opt, is_init); @@ -614,6 +615,7 @@ shutdown_cb_wrapper(LogicalDecodingContext *ctx) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.shutdown_cb(ctx); @@ -647,6 +649,7 @@ begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn) ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->first_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.begin_cb(ctx, txn); @@ -676,6 +679,7 @@ commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn->xid; ctx->write_location = txn->end_lsn; /* points to the end of the record */ + ctx->end_xact = true; /* do the actual work: call callback */ ctx->callbacks.commit_cb(ctx, txn, commit_lsn); @@ -713,6 +717,8 @@ change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, */ ctx->write_location = change->lsn; + ctx->end_xact = false; + ctx->callbacks.change_cb(ctx, txn, relation, change); /* Pop the error context stack */ @@ -737,6 +743,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id) /* set output state */ ctx->accept_writes = false; + ctx->end_xact = false; /* do the actual work: call callback */ ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id); @@ -772,6 +779,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->accept_writes = true; ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; ctx->write_location = message_lsn; + ctx->end_xact = false; /* do the actual work: call callback */ ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 529a5c0b48..f96fde3df0 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -48,6 +48,7 @@ static bool publications_valid; static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); +static void update_replication_progress(LogicalDecodingContext *ctx); /* Entry in the map used to remember which relation schemas we sent. */ typedef struct RelationSyncEntry @@ -246,7 +247,7 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { - OutputPluginUpdateProgress(ctx); + update_replication_progress(ctx); OutputPluginPrepareWrite(ctx, true); logicalrep_write_commit(ctx->out, txn, commit_lsn); @@ -264,6 +265,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, MemoryContext old; RelationSyncEntry *relentry; + update_replication_progress(ctx); + if (!is_publishable_relation(relation)) return; @@ -628,3 +631,36 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue) while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL) entry->replicate_valid = false; } + +/* + * Try to update progress and send a keepalive message if too many changes were + * processed. + * + * For a large transaction, if we don't send any change to the downstream for a + * long time (exceeds the wal_receiver_timeout of standby) then it can timeout. + * This can happen when all or most of the changes are not published. + */ +static void +update_replication_progress(LogicalDecodingContext *ctx) +{ + static int changes_count = 0; + + /* + * We don't want to try sending a keepalive message after processing each + * change as that can have overhead. Tests revealed that there is no + * noticeable overhead in doing it after continuously processing 100 or so + * changes. + */ +#define CHANGES_THRESHOLD 100 + + /* + * If we are at the end of transaction LSN, update progress tracking. + * Otherwise, after continuously processing CHANGES_THRESHOLD changes, we + * try to send a keepalive message if required. + */ + if (ctx->end_xact || ++changes_count >= CHANGES_THRESHOLD) + { + OutputPluginUpdateProgress(ctx); + changes_count = 0; + } +} diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 3c9b9f5376..4aac92749c 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -243,6 +243,7 @@ static void ProcessStandbyMessage(void); static void ProcessStandbyReplyMessage(void); static void ProcessStandbyHSFeedbackMessage(void); static void ProcessRepliesIfAny(void); +static void ProcessPendingWrites(void); static void WalSndKeepalive(bool requestReply); static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); @@ -1190,6 +1191,16 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, } /* If we have pending write here, go to slow path */ + ProcessPendingWrites(); +} + +/* + * Wait until there is no pending write. Also process replies from the other + * side and check timeouts during that. + */ +static void +ProcessPendingWrites(void) +{ for (;;) { int wakeEvents; @@ -1256,18 +1267,35 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId { static TimestampTz sendTime = 0; TimestampTz now = GetCurrentTimestamp(); + bool end_xact = ctx->end_xact; /* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently. + * + * We don't have a mechanism to get the ack for any LSN other than end + * xact LSN from the downstream. So, we track lag only for end of + * transaction LSN. */ #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 - if (!TimestampDifferenceExceeds(sendTime, now, - WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) - return; + if (end_xact && TimestampDifferenceExceeds(sendTime, now, + WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS)) + { + LagTrackerWrite(lsn, now); + sendTime = now; + } - LagTrackerWrite(lsn, now); - sendTime = now; + /* + * Try to send a keepalive if required. We don't need to try sending keep + * alive messages at the transaction end as that will be done at a later + * point in time. This is required only for large transactions where we + * don't send any changes to the downstream and the receiver can timeout + * due to that. + */ + if (!end_xact && + now >= TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2)) + ProcessPendingWrites(); } /* diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 7f0e0fa881..b25a935bc0 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -80,6 +80,7 @@ typedef struct LogicalDecodingContext */ bool accept_writes; bool prepared_write; + bool end_xact; XLogRecPtr write_location; TransactionId write_xid; } LogicalDecodingContext; -- 2.23.0.windows.1 ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-09 06:47 Amit Kapila <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 2 replies; 14+ messages in thread From: Amit Kapila @ 2022-05-09 06:47 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Fri, May 6, 2022 at 12:42 PM [email protected] <[email protected]> wrote: > > On Fri, May 6, 2022 at 9:54 AM Masahiko Sawada <[email protected]> wrote: > > On Wed, May 4, 2022 at 7:18 PM Amit Kapila <[email protected]> wrote: > > > > > > On Mon, May 2, 2022 at 8:07 AM Masahiko Sawada <[email protected]> > > wrote: > > > > > > > > On Mon, May 2, 2022 at 11:32 AM Amit Kapila <[email protected]> > > wrote: > > > > > > > > > > > > > > > So, shall we go back to the previous approach of using a separate > > > > > function update_replication_progress? > > > > > > > > Ok, agreed. > > > > > > > > > > Attached, please find the updated patch accordingly. Currently, I have > > > prepared it for HEAD, if you don't see any problem with this, we can > > > prepare the back-branch patches based on this. > > > > Thank you for updating the patch. Looks good to me. > Thanks for your review. > > Improve the back-branch patches according to the discussion. > Thanks. The patch LGTM. I'll push and back-patch this after the current minor release is done unless there are more comments related to this work. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-09 08:46 Masahiko Sawada <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 1 reply; 14+ messages in thread From: Masahiko Sawada @ 2022-05-09 08:46 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, May 9, 2022 at 3:47 PM Amit Kapila <[email protected]> wrote: > > On Fri, May 6, 2022 at 12:42 PM [email protected] > <[email protected]> wrote: > > > > On Fri, May 6, 2022 at 9:54 AM Masahiko Sawada <[email protected]> wrote: > > > On Wed, May 4, 2022 at 7:18 PM Amit Kapila <[email protected]> wrote: > > > > > > > > On Mon, May 2, 2022 at 8:07 AM Masahiko Sawada <[email protected]> > > > wrote: > > > > > > > > > > On Mon, May 2, 2022 at 11:32 AM Amit Kapila <[email protected]> > > > wrote: > > > > > > > > > > > > > > > > > > So, shall we go back to the previous approach of using a separate > > > > > > function update_replication_progress? > > > > > > > > > > Ok, agreed. > > > > > > > > > > > > > Attached, please find the updated patch accordingly. Currently, I have > > > > prepared it for HEAD, if you don't see any problem with this, we can > > > > prepare the back-branch patches based on this. > > > > > > Thank you for updating the patch. Looks good to me. > > Thanks for your review. > > > > Improve the back-branch patches according to the discussion. > > > The patches look good to me too. Regards, -- Masahiko Sawada EDB: https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-09 13:30 Euler Taveira <[email protected]> parent: Amit Kapila <[email protected]> 1 sibling, 1 reply; 14+ messages in thread From: Euler Taveira @ 2022-05-09 13:30 UTC (permalink / raw) To: Amit Kapila <[email protected]>; [email protected] <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, May 9, 2022, at 3:47 AM, Amit Kapila wrote: > Thanks. The patch LGTM. I'll push and back-patch this after the > current minor release is done unless there are more comments related > to this work. Looks sane to me. (I only tested the HEAD version) + bool end_xact = ctx->end_xact; Do you really need a new variable here? It has the same name and the new one isn't changed during the execution. Does this issue deserve a test? A small wal_receiver_timeout. Although, I'm not sure how stable the test will be. -- Euler Taveira EDB https://www.enterprisedb.com/ ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-10 03:22 Amit Kapila <[email protected]> parent: Euler Taveira <[email protected]> 0 siblings, 1 reply; 14+ messages in thread From: Amit Kapila @ 2022-05-10 03:22 UTC (permalink / raw) To: Euler Taveira <[email protected]>; +Cc: [email protected] <[email protected]>; Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, May 9, 2022 at 7:01 PM Euler Taveira <[email protected]> wrote: > > On Mon, May 9, 2022, at 3:47 AM, Amit Kapila wrote: > > Thanks. The patch LGTM. I'll push and back-patch this after the > current minor release is done unless there are more comments related > to this work. > > Looks sane to me. (I only tested the HEAD version) > > + bool end_xact = ctx->end_xact; > > Do you really need a new variable here? It has the same name and the new one > isn't changed during the execution. > I think both ways should be okay. I thought the proposed way is okay because it is used in more than one place and is probably slightly easier to follow by having a separate variable. > Does this issue deserve a test? A small wal_receiver_timeout. Although, I'm not > sure how stable the test will be. > Yes, the main part is how to write a stable test because estimating how many changes are enough for the configured wal_receiver_timeout to pass on all the buildfarm machines is tricky. Also, I am not sure how important is to test this behavior because based on this theory we should have tests for all kinds of timeouts. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 14+ messages in thread
* RE: Logical replication timeout problem @ 2022-05-10 03:36 [email protected] <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: [email protected] @ 2022-05-10 03:36 UTC (permalink / raw) To: Amit Kapila <[email protected]>; Euler Taveira <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; [email protected] <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, May 9, 2022 at 11:23 AM Amit Kapila <[email protected]> wrote: > On Mon, May 9, 2022 at 7:01 PM Euler Taveira <[email protected]> wrote: > > > > On Mon, May 9, 2022, at 3:47 AM, Amit Kapila wrote: > > > > Thanks. The patch LGTM. I'll push and back-patch this after the > > current minor release is done unless there are more comments related > > to this work. > > ...... > > Does this issue deserve a test? A small wal_receiver_timeout. Although, I'm > not > > sure how stable the test will be. > > > > Yes, the main part is how to write a stable test because estimating > how many changes are enough for the configured wal_receiver_timeout to > pass on all the buildfarm machines is tricky. Also, I am not sure how > important is to test this behavior because based on this theory we > should have tests for all kinds of timeouts. Yse, I think we could not guarantee the stability of this test. In addition, if we set wal_receiver_timeout too small, it may cause timeout unrelated to this bug. And if we set bigger wal_receiver_timeout and use larger transaction in order to minimize the impact of machine performance, I think this might take some time and might risk making the build farm slow. Regards, Wang wei ^ permalink raw reply [nested|flat] 14+ messages in thread
* Re: Logical replication timeout problem @ 2022-05-11 11:33 Amit Kapila <[email protected]> parent: Masahiko Sawada <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Amit Kapila @ 2022-05-11 11:33 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; +Cc: [email protected] <[email protected]>; [email protected] <[email protected]>; Euler Taveira <[email protected]>; [email protected] <[email protected]>; Peter Smith <[email protected]>; Fabrice Chapuis <[email protected]>; Simon Riggs <[email protected]>; Petr Jelinek <[email protected]>; [email protected] <[email protected]>; PostgreSQL Hackers <[email protected]>; Ajin Cherian <[email protected]> On Mon, May 9, 2022 at 2:17 PM Masahiko Sawada <[email protected]> wrote: > > The patches look good to me too. > Pushed. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 14+ messages in thread
* [PATCH v38 06/11] Add Incremental View Maintenance support @ 2026-05-29 09:06 Yugo Nagata <[email protected]> 0 siblings, 0 replies; 14+ messages in thread From: Yugo Nagata @ 2026-05-29 09:06 UTC (permalink / raw) In this implementation, AFTER triggers are used to collect tuplestores containing transition table contents. When multiple tables are changed, multiple AFTER triggers are invoked, then the final AFTER trigger performs actual update of the matview. In addition, BEFORE triggers are also used to handle global information for view maintenance. To calculate view deltas, we need both pre-state and post-state of base tables. Post-update states are available in AFTER trigger, and pre-update states can be calculated by removing inserted tuples and appending deleted tuples. Insterted tuples are filtered using the snapshot taken before table modification, and deleted tuples are contained in the old transition table. Incrementally Maintainable Materialized Views (IMMV) can contain duplicated tuples. This patch also allows self-join, simultaneous updates of more than one base table, and multiple updates of the same base table. --- src/backend/access/transam/xact.c | 8 + src/backend/commands/createas.c | 709 +++++++ src/backend/commands/matview.c | 1835 ++++++++++++++++- src/backend/commands/tablecmds.c | 4 + .../utils/activity/wait_event_names.txt | 1 + src/include/catalog/pg_proc.dat | 10 + src/include/commands/createas.h | 4 + src/include/commands/matview.h | 11 + src/include/storage/lwlocklist.h | 1 + src/include/storage/subsystemlist.h | 3 + 10 files changed, 2548 insertions(+), 38 deletions(-) diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 3a89149016f..0e6cb5493d2 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -37,6 +37,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/matview.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -2322,6 +2323,9 @@ CommitTransaction(void) CallXactCallbacks(is_parallel_worker ? XACT_EVENT_PARALLEL_PRE_COMMIT : XACT_EVENT_PRE_COMMIT); + /* Store the transaction ID that updated the view incrementally */ + AtPreCommit_IVM(); + /* * If this xact has started any unfinished parallel operation, clean up * its workers, warning about leaked resources. (But we don't actually @@ -2973,6 +2977,7 @@ AbortTransaction(void) AtAbort_Notify(); AtEOXact_RelationMap(false, is_parallel_worker); AtAbort_Twophase(); + AtAbort_IVM(InvalidSubTransactionId); /* * Advertise the fact that we aborted in pg_xact (assuming that we got as @@ -5304,6 +5309,9 @@ AbortSubTransaction(void) UnlockBuffers(); + /* Clean up hash entries for incremental view maintenance */ + AtAbort_IVM(s->subTransactionId); + /* Reset WAL record construction state */ XLogResetInsertion(); diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c index 6dbb831ca89..a499688b79d 100644 --- a/src/backend/commands/createas.c +++ b/src/backend/commands/createas.c @@ -29,18 +29,32 @@ #include "access/tableam.h" #include "access/xact.h" #include "catalog/namespace.h" +#include "catalog/index.h" +#include "catalog/pg_am_d.h" +#include "catalog/pg_constraint.h" +#include "catalog/pg_inherits.h" +#include "catalog/pg_trigger.h" #include "catalog/toasting.h" #include "commands/createas.h" +#include "commands/defrem.h" #include "commands/matview.h" #include "commands/prepare.h" #include "commands/tablecmds.h" +#include "commands/tablespace.h" +#include "commands/trigger.h" #include "commands/view.h" #include "executor/execdesc.h" #include "executor/executor.h" +#include "miscadmin.h" #include "nodes/makefuncs.h" #include "nodes/nodeFuncs.h" #include "nodes/queryjumble.h" +#include "optimizer/optimizer.h" +#include "optimizer/prep.h" #include "parser/analyze.h" +#include "parser/parser.h" +#include "parser/parsetree.h" +#include "parser/parse_clause.h" #include "rewrite/rewriteHandler.h" #include "tcop/tcopprot.h" #include "utils/builtins.h" @@ -70,6 +84,12 @@ static bool intorel_receive(TupleTableSlot *slot, DestReceiver *self); static void intorel_shutdown(DestReceiver *self); static void intorel_destroy(DestReceiver *self); +static void CreateIvmTriggersOnBaseTablesRecurse(Query *qry, Node *node, Oid matviewOid, + List **relids, bool ex_lock); +static void CreateIvmTrigger(Oid relOid, Oid viewOid, int16 type, int16 timing, bool ex_lock); +static void check_ivm_restriction(Node *node); +static bool check_ivm_restriction_walker(Node *node, void *context); +static Bitmapset *get_primary_key_attnos_from_query(Query *query, List **constraintList); /* * create_ctas_internal @@ -277,6 +297,18 @@ ExecCreateTableAs(ParseState *pstate, CreateTableAsStmt *stmt, into->skipData = true; } + if (is_matview && into->ivm) + { + /* check if the query is supported in IMMV definition */ + if (contain_mutable_functions((Node *) query)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("mutable function is not supported on incrementally maintainable materialized view"), + errhint("functions must be marked IMMUTABLE"))); + + check_ivm_restriction((Node *) query); + } + if (into->skipData) { /* @@ -287,15 +319,36 @@ ExecCreateTableAs(ParseState *pstate, CreateTableAsStmt *stmt, */ address = create_ctas_nodata(query->targetList, into); + /* + * For IVM, mark the matview before refresh so RelationIsIVM() + * returns true inside RefreshMatViewByOid. + */ + if (is_matview && into->ivm) + { + Relation matviewRel = table_open(address.objectId, NoLock); + + SetMatViewIVMState(matviewRel, true); + table_close(matviewRel, NoLock); + CommandCounterIncrement(); + } + /* * For materialized views, reuse the REFRESH logic, which locks down * security-restricted operations and restricts the search_path. This * reduces the chance that a subsequent refresh will fail. */ if (do_refresh) + { RefreshMatViewByOid(address.objectId, true, false, false, pstate->p_sourcetext, qc); + if (is_matview && into->ivm && IsolationUsesXactSnapshot()) + ereport(WARNING, + (errmsg("inconsistent view can be created in isolation level SERIALIZABLE or REPEATABLE READ"), + errdetail("The view may not include effects of a concurrent transaction."), + errhint("The view with incremental maintenance should be created in isolation level READ COMMITTED, " + "or refreshed manually to make sure the view is consistent."))); + } } else { @@ -635,3 +688,659 @@ intorel_destroy(DestReceiver *self) { pfree(self); } + +/* + * CreateIvmTriggersOnBaseTables -- create IVM triggers on all base tables + */ +void +CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid) +{ + List *relids = NIL; + bool ex_lock = false; + RangeTblEntry *rte; + + /* Immediately return if we don't have any base tables. */ + if (list_length(qry->rtable) < 1) + return; + + /* + * If the view has more than one base tables, we need an exclusive lock + * on the view so that the view would be maintained serially to avoid + * the inconsistency that occurs when two base tables are modified in + * concurrent transactions. However, if the view has only one table, + * we can use a weaker lock. + * + * The type of lock should be determined here, because if we check the + * view definition at maintenance time, we need to acquire a weaker lock, + * and upgrading the lock level after this increases probability of + * deadlock. + */ + + rte = list_nth(qry->rtable, 0); + if (list_length(qry->rtable) > 1 || rte->rtekind != RTE_RELATION) + ex_lock = true; + + CreateIvmTriggersOnBaseTablesRecurse(qry, (Node *)qry, matviewOid, &relids, ex_lock); + + list_free(relids); +} + +static void +CreateIvmTriggersOnBaseTablesRecurse(Query *qry, Node *node, Oid matviewOid, + List **relids, bool ex_lock) +{ + if (node == NULL) + return; + + /* This can recurse, so check for excessive recursion */ + check_stack_depth(); + + switch (nodeTag(node)) + { + case T_Query: + { + Query *query = (Query *) node; + + CreateIvmTriggersOnBaseTablesRecurse(qry, (Node *)query->jointree, matviewOid, relids, ex_lock); + } + break; + + case T_RangeTblRef: + { + int rti = ((RangeTblRef *) node)->rtindex; + RangeTblEntry *rte = rt_fetch(rti, qry->rtable); + + if (rte->rtekind == RTE_RELATION && !list_member_oid(*relids, rte->relid)) + { + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_INSERT, TRIGGER_TYPE_BEFORE, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_DELETE, TRIGGER_TYPE_BEFORE, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_UPDATE, TRIGGER_TYPE_BEFORE, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_TRUNCATE, TRIGGER_TYPE_BEFORE, true); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_INSERT, TRIGGER_TYPE_AFTER, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_DELETE, TRIGGER_TYPE_AFTER, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_UPDATE, TRIGGER_TYPE_AFTER, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_TRUNCATE, TRIGGER_TYPE_AFTER, true); + + *relids = lappend_oid(*relids, rte->relid); + } + } + break; + + case T_FromExpr: + { + FromExpr *f = (FromExpr *) node; + ListCell *l; + + foreach(l, f->fromlist) + CreateIvmTriggersOnBaseTablesRecurse(qry, lfirst(l), matviewOid, relids, ex_lock); + } + break; + + case T_JoinExpr: + { + JoinExpr *j = (JoinExpr *) node; + + CreateIvmTriggersOnBaseTablesRecurse(qry, j->larg, matviewOid, relids, ex_lock); + CreateIvmTriggersOnBaseTablesRecurse(qry, j->rarg, matviewOid, relids, ex_lock); + } + break; + + default: + elog(ERROR, "unrecognized node type: %d", (int) nodeTag(node)); + } +} + +/* + * CreateIvmTrigger -- create IVM trigger on a base table + */ +static void +CreateIvmTrigger(Oid relOid, Oid viewOid, int16 type, int16 timing, bool ex_lock) +{ + ObjectAddress refaddr; + ObjectAddress address; + CreateTrigStmt *ivm_trigger; + List *transitionRels = NIL; + + Assert(timing == TRIGGER_TYPE_BEFORE || timing == TRIGGER_TYPE_AFTER); + + refaddr.classId = RelationRelationId; + refaddr.objectId = viewOid; + refaddr.objectSubId = 0; + + ivm_trigger = makeNode(CreateTrigStmt); + ivm_trigger->relation = NULL; + ivm_trigger->row = false; + + ivm_trigger->timing = timing; + ivm_trigger->events = type; + + switch (type) + { + case TRIGGER_TYPE_INSERT: + ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_ins_before" : "IVM_trigger_ins_after"); + break; + case TRIGGER_TYPE_DELETE: + ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_del_before" : "IVM_trigger_del_after"); + break; + case TRIGGER_TYPE_UPDATE: + ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_upd_before" : "IVM_trigger_upd_after"); + break; + case TRIGGER_TYPE_TRUNCATE: + ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_truncate_before" : "IVM_trigger_truncate_after"); + break; + default: + elog(ERROR, "unsupported trigger type"); + } + + if (timing == TRIGGER_TYPE_AFTER) + { + if (type == TRIGGER_TYPE_INSERT || type == TRIGGER_TYPE_UPDATE) + { + TriggerTransition *n = makeNode(TriggerTransition); + n->name = "__ivm_newtable"; + n->isNew = true; + n->isTable = true; + + transitionRels = lappend(transitionRels, n); + } + if (type == TRIGGER_TYPE_DELETE || type == TRIGGER_TYPE_UPDATE) + { + TriggerTransition *n = makeNode(TriggerTransition); + n->name = "__ivm_oldtable"; + n->isNew = false; + n->isTable = true; + + transitionRels = lappend(transitionRels, n); + } + } + + /* + * XXX: When using DELETE or UPDATE, we must use exclusive lock for now + * because apply_old_delta(_with_count) uses ctid to identify the tuple + * to be deleted/deleted, but doesn't work in concurrent situations. + * + * If the view doesn't have aggregate, distinct, or tuple duplicate, + * then it would work even in concurrent situations. However, we don't have + * any way to guarantee the view has a unique key before opening the IMMV + * at the maintenance time because users may drop the unique index. + */ + + if (type == TRIGGER_TYPE_DELETE || type == TRIGGER_TYPE_UPDATE) + ex_lock = true; + + ivm_trigger->funcname = + (timing == TRIGGER_TYPE_BEFORE ? SystemFuncName("IVM_immediate_before") : SystemFuncName("IVM_immediate_maintenance")); + + ivm_trigger->columns = NIL; + ivm_trigger->transitionRels = transitionRels; + ivm_trigger->whenClause = NULL; + ivm_trigger->isconstraint = false; + ivm_trigger->deferrable = false; + ivm_trigger->initdeferred = false; + ivm_trigger->constrrel = NULL; + ivm_trigger->args = list_make2( + makeString(DatumGetPointer(DirectFunctionCall1(oidout, ObjectIdGetDatum(viewOid)))), + makeString(DatumGetPointer(DirectFunctionCall1(boolout, BoolGetDatum(ex_lock)))) + ); + + address = CreateTrigger(ivm_trigger, NULL, relOid, InvalidOid, InvalidOid, + InvalidOid, InvalidOid, InvalidOid, NULL, true, false); + + recordDependencyOn(&address, &refaddr, DEPENDENCY_AUTO); + + /* Make changes-so-far visible */ + CommandCounterIncrement(); +} + +/* + * check_ivm_restriction --- look for specify nodes in the query tree + */ +static void +check_ivm_restriction(Node *node) +{ + check_ivm_restriction_walker(node, NULL); +} + +static bool +check_ivm_restriction_walker(Node *node, void *context) +{ + if (node == NULL) + return false; + + /* + * We currently don't support Sub-Query. + */ + if (IsA(node, SubPlan) || IsA(node, SubLink)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("subquery is not supported on incrementally maintainable materialized view"))); + + /* This can recurse, so check for excessive recursion */ + check_stack_depth(); + + switch (nodeTag(node)) + { + case T_Query: + { + Query *qry = (Query *)node; + ListCell *lc; + List *vars; + + /* if contained CTE, return error */ + if (qry->cteList != NIL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("CTE is not supported on incrementally maintainable materialized view"))); + if (qry->havingQual != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg(" HAVING clause is not supported on incrementally maintainable materialized view"))); + if (qry->sortClause != NIL) /* There is a possibility that we don't need to return an error */ + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("ORDER BY clause is not supported on incrementally maintainable materialized view"))); + if (qry->limitOffset != NULL || qry->limitCount != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("LIMIT/OFFSET clause is not supported on incrementally maintainable materialized view"))); + if (qry->distinctClause) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("DISTINCT is not supported on incrementally maintainable materialized view"))); + if (qry->hasDistinctOn) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("DISTINCT ON is not supported on incrementally maintainable materialized view"))); + if (qry->hasWindowFuncs) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("window functions are not supported on incrementally maintainable materialized view"))); + if (qry->groupingSets != NIL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("GROUPING SETS, ROLLUP, or CUBE clauses is not supported on incrementally maintainable materialized view"))); + if (qry->setOperations != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("UNION/INTERSECT/EXCEPT statements are not supported on incrementally maintainable materialized view"))); + if (list_length(qry->targetList) == 0) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("empty target list is not supported on incrementally maintainable materialized view"))); + if (qry->rowMarks != NIL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("FOR UPDATE/SHARE clause is not supported on incrementally maintainable materialized view"))); + + /* system column restrictions */ + vars = pull_vars_of_level((Node *) qry, 0); + foreach(lc, vars) + { + if (IsA(lfirst(lc), Var)) + { + Var *var = (Var *) lfirst(lc); + /* if system column, return error */ + if (var->varattno < 0) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("system column is not supported on incrementally maintainable materialized view"))); + } + } + + /* check if type in the target list had an equality operator */ + foreach(lc, qry->targetList) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + Oid atttype = exprType((Node *) tle->expr); + Oid opclass; + + opclass = GetDefaultOpClass(atttype, BTREE_AM_OID); + if (!OidIsValid(opclass)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("data type %s has no default operator class for access method \"%s\"", + format_type_be(atttype), "btree"))); + } + + /* restrictions for rtable */ + foreach(lc, qry->rtable) + { + RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc); + + if (rte->subquery) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("subquery is not supported on incrementally maintainable materialized view"))); + + if (rte->tablesample != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("TABLESAMPLE clause is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_PARTITIONED_TABLE) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("partitioned table is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_RELATION && has_superclass(rte->relid)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("partitions is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_RELATION && find_inheritance_children(rte->relid, NoLock) != NIL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("inheritance parent is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_FOREIGN_TABLE) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("foreign table is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_VIEW || + rte->relkind == RELKIND_MATVIEW) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("VIEW or MATERIALIZED VIEW is not supported on incrementally maintainable materialized view"))); + + if (rte->rtekind == RTE_VALUES) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("VALUES is not supported on incrementally maintainable materialized view"))); + + } + + query_tree_walker(qry, check_ivm_restriction_walker, NULL, QTW_IGNORE_RANGE_TABLE); + + break; + } + case T_TargetEntry: + { + TargetEntry *tle = (TargetEntry *)node; + if (isIvmName(tle->resname)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("column name %s is not supported on incrementally maintainable materialized view", tle->resname))); + + expression_tree_walker(node, check_ivm_restriction_walker, NULL); + break; + } + case T_JoinExpr: + { + JoinExpr *joinexpr = (JoinExpr *)node; + + if (joinexpr->jointype > JOIN_INNER) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("OUTER JOIN is not supported on incrementally maintainable materialized view"))); + + expression_tree_walker(node, check_ivm_restriction_walker, NULL); + } + break; + case T_Aggref: + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("aggregate function is not supported on incrementally maintainable materialized view"))); + break; + default: + expression_tree_walker(node, check_ivm_restriction_walker, (void *) context); + break; + } + return false; +} + +/* + * CreateIndexOnIMMV + * + * Create a unique index on incremental maintainable materialized view. + * If the view definition query has a GROUP BY clause, the index is created + * on the columns of GROUP BY expressions. Otherwise, if the view contains + * all primary key attributes of its base tables in the target list, the index + * is created on these attributes. In other cases, no index is created. + */ +void +CreateIndexOnIMMV(Query *query, Relation matviewRel) +{ + ListCell *lc; + IndexStmt *index; + ObjectAddress address; + List *constraintList = NIL; + char idxname[NAMEDATALEN]; + List *indexoidlist = RelationGetIndexList(matviewRel); + ListCell *indexoidscan; + Bitmapset *key_attnos; + + snprintf(idxname, sizeof(idxname), "%s_index", RelationGetRelationName(matviewRel)); + + index = makeNode(IndexStmt); + + index->unique = true; + index->primary = false; + index->isconstraint = false; + index->deferrable = false; + index->initdeferred = false; + index->idxname = idxname; + index->relation = + makeRangeVar(get_namespace_name(RelationGetNamespace(matviewRel)), + pstrdup(RelationGetRelationName(matviewRel)), + -1); + index->accessMethod = DEFAULT_INDEX_TYPE; + index->options = NIL; + index->tableSpace = get_tablespace_name(matviewRel->rd_rel->reltablespace); + index->whereClause = NULL; + index->indexParams = NIL; + index->indexIncludingParams = NIL; + index->excludeOpNames = NIL; + index->idxcomment = NULL; + index->indexOid = InvalidOid; + index->oldNumber = InvalidRelFileNumber; + index->oldCreateSubid = InvalidSubTransactionId; + index->oldFirstRelfilelocatorSubid = InvalidSubTransactionId; + index->transformed = true; + index->concurrent = false; + index->if_not_exists = false; + + /* create index on the base tables' primary key columns */ + key_attnos = get_primary_key_attnos_from_query(query, &constraintList); + if (key_attnos) + { + foreach(lc, query->targetList) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, tle->resno - 1); + + if (bms_is_member(tle->resno - FirstLowInvalidHeapAttributeNumber, key_attnos)) + { + IndexElem *iparam; + + iparam = makeNode(IndexElem); + iparam->name = pstrdup(NameStr(attr->attname)); + iparam->expr = NULL; + iparam->indexcolname = NULL; + iparam->collation = NIL; + iparam->opclass = NIL; + iparam->opclassopts = NIL; + iparam->ordering = SORTBY_DEFAULT; + iparam->nulls_ordering = SORTBY_NULLS_DEFAULT; + index->indexParams = lappend(index->indexParams, iparam); + } + } + } + else + { + /* create no index, just notice that an appropriate index is necessary for efficient IVM */ + ereport(NOTICE, + (errmsg("could not create an index on materialized view \"%s\" automatically", + RelationGetRelationName(matviewRel)), + errdetail("This target list does not have all the primary key columns. "), + errhint("Create an index on the materialized view for efficient incremental maintenance."))); + return; + } + + /* If we have a compatible index, we don't need to create another. */ + foreach(indexoidscan, indexoidlist) + { + Oid indexoid = lfirst_oid(indexoidscan); + Relation indexRel; + bool hasCompatibleIndex = false; + + indexRel = index_open(indexoid, AccessShareLock); + + if (CheckIndexCompatible(indexRel->rd_id, + index->accessMethod, + index->indexParams, + index->excludeOpNames, + false)) + hasCompatibleIndex = true; + + index_close(indexRel, AccessShareLock); + + if (hasCompatibleIndex) + return; + } + + address = DefineIndex(NULL, + RelationGetRelid(matviewRel), + index, + InvalidOid, + InvalidOid, + InvalidOid, + -1, + false, true, false, false, true); + + ereport(NOTICE, + (errmsg("created index \"%s\" on materialized view \"%s\"", + idxname, RelationGetRelationName(matviewRel)))); + + /* + * Make dependencies so that the index is dropped if any base tables' + * primary key is dropped. + */ + foreach(lc, constraintList) + { + Oid constraintOid = lfirst_oid(lc); + ObjectAddress refaddr; + + refaddr.classId = ConstraintRelationId; + refaddr.objectId = constraintOid; + refaddr.objectSubId = 0; + + recordDependencyOn(&address, &refaddr, DEPENDENCY_NORMAL); + } +} + + +/* + * get_primary_key_attnos_from_query + * + * Identify the columns in base tables' primary keys in the target list. + * + * Returns a Bitmapset of the column attnos of the primary key's columns of + * tables that used in the query. The attnos are offset by + * FirstLowInvalidHeapAttributeNumber as same as get_primary_key_attnos. + * + * If any table has no primary key or any primary key's columns is not in + * the target list, return NULL. We also return NULL if any pkey constraint + * is deferrable. + * + * constraintList is set to a list of the OIDs of the pkey constraints. + */ +static Bitmapset * +get_primary_key_attnos_from_query(Query *query, List **constraintList) +{ + List *key_attnos_list = NIL; + ListCell *lc; + int i; + Bitmapset *keys = NULL; + Relids rels_in_from; + + /* + * Collect primary key attributes from all tables used in query. The key attributes + * sets for each table are stored in key_attnos_list in order by RTE index. + */ + foreach(lc, query->rtable) + { + RangeTblEntry *r = (RangeTblEntry*) lfirst(lc); + Bitmapset *key_attnos; + bool has_no_pkey = false; + + /* for tables, call get_primary_key_attnos */ + if (r->rtekind == RTE_RELATION) + { + Oid constraintOid; + key_attnos = get_primary_key_attnos(r->relid, false, &constraintOid); + *constraintList = lappend_oid(*constraintList, constraintOid); + has_no_pkey = (key_attnos == NULL); + } + /* + * Ignore join rels, because they are flatten later by + * flatten_join_alias_vars(). Store NULL into key_attnos_list + * as a dummy. + */ + else if (r->rtekind == RTE_JOIN) + { + key_attnos = NULL; + } + /* for other RTEs, store NULL into key_attnos_list */ + else + has_no_pkey = true; + + /* + * If any table or subquery has no primary key or its pkey constraint is deferrable, + * we cannot get key attributes for this query, so return NULL. + */ + if (has_no_pkey) + return NULL; + + key_attnos_list = lappend(key_attnos_list, key_attnos); + } + + /* Collect key attributes appearing in the target list */ + i = 1; + foreach(lc, query->targetList) + { + TargetEntry *tle = (TargetEntry *) flatten_join_alias_vars(NULL, query, lfirst(lc)); + + if (IsA(tle->expr, Var)) + { + Var *var = (Var*) tle->expr; + Bitmapset *key_attnos = list_nth(key_attnos_list, var->varno - 1); + + /* check if this attribute is from a base table's primary key */ + if (bms_is_member(var->varattno - FirstLowInvalidHeapAttributeNumber, key_attnos)) + { + /* + * Remove found key attributes from key_attnos_list, and add this + * to the result list. + */ + key_attnos = bms_del_member(key_attnos, var->varattno - FirstLowInvalidHeapAttributeNumber); + if (bms_is_empty(key_attnos)) + { + key_attnos_list = list_delete_nth_cell(key_attnos_list, var->varno - 1); + key_attnos_list = list_insert_nth(key_attnos_list, var->varno - 1, NULL); + } + keys = bms_add_member(keys, i - FirstLowInvalidHeapAttributeNumber); + } + } + i++; + } + + /* Collect RTE indexes of relations appearing in the FROM clause */ + rels_in_from = get_relids_in_jointree((Node *) query->jointree, false, false); + + /* + * Check if all key attributes of relations in FROM are appearing in the target + * list. If an attribute remains in key_attnos_list in spite of the table is used + * in FROM clause, the target is missing this key attribute, so we return NULL. + */ + i = 1; + foreach(lc, key_attnos_list) + { + Bitmapset *bms = (Bitmapset *)lfirst(lc); + if (!bms_is_empty(bms) && bms_is_member(i, rels_in_from)) + return NULL; + i++; + } + + return keys; +} diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c index f7d8007f796..eaf39f80cd3 100644 --- a/src/backend/commands/matview.c +++ b/src/backend/commands/matview.c @@ -23,24 +23,36 @@ #include "catalog/indexing.h" #include "catalog/namespace.h" #include "catalog/pg_am.h" +#include "catalog/pg_depend.h" +#include "catalog/pg_trigger.h" #include "catalog/pg_opclass.h" #include "commands/matview.h" #include "commands/repack.h" #include "commands/tablecmds.h" #include "commands/tablespace.h" +#include "commands/createas.h" #include "executor/executor.h" #include "executor/spi.h" +#include "executor/tstoreReceiver.h" #include "miscadmin.h" +#include "nodes/makefuncs.h" +#include "parser/analyze.h" +#include "parser/parse_clause.h" +#include "parser/parse_func.h" +#include "parser/parse_relation.h" #include "pgstat.h" #include "rewrite/rewriteHandler.h" +#include "rewrite/rowsecurity.h" #include "storage/lmgr.h" +#include "storage/lwlock.h" #include "tcop/tcopprot.h" #include "utils/builtins.h" +#include "utils/fmgroids.h" #include "utils/lsyscache.h" #include "utils/rel.h" #include "utils/snapmgr.h" #include "utils/syscache.h" - +#include "utils/typcache.h" typedef struct { @@ -53,6 +65,97 @@ typedef struct BulkInsertState bistate; /* bulk insert state */ } DR_transientrel; +#define MV_INIT_QUERYHASHSIZE 16 + +/* + * MV_TriggerHashEntry + * + * Hash entry for base tables on which IVM trigger is invoked + */ +typedef struct MV_TriggerHashEntry +{ + Oid matview_id; /* OID of the materialized view */ + int before_trig_count; /* count of before triggers invoked */ + int after_trig_count; /* count of after triggers invoked */ + + Snapshot snapshot; /* Snapshot just before table change */ + + List *tables; /* List of MV_TriggerTable */ + bool has_old; /* tuples are deleted from any table? */ + bool has_new; /* tuples are inserted into any table? */ + + /* + * List of sub-transaction IDs that incrementally updated the view. + * This list is maintained through a transaction, and an ID is removed + * when a sub-transaction is aborted. If any ID is left when the + * transaction is committed, this means the view is incrementally + * updated in this transaction. + */ + List *subxids; +} MV_TriggerHashEntry; + +/* + * MV_TriggerTable + * + * IVM related data for tables on which the trigger is invoked. + */ +typedef struct MV_TriggerTable +{ + Oid table_id; /* OID of the modified table */ + List *old_tuplestores; /* tuplestores for deleted tuples */ + List *new_tuplestores; /* tuplestores for inserted tuples */ + + List *rte_indexes; /* List of RTE index of the modified table */ + RangeTblEntry *original_rte; /* the original RTE saved before rewriting query */ + + Relation rel; /* relation of the modified table */ + TupleTableSlot *slot; /* for checking visibility in the pre-state table */ +} MV_TriggerTable; + +typedef struct DroppedImmvInfo +{ + Oid immv_oid; + SubTransactionId subxid; +} DroppedImmvInfo; + +static HTAB *mv_trigger_info = NULL; +static HTAB *dropped_immv = NULL; + +static HTAB *LastIvmUpdateHash; + +#define LAST_IVM_UPDATE_HASH_SIZE 256 + +static void IvmShmemRequest(void *arg); + +/* hash table entries */ +typedef struct LastIvmUpdateEntry +{ + Oid oid; /* hash key */ + FullTransactionId last_ivm_update; +} LastIvmUpdateEntry; + +const ShmemCallbacks IvmShmemCallbacks = { + .request_fn = IvmShmemRequest, +}; + +static void +IvmShmemRequest(void *arg) +{ + ShmemRequestHash(.name = "LastIvmUpdate hash", + .nelems = LAST_IVM_UPDATE_HASH_SIZE, + .ptr = &LastIvmUpdateHash, + .hash_info.keysize = sizeof(Oid), + .hash_info.entrysize = sizeof(LastIvmUpdateEntry), + .hash_flags = HASH_ELEM | HASH_BLOBS, + ); +} + +static bool in_delta_calculation = false; + +/* ENR name for materialized view delta */ +#define NEW_DELTA_ENRNAME "new_delta" +#define OLD_DELTA_ENRNAME "old_delta" + static int matview_maintenance_depth = 0; static void transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo); @@ -60,6 +163,8 @@ static bool transientrel_receive(TupleTableSlot *slot, DestReceiver *self); static void transientrel_shutdown(DestReceiver *self); static void transientrel_destroy(DestReceiver *self); static uint64 refresh_matview_datafill(DestReceiver *dest, Query *query, + QueryEnvironment *queryEnv, + TupleDesc *resultTupleDesc, const char *queryString, bool is_create); static void refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner, int save_sec_context); @@ -67,6 +172,41 @@ static void refresh_by_heap_swap(Oid matviewOid, Oid OIDNewHeap, char relpersist static bool is_usable_unique_index(Relation indexRel); static void OpenMatViewIncrementalMaintenance(void); static void CloseMatViewIncrementalMaintenance(void); +static Query *get_matview_query(Relation matviewRel); + +static Query *rewrite_query_for_preupdate_state(Query *query, List *tables, + ParseState *pstate, Oid matviewid); +static void register_delta_ENRs(ParseState *pstate, Query *query, List *tables); +static char *make_subquery_targetlist_from_table(MV_TriggerTable *table); +static char *make_delta_enr_name(const char *prefix, Oid relid, int count); +static RangeTblEntry *get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table, + QueryEnvironment *queryEnv, Oid matviewid); +static RangeTblEntry *makeDeltaTable(RangeTblEntry *rte, MV_TriggerTable *table, + bool is_new, QueryEnvironment *queryEnv); +static Query *rewrite_query_for_counting(Query *query, ParseState *pstate); + +static void calc_delta(MV_TriggerTable *table, int rte_index, Query *query, + DestReceiver *dest_old, DestReceiver *dest_new, + TupleDesc *tupdesc_old, TupleDesc *tupdesc_new, + QueryEnvironment *queryEnv); +static Query *rewrite_query_for_postupdate_state(Query *query, MV_TriggerTable *table, int rte_index); + +static void apply_delta(Oid matviewOid, Tuplestorestate *old_tuplestores, Tuplestorestate *new_tuplestores, + TupleDesc tupdesc_old, TupleDesc tupdesc_new, + Query *query); +static void apply_old_delta(const char *matviewname, const char *deltaname_old, + List *keys); +static void apply_new_delta(const char *matviewname, const char *deltaname_new, + StringInfo target_list); +static char *get_matching_condition_string(List *keys); +static void generate_equal(StringInfo querybuf, Oid opttype, + const char *leftop, const char *rightop); + +static void mv_InitHashTables(void); +static void clean_up_IVM_hash_entry(MV_TriggerHashEntry *entry, bool is_abort, + SubTransactionId subxid); +static void setLastUpdateXid(Oid immv_oid, FullTransactionId xid); +static FullTransactionId getLastUpdateXid(Oid immv_oid); /* * SetMatViewPopulatedState @@ -108,6 +248,46 @@ SetMatViewPopulatedState(Relation relation, bool newstate) CommandCounterIncrement(); } +/* + * SetMatViewIVMState + * Mark a materialized view as IVM, or not. + * + * NOTE: caller must be holding an appropriate lock on the relation. + */ +void +SetMatViewIVMState(Relation relation, bool newstate) +{ + Relation pgrel; + HeapTuple tuple; + + Assert(relation->rd_rel->relkind == RELKIND_MATVIEW); + + /* + * Update relation's pg_class entry. Crucial side-effect: other backends + * (and this one too!) are sent SI message to make them rebuild relcache + * entries. + */ + pgrel = table_open(RelationRelationId, RowExclusiveLock); + tuple = SearchSysCacheCopy1(RELOID, + ObjectIdGetDatum(RelationGetRelid(relation))); + if (!HeapTupleIsValid(tuple)) + elog(ERROR, "cache lookup failed for relation %u", + RelationGetRelid(relation)); + + ((Form_pg_class) GETSTRUCT(tuple))->relisivm = newstate; + + CatalogTupleUpdate(pgrel, &tuple->t_self, tuple); + + heap_freetuple(tuple); + table_close(pgrel, RowExclusiveLock); + + /* + * Advance command counter to make the updated pg_class row locally + * visible. + */ + CommandCounterIncrement(); +} + /* * ExecRefreshMatView -- execute a REFRESH MATERIALIZED VIEW command * @@ -166,8 +346,6 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, QueryCompletion *qc) { Relation matviewRel; - RewriteRule *rule; - List *actions; Query *dataQuery; Oid tableSpace; Oid relowner; @@ -178,6 +356,7 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, int save_sec_context; int save_nestlevel; ObjectAddress address; + bool oldPopulated; matviewRel = table_open(matviewOid, NoLock); relowner = matviewRel->rd_rel->relowner; @@ -193,6 +372,8 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, save_nestlevel = NewGUCNestLevel(); RestrictSearchPath(); + oldPopulated = RelationIsPopulated(matviewRel); + /* Make sure it is a materialized view. */ if (matviewRel->rd_rel->relkind != RELKIND_MATVIEW) ereport(ERROR, @@ -213,32 +394,7 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, errmsg("%s options %s and %s cannot be used together", "REFRESH", "CONCURRENTLY", "WITH NO DATA"))); - /* - * Check that everything is correct for a refresh. Problems at this point - * are internal errors, so elog is sufficient. - */ - if (matviewRel->rd_rel->relhasrules == false || - matviewRel->rd_rules->numLocks < 1) - elog(ERROR, - "materialized view \"%s\" is missing rewrite information", - RelationGetRelationName(matviewRel)); - - if (matviewRel->rd_rules->numLocks > 1) - elog(ERROR, - "materialized view \"%s\" has too many rules", - RelationGetRelationName(matviewRel)); - - rule = matviewRel->rd_rules->rules[0]; - if (rule->event != CMD_SELECT || !(rule->isInstead)) - elog(ERROR, - "the rule for materialized view \"%s\" is not a SELECT INSTEAD OF rule", - RelationGetRelationName(matviewRel)); - - actions = rule->actions; - if (list_length(actions) != 1) - elog(ERROR, - "the rule for materialized view \"%s\" is not a single action", - RelationGetRelationName(matviewRel)); + dataQuery = get_matview_query(matviewRel); /* * Check that there is a unique index with no WHERE clause on one or more @@ -275,12 +431,6 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, errhint("Create a unique index with no WHERE clause on one or more columns of the materialized view."))); } - /* - * The stored query was rewritten at the time of the MV definition, but - * has not been scribbled on by the planner. - */ - dataQuery = linitial_node(Query, actions); - /* * Check for active uses of the relation in the current transaction, such * as open scans. @@ -310,6 +460,90 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, relpersistence = matviewRel->rd_rel->relpersistence; } + /* delete IMMV triggers. */ + if (RelationIsIVM(matviewRel) && skipData) + { + Relation tgRel; + Relation depRel; + ScanKeyData key; + SysScanDesc scan; + HeapTuple tup; + ObjectAddresses *immv_triggers; + + immv_triggers = new_object_addresses(); + + tgRel = table_open(TriggerRelationId, RowExclusiveLock); + depRel = table_open(DependRelationId, RowExclusiveLock); + + /* search triggers that depends on IMMV. */ + ScanKeyInit(&key, + Anum_pg_depend_refobjid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(matviewOid)); + scan = systable_beginscan(depRel, DependReferenceIndexId, true, + NULL, 1, &key); + while ((tup = systable_getnext(scan)) != NULL) + { + ObjectAddress obj; + Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(tup); + + if (foundDep->classid == TriggerRelationId) + { + HeapTuple tgtup; + ScanKeyData tgkey[1]; + SysScanDesc tgscan; + Form_pg_trigger tgform; + + /* Find the trigger name. */ + ScanKeyInit(&tgkey[0], + Anum_pg_trigger_oid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(foundDep->objid)); + + tgscan = systable_beginscan(tgRel, TriggerOidIndexId, true, + NULL, 1, tgkey); + tgtup = systable_getnext(tgscan); + if (!HeapTupleIsValid(tgtup)) + elog(ERROR, "could not find tuple for immv trigger %u", foundDep->objid); + + tgform = (Form_pg_trigger) GETSTRUCT(tgtup); + + /* If trigger is created by IMMV, delete it. */ + if (strncmp(NameStr(tgform->tgname), "IVM_trigger_", 12) == 0) + { + obj.classId = foundDep->classid; + obj.objectId = foundDep->objid; + obj.objectSubId = foundDep->refobjsubid; + add_exact_object_address(&obj, immv_triggers); + } + systable_endscan(tgscan); + } + } + systable_endscan(scan); + + performMultipleDeletions(immv_triggers, DROP_RESTRICT, PERFORM_DELETION_INTERNAL); + + table_close(depRel, RowExclusiveLock); + table_close(tgRel, RowExclusiveLock); + free_object_addresses(immv_triggers); + } + + /* + * Create triggers on incremental maintainable materialized view + * This argument should use 'dataQuery'. This needs to use a rewritten query, + * because a sublink in jointree is not supported by this function. + * + * This is performed before generating data because we have to wait + * concurrent transactions modifying a base table and then take a snapshot + * to see changes by these transactions to make sure a consistent view + * is created. + */ + if (RelationIsIVM(matviewRel) && !skipData && !oldPopulated) + { + CreateIvmTriggersOnBaseTables(dataQuery, matviewOid); + CreateIndexOnIMMV(dataQuery, matviewRel); + } + /* * Create the transient table that will receive the regenerated data. Lock * it against access by any other process until commit (by which time it @@ -326,8 +560,43 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, DestReceiver *dest; dest = CreateTransientRelDestReceiver(OIDNewHeap); - processed = refresh_matview_datafill(dest, dataQuery, queryString, - is_create); + + if (RelationIsIVM(matviewRel)) + { + /* + * In READ COMMITTED, get and push the latest snapshot again to see the + * results of concurrent transactions committed after the current + * transaction started. + */ + if (!IsolationUsesXactSnapshot()) + PushActiveSnapshot(GetTransactionSnapshot()); + + /* + * If a concurrent transaction updated the view incrementally and was + * committed before we acquired the lock, the results of refresh_immv could + * be inconsistent. Therefore, we have to check the transaction ID of the + * most recent update of the view, and if this was in progress at the + * transaction start, raise an error to prevent anomalies. + */ + if (!is_create) + { + FullTransactionId xid; + + xid = getLastUpdateXid(matviewOid); + if (XidInMVCCSnapshot(XidFromFullTransactionId(xid), GetActiveSnapshot())) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("the materialized view is incrementally updated in concurrent transaction"), + errhint("The transaction might succeed if retried."))); + } + } + + processed = refresh_matview_datafill(dest, dataQuery, NULL, NULL, + queryString, is_create); + + /* Pop the original snapshot. */ + if (RelationIsIVM(matviewRel) && !IsolationUsesXactSnapshot()) + PopActiveSnapshot(); } /* Make the matview match the newly generated data. */ @@ -402,6 +671,8 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, */ static uint64 refresh_matview_datafill(DestReceiver *dest, Query *query, + QueryEnvironment *queryEnv, + TupleDesc *resultTupleDesc, const char *queryString, bool is_create) { List *rewritten; @@ -439,7 +710,7 @@ refresh_matview_datafill(DestReceiver *dest, Query *query, /* Create a QueryDesc, redirecting output to our tuple receiver */ queryDesc = CreateQueryDesc(plan, queryString, GetActiveSnapshot(), InvalidSnapshot, - dest, NULL, NULL, 0); + dest, NULL, queryEnv ? queryEnv: NULL, 0); /* call ExecutorStart to prepare the plan for execution */ ExecutorStart(queryDesc, 0); @@ -449,6 +720,9 @@ refresh_matview_datafill(DestReceiver *dest, Query *query, processed = queryDesc->estate->es_processed; + if (resultTupleDesc) + *resultTupleDesc = CreateTupleDescCopy(queryDesc->tupDesc); + /* and clean up */ ExecutorFinish(queryDesc); ExecutorEnd(queryDesc); @@ -967,3 +1241,1488 @@ CloseMatViewIncrementalMaintenance(void) matview_maintenance_depth--; Assert(matview_maintenance_depth >= 0); } + +/* + * get_matview_query - get the Query from a matview's _RETURN rule. + */ +static Query * +get_matview_query(Relation matviewRel) +{ + RewriteRule *rule; + List * actions; + + /* + * Check that everything is correct for a refresh. Problems at this point + * are internal errors, so elog is sufficient. + */ + if (matviewRel->rd_rel->relhasrules == false || + matviewRel->rd_rules->numLocks < 1) + elog(ERROR, + "materialized view \"%s\" is missing rewrite information", + RelationGetRelationName(matviewRel)); + + if (matviewRel->rd_rules->numLocks > 1) + elog(ERROR, + "materialized view \"%s\" has too many rules", + RelationGetRelationName(matviewRel)); + + rule = matviewRel->rd_rules->rules[0]; + if (rule->event != CMD_SELECT || !(rule->isInstead)) + elog(ERROR, + "the rule for materialized view \"%s\" is not a SELECT INSTEAD OF rule", + RelationGetRelationName(matviewRel)); + + actions = rule->actions; + if (list_length(actions) != 1) + elog(ERROR, + "the rule for materialized view \"%s\" is not a single action", + RelationGetRelationName(matviewRel)); + + /* + * The stored query was rewritten at the time of the MV definition, but + * has not been scribbled on by the planner. + */ + return linitial_node(Query, actions); +} + + +/* ---------------------------------------------------- + * Incremental View Maintenance routines + * --------------------------------------------------- + */ + +/* + * IVM_immediate_before + * + * IVM trigger function invoked before base table is modified. If this is + * invoked firstly in the same statement, we save the transaction id and the + * command id at that time. + */ +Datum +IVM_immediate_before(PG_FUNCTION_ARGS) +{ + TriggerData *trigdata = (TriggerData *) fcinfo->context; + char *matviewOid_text = trigdata->tg_trigger->tgargs[0]; + char *ex_lock_text = trigdata->tg_trigger->tgargs[1]; + Oid matviewOid; + MV_TriggerHashEntry *entry; + bool found; + bool ex_lock; + + matviewOid = DatumGetObjectId(DirectFunctionCall1(oidin, CStringGetDatum(matviewOid_text))); + ex_lock = DatumGetBool(DirectFunctionCall1(boolin, CStringGetDatum(ex_lock_text))); + + /* If the view has more than one tables, we have to use an exclusive lock. */ + if (ex_lock) + { + FullTransactionId xid; + + /* + * Wait for concurrent transactions which update this materialized view at + * READ COMMITED. This is needed to see changes committed in other + * transactions. No wait and raise an error at REPEATABLE READ or + * SERIALIZABLE to prevent update anomalies of matviews. + * XXX: dead-lock is possible here. + */ + if (!IsolationUsesXactSnapshot()) + LockRelationOid(matviewOid, ExclusiveLock); + else if (!ConditionalLockRelationOid(matviewOid, ExclusiveLock)) + { + /* try to throw error by name; relation could be deleted... */ + char *relname = get_rel_name(matviewOid); + + if (!relname) + ereport(ERROR, + (errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on materialized view during incremental maintenance"))); + + ereport(ERROR, + (errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on materialized view \"%s\" during incremental maintenance", + relname))); + } + + /* + * Even if we can acquire an lock, a concurrent transaction could have + * updated the view incrementally and been committed before we acquired + * the lock. Therefore, we have to check the transaction ID of the most + * recent update of the view, and if this was in progress at the + * transaction start, raise an error to prevent anomalies. + */ + xid = getLastUpdateXid(matviewOid); + if (XidInMVCCSnapshot(XidFromFullTransactionId(xid), GetTransactionSnapshot())) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("the materialized view is incrementally updated in concurrent transaction"), + errhint("The transaction might succeed if retried."))); + } + else + LockRelationOid(matviewOid, RowExclusiveLock); + + /* + * On the first call initialize the hashtable + */ + if (!mv_trigger_info) + mv_InitHashTables(); + + entry = (MV_TriggerHashEntry *) hash_search(mv_trigger_info, + (void *) &matviewOid, + HASH_ENTER, &found); + + /* On the first BEFORE to update the view, initialize trigger data */ + if (!found || entry->snapshot == InvalidSnapshot) + { + Snapshot snapshot; + + /* + * Get a snapshot just before the table was modified for checking + * tuple visibility in the pre-update state of the table. + * + * In READ COMMITTED, use the latest snapshot again to see the + * results of concurrent transactions committed after the current + * transaction started. + */ + if (IsolationUsesXactSnapshot()) + snapshot = GetActiveSnapshot(); + else + snapshot = GetTransactionSnapshot(); + + entry->matview_id = matviewOid; + entry->before_trig_count = 0; + entry->after_trig_count = 0; + entry->snapshot = RegisterSnapshot(snapshot); + entry->tables = NIL; + entry->has_old = false; + entry->has_new = false; + + /* + * If this is the first table modifying query in the transaction, + * initialize the list of subxids. + */ + if (!found) + entry->subxids = NIL; + } + + entry->before_trig_count++; + + return PointerGetDatum(NULL); +} + +/* + * IVM_immediate_maintenance + * + * IVM trigger function invoked after base table is modified. + * For each table, tuplestores of transition tables are collected. + * and after the last modification + */ +Datum +IVM_immediate_maintenance(PG_FUNCTION_ARGS) +{ + TriggerData *trigdata = (TriggerData *) fcinfo->context; + Relation rel; + Oid relid; + Oid matviewOid; + Query *query; + Query *rewritten = NULL; + char *matviewOid_text = trigdata->tg_trigger->tgargs[0]; + Relation matviewRel; + int old_depth = matview_maintenance_depth; + SubTransactionId subxid; + + Oid relowner; + Tuplestorestate *old_tuplestore = NULL; + Tuplestorestate *new_tuplestore = NULL; + DestReceiver *dest_new = NULL, *dest_old = NULL; + Oid save_userid; + int save_sec_context; + int save_nestlevel; + + MV_TriggerHashEntry *entry; + MV_TriggerTable *table; + bool found; + + ParseState *pstate; + QueryEnvironment *queryEnv = create_queryEnv(); + MemoryContext oldcxt; + ListCell *lc; + int i; + + + /* Create a ParseState for rewriting the view definition query */ + pstate = make_parsestate(NULL); + pstate->p_queryEnv = queryEnv; + pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET; + + rel = trigdata->tg_relation; + relid = rel->rd_id; + + matviewOid = DatumGetObjectId(DirectFunctionCall1(oidin, CStringGetDatum(matviewOid_text))); + + /* + * On the first call initialize the hashtable + */ + if (!mv_trigger_info) + mv_InitHashTables(); + + /* get the entry for this materialized view */ + entry = (MV_TriggerHashEntry *) hash_search(mv_trigger_info, + (void *) &matviewOid, + HASH_FIND, &found); + Assert (found && entry != NULL); + entry->after_trig_count++; + + /* search the entry for the modified table and create new entry if not found */ + found = false; + foreach(lc, entry->tables) + { + table = (MV_TriggerTable *) lfirst(lc); + if (table->table_id == relid) + { + found = true; + break; + } + } + if (!found) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + + table = (MV_TriggerTable *) palloc0(sizeof(MV_TriggerTable)); + table->table_id = relid; + table->old_tuplestores = NIL; + table->new_tuplestores = NIL; + table->rte_indexes = NIL; + table->slot = MakeSingleTupleTableSlot(RelationGetDescr(rel), table_slot_callbacks(rel)); + /* We assume we have at least RowExclusiveLock on modified tables. */ + table->rel = table_open(RelationGetRelid(rel), NoLock); + entry->tables = lappend(entry->tables, table); + + MemoryContextSwitchTo(oldcxt); + } + + /* Save the transition tables and make a request to not free immediately */ + if (trigdata->tg_oldtable) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + table->old_tuplestores = lappend(table->old_tuplestores, trigdata->tg_oldtable); + entry->has_old = true; + MemoryContextSwitchTo(oldcxt); + } + if (trigdata->tg_newtable) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + table->new_tuplestores = lappend(table->new_tuplestores, trigdata->tg_newtable); + entry->has_new = true; + MemoryContextSwitchTo(oldcxt); + } + if (entry->has_new || entry->has_old) + { + CmdType cmd; + + if (TRIGGER_FIRED_BY_INSERT(trigdata->tg_event)) + cmd = CMD_INSERT; + else if (TRIGGER_FIRED_BY_DELETE(trigdata->tg_event)) + cmd = CMD_DELETE; + else if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event)) + cmd = CMD_UPDATE; + else + elog(ERROR,"unsupported trigger type"); + + /* Prolong lifespan of transition tables to the end of the last AFTER trigger */ + SetTransitionTablePreserved(relid, cmd); + } + + + /* If this is not the last AFTER trigger call, immediately exit. */ + Assert (entry->before_trig_count >= entry->after_trig_count); + if (entry->before_trig_count != entry->after_trig_count) + return PointerGetDatum(NULL); + + /* + * If this is the last AFTER trigger call, continue and update the view. + */ + + /* record the subxid that updated the view incrementally */ + subxid = GetCurrentSubTransactionId(); + if (!list_member_xid(entry->subxids, subxid)) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + entry->subxids = lappend_xid(entry->subxids, subxid); + MemoryContextSwitchTo(oldcxt); + } + + /* + * Advance command counter to make the updated base table rows locally + * visible. + */ + CommandCounterIncrement(); + + matviewRel = table_open(matviewOid, NoLock); + + /* Make sure it is a materialized view. */ + Assert(matviewRel->rd_rel->relkind == RELKIND_MATVIEW); + + /* + * In READ COMMITTED, get and push the latest snapshot again to see the + * results of concurrent transactions committed after the current + * transaction started. + */ + if (!IsolationUsesXactSnapshot()) + PushActiveSnapshot(GetTransactionSnapshot()); + + /* + * Check for active uses of the relation in the current transaction, such + * as open scans. + * + * NB: We count on this to protect us against problems with refreshing the + * data using TABLE_INSERT_FROZEN. + */ + CheckTableNotInUse(matviewRel, "refresh a materialized view incrementally"); + + /* + * Switch to the owner's userid, so that any functions are run as that + * user. Also arrange to make GUC variable changes local to this command. + * We will switch modes when we are about to execute user code. + */ + relowner = matviewRel->rd_rel->relowner; + GetUserIdAndSecContext(&save_userid, &save_sec_context); + SetUserIdAndSecContext(relowner, + save_sec_context | SECURITY_RESTRICTED_OPERATION); + save_nestlevel = NewGUCNestLevel(); + + /* get view query*/ + query = get_matview_query(matviewRel); + + /* + * When a base table is truncated, the view content will be empty if the + * view definition query does not contain an aggregate without a GROUP clause. + * Therefore, such views can be truncated. + */ + if (TRIGGER_FIRED_BY_TRUNCATE(trigdata->tg_event)) + { + ExecuteTruncateGuts(list_make1(matviewRel), list_make1_oid(matviewOid), + NIL, DROP_RESTRICT, false, false); + + /* Clean up hash entry and delete tuplestores */ + clean_up_IVM_hash_entry(entry, false, InvalidSubTransactionId); + + /* Pop the original snapshot. */ + if (!IsolationUsesXactSnapshot()) + PopActiveSnapshot(); + + table_close(matviewRel, NoLock); + + /* Roll back any GUC changes */ + AtEOXact_GUC(false, save_nestlevel); + + /* Restore userid and security context */ + SetUserIdAndSecContext(save_userid, save_sec_context); + + return PointerGetDatum(NULL); + } + + /* + * rewrite query for calculating deltas + */ + + rewritten = copyObject(query); + + /* Replace resnames in a target list with materialized view's attnames */ + i = 0; + foreach (lc, rewritten->targetList) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, i); + char *resname = NameStr(attr->attname); + + tle->resname = pstrdup(resname); + i++; + } + + /* Set all tables in the query to pre-update state */ + rewritten = rewrite_query_for_preupdate_state(rewritten, entry->tables, + pstate, matviewOid); + /* Rewrite for counting duplicated tuples */ + rewritten = rewrite_query_for_counting(rewritten, pstate); + + /* Create tuplestores to store view deltas */ + if (entry->has_old) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + + old_tuplestore = tuplestore_begin_heap(false, false, work_mem); + dest_old = CreateDestReceiver(DestTuplestore); + SetTuplestoreDestReceiverParams(dest_old, + old_tuplestore, + TopTransactionContext, + false, + NULL, + NULL); + + MemoryContextSwitchTo(oldcxt); + } + if (entry->has_new) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + + new_tuplestore = tuplestore_begin_heap(false, false, work_mem); + dest_new = CreateDestReceiver(DestTuplestore); + SetTuplestoreDestReceiverParams(dest_new, + new_tuplestore, + TopTransactionContext, + false, + NULL, + NULL); + MemoryContextSwitchTo(oldcxt); + } + + /* for all modified tables */ + foreach(lc, entry->tables) + { + ListCell *lc2; + + table = (MV_TriggerTable *) lfirst(lc); + + /* loop for self-join */ + foreach(lc2, table->rte_indexes) + { + int rte_index = lfirst_int(lc2); + TupleDesc tupdesc_old; + TupleDesc tupdesc_new; + + /* calculate delta tables */ + calc_delta(table, rte_index, rewritten, dest_old, dest_new, + &tupdesc_old, &tupdesc_new, queryEnv); + + /* Set the table in the query to post-update state */ + rewritten = rewrite_query_for_postupdate_state(rewritten, table, rte_index); + + PG_TRY(); + { + /* apply the delta tables to the materialized view */ + apply_delta(matviewOid, old_tuplestore, new_tuplestore, + tupdesc_old, tupdesc_new, query); + } + PG_CATCH(); + { + matview_maintenance_depth = old_depth; + PG_RE_THROW(); + } + PG_END_TRY(); + + /* clear view delta tuplestores */ + if (old_tuplestore) + tuplestore_clear(old_tuplestore); + if (new_tuplestore) + tuplestore_clear(new_tuplestore); + } + } + + /* Clean up hash entry and delete tuplestores */ + clean_up_IVM_hash_entry(entry, false, InvalidSubTransactionId); + if (old_tuplestore) + { + dest_old->rDestroy(dest_old); + tuplestore_end(old_tuplestore); + } + if (new_tuplestore) + { + dest_new->rDestroy(dest_new); + tuplestore_end(new_tuplestore); + } + + /* Pop the original snapshot. */ + if (!IsolationUsesXactSnapshot()) + PopActiveSnapshot(); + + table_close(matviewRel, NoLock); + + /* Roll back any GUC changes */ + AtEOXact_GUC(false, save_nestlevel); + + /* Restore userid and security context */ + SetUserIdAndSecContext(save_userid, save_sec_context); + + return PointerGetDatum(NULL); +} + +/* + * rewrite_query_for_preupdate_state + * + * Rewrite the query so that base tables' RTEs will represent "pre-update" + * state of tables. This is necessary to calculate view delta after multiple + * tables are modified. + */ +static Query* +rewrite_query_for_preupdate_state(Query *query, List *tables, + ParseState *pstate, Oid matviewid) +{ + ListCell *lc; + int num_rte = list_length(query->rtable); + int i; + + + /* register delta ENRs */ + register_delta_ENRs(pstate, query, tables); + + /* XXX: Is necessary? Is this right timing? */ + AcquireRewriteLocks(query, true, false); + + i = 1; + foreach(lc, query->rtable) + { + RangeTblEntry *r = (RangeTblEntry*) lfirst(lc); + + ListCell *lc2; + foreach(lc2, tables) + { + MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc2); + /* + * if the modified table is found then replace the original RTE with + * "pre-state" RTE and append its index to the list. + */ + if (r->relid == table->table_id) + { + List *securityQuals; + List *withCheckOptions; + bool hasRowSecurity; + bool hasSubLinks; + + RangeTblEntry *rte_pre = get_prestate_rte(r, table, pstate->p_queryEnv, matviewid); + + /* + * Set a row security poslicies of the modified table to the subquery RTE which + * represents the pre-update state of the table. + */ + get_row_security_policies(query, table->original_rte, i, + &securityQuals, &withCheckOptions, + &hasRowSecurity, &hasSubLinks); + + if (hasRowSecurity) + { + query->hasRowSecurity = true; + rte_pre->security_barrier = true; + } + if (hasSubLinks) + query->hasSubLinks = true; + + rte_pre->securityQuals = securityQuals; + lfirst(lc) = rte_pre; + + table->rte_indexes = lappend_int(table->rte_indexes, i); + break; + } + } + + /* finish the loop if we processed all RTE included in the original query */ + if (i++ >= num_rte) + break; + } + + return query; +} + +/* + * register_delta_ENRs + * + * For all modified tables, make ENRs for their transition tables + * and register them to the queryEnv. ENR's RTEs are also appended + * into the list in query tree. + */ +static void +register_delta_ENRs(ParseState *pstate, Query *query, List *tables) +{ + QueryEnvironment *queryEnv = pstate->p_queryEnv; + ListCell *lc; + RangeTblEntry *rte; + + foreach(lc, tables) + { + MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc); + ListCell *lc2; + int count; + + count = 0; + foreach(lc2, table->old_tuplestores) + { + Tuplestorestate *oldtable = (Tuplestorestate *) lfirst(lc2); + EphemeralNamedRelation enr = + palloc(sizeof(EphemeralNamedRelationData)); + ParseNamespaceItem *nsitem; + + enr->md.name = make_delta_enr_name("old", table->table_id, count); + enr->md.reliddesc = table->table_id; + enr->md.tupdesc = NULL; + enr->md.enrtype = ENR_NAMED_TUPLESTORE; + enr->md.enrtuples = tuplestore_tuple_count(oldtable); + enr->reldata = oldtable; + register_ENR(queryEnv, enr); + + nsitem = addRangeTableEntryForENR(pstate, makeRangeVar(NULL, enr->md.name, -1), true); + rte = nsitem->p_rte; + + query->rtable = lappend(query->rtable, rte); + + count++; + } + + count = 0; + foreach(lc2, table->new_tuplestores) + { + Tuplestorestate *newtable = (Tuplestorestate *) lfirst(lc2); + EphemeralNamedRelation enr = + palloc(sizeof(EphemeralNamedRelationData)); + ParseNamespaceItem *nsitem; + + enr->md.name = make_delta_enr_name("new", table->table_id, count); + enr->md.reliddesc = table->table_id; + enr->md.tupdesc = NULL; + enr->md.enrtype = ENR_NAMED_TUPLESTORE; + enr->md.enrtuples = tuplestore_tuple_count(newtable); + enr->reldata = newtable; + register_ENR(queryEnv, enr); + + nsitem = addRangeTableEntryForENR(pstate, makeRangeVar(NULL, enr->md.name, -1), true); + rte = nsitem->p_rte; + + query->rtable = lappend(query->rtable, rte); + + count++; + } + } +} + +#define DatumGetItemPointer(X) ((ItemPointer) DatumGetPointer(X)) +#define PG_GETARG_ITEMPOINTER(n) DatumGetItemPointer(PG_GETARG_DATUM(n)) + +/* + * ivm_visible_in_prestate + * + * Check visibility of a tuple specified by the tableoid and item pointer + * using the snapshot taken just before the table was modified. + */ +Datum +ivm_visible_in_prestate(PG_FUNCTION_ARGS) +{ + Oid tableoid = PG_GETARG_OID(0); + ItemPointer itemPtr = PG_GETARG_ITEMPOINTER(1); + Oid matviewOid = PG_GETARG_OID(2); + MV_TriggerHashEntry *entry; + MV_TriggerTable *table = NULL; + ListCell *lc; + bool found; + bool result; + + if (!in_delta_calculation) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("ivm_visible_in_prestate can be called only in delta calculation"))); + + entry = (MV_TriggerHashEntry *) hash_search(mv_trigger_info, + (void *) &matviewOid, + HASH_FIND, &found); + Assert (found && entry != NULL); + + foreach(lc, entry->tables) + { + table = (MV_TriggerTable *) lfirst(lc); + if (table->table_id == tableoid) + break; + } + + Assert (table != NULL); + + result = table_tuple_fetch_row_version(table->rel, itemPtr, entry->snapshot, table->slot); + + PG_RETURN_BOOL(result); +} + +/* + * get_prestate_rte + * + * Rewrite RTE of the modified table to a subquery which represents + * "pre-state" table. The original RTE is saved in table->rte_original. + */ +static RangeTblEntry* +get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table, + QueryEnvironment *queryEnv, Oid matviewid) +{ + StringInfoData str; + RawStmt *raw; + Query *subquery; + ParseState *pstate; + char *relname; + static char *subquery_tl; + int i; + + pstate = make_parsestate(NULL); + pstate->p_queryEnv = queryEnv; + pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET; + + relname = quote_qualified_identifier( + get_namespace_name(RelationGetNamespace(table->rel)), + RelationGetRelationName(table->rel)); + + subquery_tl = make_subquery_targetlist_from_table(table); + + /* + * Filtering inserted row using the snapshot taken before the table + * is modified. ctid is required for maintaining outer join views. + */ + initStringInfo(&str); + appendStringInfo(&str, + "SELECT %s FROM %s t" + " WHERE pg_catalog.ivm_visible_in_prestate(t.tableoid, t.ctid ,%d::pg_catalog.oid)", + subquery_tl, relname, matviewid); + + /* + * Re-add rows deleted from the old transition tables, excluding those + * also present in the new transition tables. + */ + if (list_length(table->old_tuplestores) > 0) + { + appendStringInfo(&str," UNION ALL SELECT %s FROM (", + subquery_tl); + + for (i = 0; i < list_length(table->old_tuplestores); i++) + { + if (i != 0) + appendStringInfo(&str, " UNION ALL "); + appendStringInfo(&str," TABLE %s", + make_delta_enr_name("old", table->table_id, i)); + } + for (i = 0; i < list_length(table->new_tuplestores); i++) + { + appendStringInfo(&str, " EXCEPT ALL "); + appendStringInfo(&str," TABLE %s", + make_delta_enr_name("new", table->table_id, i)); + } + appendStringInfo(&str,")"); + } + + /* Get a subquery representing pre-state of the table */ + raw = (RawStmt*)linitial(raw_parser(str.data, RAW_PARSE_DEFAULT)); + subquery = transformStmt(pstate, raw->stmt); + + /* save the original RTE */ + table->original_rte = copyObject(rte); + + rte->rtekind = RTE_SUBQUERY; + rte->subquery = subquery; + rte->security_barrier = false; + + /* Clear fields that should not be set in a subquery RTE */ + rte->relid = InvalidOid; + rte->relkind = 0; + rte->rellockmode = 0; + rte->tablesample = NULL; + rte->perminfoindex = 0; /* no permission checking for this RTE */ + rte->inh = false; /* must not be set for a subquery */ + + return rte; +} + +/* + * make_subquery_targetlist_from_table + * + * Make a targetlist string of a subquery representing a delta table or a + * pre-update state table. This subquery substitutes a modified table RTE + * in the view definition query during view maintenance. In the targetlist, + * column names appear in order of the table definition. However, for + * attribute numbers of vars in the query tree to reference columns of the + * subquery correctly even though the table has a dropped column, put "null" + * as a dummy value at the position of a dropped column. + * + * We would also able to walk the query tree to rewrite varattnos, but + * crafting targetlist is more simple and reasonable. + */ +static char* +make_subquery_targetlist_from_table(MV_TriggerTable *table) +{ + StringInfoData str; + TupleDesc tupdesc; + int i; + + tupdesc = RelationGetDescr(table->rel); + initStringInfo(&str); + for (i = 0; i < tupdesc->natts; i++) + { + Form_pg_attribute attr = TupleDescAttr(tupdesc, i); + + if (i > 0) + appendStringInfo(&str, ", "); + + if (attr->attisdropped) + appendStringInfo(&str, "null"); + else + appendStringInfo(&str, "%s", quote_identifier(NameStr(attr->attname))); + } + + return str.data; +} + +/* + * make_delta_enr_name + * + * Make a name for ENR of a transition table from the base table's oid. + * prefix will be "new" or "old" depending on its transition table kind.. + */ +static char* +make_delta_enr_name(const char *prefix, Oid relid, int count) +{ + char buf[NAMEDATALEN]; + char *name; + + snprintf(buf, NAMEDATALEN, "__ivm_%s_%u_%u", prefix, relid, count); + name = pstrdup(buf); + + return name; +} + +/* + * makeDeltaTable + * + * Make a RTE representing a delta of the specified table. + */ +static RangeTblEntry* +makeDeltaTable(RangeTblEntry *rte, MV_TriggerTable *table, + bool is_new, QueryEnvironment *queryEnv) +{ + StringInfoData str; + ParseState *pstate; + RawStmt *raw; + Query *sub; + int i; + + const char *prefix_union = is_new ? "new" : "old"; + const char *prefix_except = is_new ? "old" : "new"; + int num_union = is_new ? list_length(table->new_tuplestores) : list_length(table->old_tuplestores); + int num_except = is_new ? list_length(table->old_tuplestores) : list_length(table->new_tuplestores); + + /* the previous RTE must be a subquery which represents "pre-state" table */ + Assert(rte->rtekind == RTE_SUBQUERY); + + /* Create a ParseState for rewriting the view definition query */ + pstate = make_parsestate(NULL); + pstate->p_queryEnv = queryEnv; + pstate->p_expr_kind = EXPR_KIND_NONE; + + initStringInfo(&str); + appendStringInfo(&str, + " SELECT %s FROM ( ", + make_subquery_targetlist_from_table(table)); + + for (i = 0; i < num_union; i++) + { + if (i > 0) + appendStringInfo(&str, " UNION ALL "); + + appendStringInfo(&str, + " TABLE %s", + make_delta_enr_name(prefix_union, table->table_id, i)); + } + for (i = 0; i < num_except; i++) + { + appendStringInfo(&str, " EXCEPT ALL "); + + appendStringInfo(&str, + " TABLE %s", + make_delta_enr_name(prefix_except, table->table_id, i)); + } + appendStringInfo(&str,")"); + + raw = (RawStmt*)linitial(raw_parser(str.data, RAW_PARSE_DEFAULT)); + sub = transformStmt(pstate, raw->stmt); + + /* + * Update the subquery so that it represent the combined transition + * table. Note that we leave the security_barrier and securityQuals + * fields so that the subquery relation can be protected by the RLS + * policy as same as the modified table. + */ + rte->rtekind = RTE_SUBQUERY; + rte->subquery = sub; + + return rte; +} + +/* + * rewrite_query_for_counting + * + * Rewrite query for counting duplicated tuples. + */ +static Query * +rewrite_query_for_counting(Query *query, ParseState *pstate) +{ + TargetEntry *tle_count; + FuncCall *fn; + Node *node; + + /* Add count(*) for counting distinct tuples in views */ + fn = makeFuncCall(SystemFuncName("count"), NIL, COERCE_EXPLICIT_CALL, -1); + fn->agg_star = true; + if (!query->groupClause && !query->hasAggs) + query->groupClause = transformDistinctClause(NULL, &query->targetList, query->sortClause, false); + + node = ParseFuncOrColumn(pstate, fn->funcname, NIL, NULL, fn, false, -1); + + tle_count = makeTargetEntry((Expr *) node, + list_length(query->targetList) + 1, + pstrdup("__ivm_count__"), + false); + query->targetList = lappend(query->targetList, tle_count); + query->hasAggs = true; + + return query; +} + +/* + * calc_delta + * + * Calculate view deltas generated under the modification of a table specified + * by the RTE index. + */ +static void +calc_delta(MV_TriggerTable *table, int rte_index, Query *query, + DestReceiver *dest_old, DestReceiver *dest_new, + TupleDesc *tupdesc_old, TupleDesc *tupdesc_new, + QueryEnvironment *queryEnv) +{ + ListCell *lc = list_nth_cell(query->rtable, rte_index - 1); + RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc); + + in_delta_calculation = true; + + /* Generate old delta */ + if (dest_old && list_length(table->old_tuplestores) > 0) + { + /* Replace the modified table with the old delta table and calculate the old view delta. */ + lfirst(lc) = makeDeltaTable(rte, table, false, queryEnv); + refresh_matview_datafill(dest_old, query, queryEnv, tupdesc_old, "", false); + } + + /* Generate new delta */ + if (dest_new && list_length(table->new_tuplestores) > 0) + { + /* Replace the modified table with the new delta table and calculate the new view delta*/ + lfirst(lc) = makeDeltaTable(rte, table, true, queryEnv); + refresh_matview_datafill(dest_new, query, queryEnv, tupdesc_new, "", false); + } + + in_delta_calculation = false; +} + +/* + * rewrite_query_for_postupdate_state + * + * Rewrite the query so that the specified base table's RTEs will represent + * "post-update" state of tables. This is called after the view delta + * calculation due to changes on this table finishes. + */ +static Query* +rewrite_query_for_postupdate_state(Query *query, MV_TriggerTable *table, int rte_index) +{ + ListCell *lc = list_nth_cell(query->rtable, rte_index - 1); + + /* Retore the original RTE */ + lfirst(lc) = table->original_rte; + + return query; +} + +/* + * apply_delta + * + * Apply deltas to the materialized view. In outer join cases, this requires + * the view maintenance graph. + */ +static void +apply_delta(Oid matviewOid, Tuplestorestate *old_tuplestores, Tuplestorestate *new_tuplestores, + TupleDesc tupdesc_old, TupleDesc tupdesc_new, + Query *query) +{ + StringInfoData querybuf; + StringInfoData target_list_buf; + Relation matviewRel; + char *matviewname; + ListCell *lc; + int i; + List *keys = NIL; + + + /* + * get names of the materialized view and delta tables + */ + + matviewRel = table_open(matviewOid, NoLock); + matviewname = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(matviewRel)), + RelationGetRelationName(matviewRel)); + + /* + * Build parts of the maintenance queries + */ + + initStringInfo(&querybuf); + initStringInfo(&target_list_buf); + + /* build string of target list */ + for (i = 0; i < matviewRel->rd_att->natts; i++) + { + Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, i); + char *resname = NameStr(attr->attname); + + if (i != 0) + appendStringInfo(&target_list_buf, ", "); + appendStringInfo(&target_list_buf, "%s", quote_qualified_identifier(NULL, resname)); + } + + i = 0; + foreach (lc, query->targetList) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, i); + + i++; + + if (tle->resjunk) + continue; + + keys = lappend(keys, attr); + } + + /* Start maintaining the materialized view. */ + OpenMatViewIncrementalMaintenance(); + + /* Open SPI context. */ + if (SPI_connect() != SPI_OK_CONNECT) + elog(ERROR, "SPI_connect failed"); + + /* For tuple deletion */ + if (old_tuplestores && tuplestore_tuple_count(old_tuplestores) > 0) + { + EphemeralNamedRelation enr = palloc(sizeof(EphemeralNamedRelationData)); + int rc; + + /* convert tuplestores to ENR, and register for SPI */ + enr->md.name = pstrdup(OLD_DELTA_ENRNAME); + enr->md.reliddesc = InvalidOid; + enr->md.tupdesc = tupdesc_old; + enr->md.enrtype = ENR_NAMED_TUPLESTORE; + enr->md.enrtuples = tuplestore_tuple_count(old_tuplestores); + enr->reldata = old_tuplestores; + + rc = SPI_register_relation(enr); + if (rc != SPI_OK_REL_REGISTER) + elog(ERROR, "SPI_register failed"); + + apply_old_delta(matviewname, OLD_DELTA_ENRNAME, keys); + + } + /* For tuple insertion */ + if (new_tuplestores && tuplestore_tuple_count(new_tuplestores) > 0) + { + EphemeralNamedRelation enr = palloc(sizeof(EphemeralNamedRelationData)); + int rc; + + /* convert tuplestores to ENR, and register for SPI */ + enr->md.name = pstrdup(NEW_DELTA_ENRNAME); + enr->md.reliddesc = InvalidOid; + enr->md.tupdesc = tupdesc_new;; + enr->md.enrtype = ENR_NAMED_TUPLESTORE; + enr->md.enrtuples = tuplestore_tuple_count(new_tuplestores); + enr->reldata = new_tuplestores; + + rc = SPI_register_relation(enr); + if (rc != SPI_OK_REL_REGISTER) + elog(ERROR, "SPI_register failed"); + + /* apply new delta */ + apply_new_delta(matviewname, NEW_DELTA_ENRNAME, &target_list_buf); + } + + /* We're done maintaining the materialized view. */ + CloseMatViewIncrementalMaintenance(); + + table_close(matviewRel, NoLock); + + /* Close SPI context. */ + if (SPI_finish() != SPI_OK_FINISH) + elog(ERROR, "SPI_finish failed"); +} + +/* + * apply_old_delta + * + * Execute a query for applying a delta table given by deltname_old + * which contains tuples to be deleted from to a materialized view given by + * matviewname. This is used when counting is not required. + */ +static void +apply_old_delta(const char *matviewname, const char *deltaname_old, + List *keys) +{ + StringInfoData querybuf; + StringInfoData keysbuf; + char *match_cond; + ListCell *lc; + + /* build WHERE condition for searching tuples to be deleted */ + match_cond = get_matching_condition_string(keys); + + /* build string of keys list */ + initStringInfo(&keysbuf); + foreach (lc, keys) + { + Form_pg_attribute attr = (Form_pg_attribute) lfirst(lc); + char *resname = NameStr(attr->attname); + appendStringInfo(&keysbuf, "%s", quote_qualified_identifier("mv", resname)); + if (lnext(keys, lc)) + appendStringInfo(&keysbuf, ", "); + } + + /* Search for matching tuples from the view and update or delete if found. */ + initStringInfo(&querybuf); + appendStringInfo(&querybuf, + "DELETE FROM %s WHERE ctid IN (" + "SELECT tid FROM (SELECT pg_catalog.row_number() over (partition by %s) AS \"__ivm_row_number__\"," + "mv.ctid AS tid," + "diff.\"__ivm_count__\"" + "FROM %s AS mv, %s AS diff " + "WHERE %s) v " + "WHERE v.\"__ivm_row_number__\" OPERATOR(pg_catalog.<=) v.\"__ivm_count__\")", + matviewname, + keysbuf.data, + matviewname, deltaname_old, + match_cond); + + if (SPI_exec(querybuf.data, 0) != SPI_OK_DELETE) + elog(ERROR, "SPI_exec failed: %s", querybuf.data); +} + +/* + * apply_new_delta + * + * Execute a query for applying a delta table given by deltname_new + * which contains tuples to be inserted into a materialized view given by + * matviewname. This is used when counting is not required. + */ +static void +apply_new_delta(const char *matviewname, const char *deltaname_new, + StringInfo target_list) +{ + StringInfoData querybuf; + + /* Search for matching tuples from the view and update or delete if found. */ + initStringInfo(&querybuf); + appendStringInfo(&querybuf, + "INSERT INTO %s (%s) SELECT %s FROM (" + "SELECT diff.*, pg_catalog.generate_series(1, diff.\"__ivm_count__\")" + " AS __ivm_generate_series__ " + "FROM %s AS diff) AS v", + matviewname, target_list->data, target_list->data, + deltaname_new); + + if (SPI_exec(querybuf.data, 0) != SPI_OK_INSERT) + elog(ERROR, "SPI_exec failed: %s", querybuf.data); +} + +/* + * get_matching_condition_string + * + * Build a predicate string for looking for a tuple with given keys. + */ +static char * +get_matching_condition_string(List *keys) +{ + StringInfoData match_cond; + ListCell *lc; + + /* If there is no key columns, the condition is always true. */ + if (keys == NIL) + return "true"; + + initStringInfo(&match_cond); + foreach (lc, keys) + { + Form_pg_attribute attr = (Form_pg_attribute) lfirst(lc); + char *resname = NameStr(attr->attname); + char *mv_resname = quote_qualified_identifier("mv", resname); + char *diff_resname = quote_qualified_identifier("diff", resname); + Oid typid = attr->atttypid; + + /* Considering NULL values, we can not use simple = operator. */ + appendStringInfo(&match_cond, "("); + generate_equal(&match_cond, typid, mv_resname, diff_resname); + appendStringInfo(&match_cond, " OR (%s IS NULL AND %s IS NULL))", + mv_resname, diff_resname); + + if (lnext(keys, lc)) + appendStringInfo(&match_cond, " AND "); + } + + return match_cond.data; +} + +/* + * generate_equals + * + * Generate an equality clause using given operands' default equality + * operator. + */ +static void +generate_equal(StringInfo querybuf, Oid opttype, + const char *leftop, const char *rightop) +{ + TypeCacheEntry *typentry; + + typentry = lookup_type_cache(opttype, TYPECACHE_EQ_OPR); + if (!OidIsValid(typentry->eq_opr)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_FUNCTION), + errmsg("could not identify an equality operator for type %s", + format_type_be_qualified(opttype)))); + + generate_operator_clause(querybuf, + leftop, opttype, + typentry->eq_opr, + rightop, opttype); +} + +/* + * mv_InitHashTables + */ +static void +mv_InitHashTables(void) +{ + HASHCTL ctl; + + memset(&ctl, 0, sizeof(ctl)); + ctl.keysize = sizeof(Oid); + ctl.entrysize = sizeof(MV_TriggerHashEntry); + mv_trigger_info = hash_create("MV trigger info", + MV_INIT_QUERYHASHSIZE, + &ctl, HASH_ELEM | HASH_BLOBS); + + memset(&ctl, 0, sizeof(ctl)); + ctl.keysize = sizeof(Oid); + ctl.entrysize = sizeof(DroppedImmvInfo); + dropped_immv = hash_create("Dropped IMMVs", 16, + &ctl, HASH_ELEM | HASH_BLOBS); +} + +/* + * AtAbort_IVM + * + * Clean up hash entries for all materialized views. This is called at + * (sub-)transaction abort. When the top-level transaction is aborted, + * InvalidSubTransactionId is set to subxid. + * + * Also, remove dropped IMMV information if it is aborted. + */ +void +AtAbort_IVM(SubTransactionId subxid) +{ + HASH_SEQ_STATUS seq; + + if (mv_trigger_info) + { + MV_TriggerHashEntry *entry; + hash_seq_init(&seq, mv_trigger_info); + while ((entry = hash_seq_search(&seq)) != NULL) + clean_up_IVM_hash_entry(entry, true, subxid); + } + + if (dropped_immv) + { + DroppedImmvInfo *entry; + hash_seq_init(&seq, dropped_immv); + while ((entry = hash_seq_search(&seq)) != NULL) + { + if (subxid == InvalidSubTransactionId || subxid == entry->subxid) + hash_search(dropped_immv, (void *) &entry->immv_oid, HASH_REMOVE, NULL); + } + } + + in_delta_calculation = false; +} + +/* + * AtPreCommit_IVM + * + * Record the transaction ID that updated the view incrementally. + * Also, remove the entry for dropped IMMVs. + */ +void +AtPreCommit_IVM(void) +{ + HASH_SEQ_STATUS seq; + + if (mv_trigger_info) + { + MV_TriggerHashEntry *entry; + + /* + * Record the transaction ID that udpate the view incrementally, + * and perform the final clean up of the entry. + */ + hash_seq_init(&seq, mv_trigger_info); + while ((entry = hash_seq_search(&seq)) != NULL) + { + setLastUpdateXid(entry->matview_id, GetTopFullTransactionId()); + hash_search(mv_trigger_info, (void *) &entry->matview_id, HASH_REMOVE, NULL); + } + } + + if (dropped_immv) + { + DroppedImmvInfo *entry; + + LWLockAcquire(IvmControlLock, LW_EXCLUSIVE); + hash_seq_init(&seq, dropped_immv); + while ((entry = hash_seq_search(&seq)) != NULL) + { + hash_search(LastIvmUpdateHash, (void *) &entry->immv_oid, HASH_REMOVE, NULL); + hash_search(dropped_immv, (void *) &entry->immv_oid, HASH_REMOVE, NULL); + } + LWLockRelease(IvmControlLock); + } + + in_delta_calculation = false; +} + +/* + * clean_up_IVM_hash_entry + * + * Clean up tuple stores and hash entries for a materialized view after its + * maintenance finished. This is called at the end of table modifying query + * or (sub-)transaction abort. When the top-level transaction is aborted, + * InvalidSubTransactionId is set to subxid. + */ +static void +clean_up_IVM_hash_entry(MV_TriggerHashEntry *entry, bool is_abort, + SubTransactionId subxid) +{ + bool found; + ListCell *lc; + + /* clean up tuple stores */ + foreach(lc, entry->tables) + { + MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc); + + list_free(table->old_tuplestores); + list_free(table->new_tuplestores); + if (!is_abort) + { + ExecDropSingleTupleTableSlot(table->slot); + table_close(table->rel, NoLock); + } + } + list_free(entry->tables); + entry->tables = NIL; + + if (is_abort) + { + bool remove_entry = false; + + /* + * When the top-level transaction is aborted, remove all subxids. + * When a sub-transaction is aborted, remove only its subxid. + */ + if (subxid == InvalidSubTransactionId) + remove_entry = true; + else + { + foreach(lc, entry->subxids) + { + if (lfirst_xid(lc) == subxid) + { + entry->subxids = list_delete_cell(entry->subxids, lc); + break; + } + } + + /* + * If all the subxid are removed, it means that the view was not + * updated at all in this transaction. + */ + if (list_length(entry->subxids) == 0) + remove_entry = true; + } + + /* + * Remove entries of not updated views from the hash table. + */ + if (remove_entry) + hash_search(mv_trigger_info, (void *) &entry->matview_id, HASH_REMOVE, &found); + } + else + { + /* When the query sucsessully finished, unregister the snapshot */ + UnregisterSnapshot(entry->snapshot); + } + + entry->snapshot = InvalidSnapshot; +} + +/* + * setLastUpdateXid + * + * Store the transaction ID that updated the view incremenally. + */ +static void +setLastUpdateXid(Oid immv_oid, FullTransactionId xid) +{ + LastIvmUpdateEntry *entry; + + LWLockAcquire(IvmControlLock, LW_EXCLUSIVE); + entry = (LastIvmUpdateEntry *) hash_search(LastIvmUpdateHash, (void *) &immv_oid, HASH_ENTER, NULL); + entry->oid = immv_oid; + entry->last_ivm_update = xid; + LWLockRelease(IvmControlLock); +} + +/* + * getLastUpdateXid + * + * Get the most recent transaction ID that updated the view incrementally. + */ +static FullTransactionId +getLastUpdateXid(Oid immv_oid) +{ + FullTransactionId xid = InvalidFullTransactionId; + LastIvmUpdateEntry *entry; + bool found; + + LWLockAcquire(IvmControlLock, LW_SHARED); + entry = (LastIvmUpdateEntry *) hash_search(LastIvmUpdateHash, (void *) &immv_oid, HASH_FIND, &found); + LWLockRelease(IvmControlLock); + if (found) + xid = entry->last_ivm_update; + + return xid; +} + +void +removeImmv(Oid immv_oid) +{ + DroppedImmvInfo *entry; + + if (!dropped_immv) + mv_InitHashTables(); + + entry = (DroppedImmvInfo *) hash_search(dropped_immv, (void *) &immv_oid, HASH_ENTER, NULL); + entry->subxid = GetCurrentSubTransactionId(); +} + +/* + * isIvmName + * + * Check if this is a IVM hidden column from the name. + */ +bool +isIvmName(const char *s) +{ + if (s) + return (strncmp(s, "__ivm_", 6) == 0); + return false; +} diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 472db112fa7..920b62d254d 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -62,6 +62,7 @@ #include "commands/defrem.h" #include "commands/event_trigger.h" #include "commands/extension.h" +#include "commands/matview.h" #include "commands/repack.h" #include "commands/sequence.h" #include "commands/tablecmds.h" @@ -1811,6 +1812,9 @@ RangeVarCallbackForDropRelation(const RangeVar *rel, Oid relOid, Oid oldRelOid, state->actual_relkind = classform->relkind; state->actual_relpersistence = classform->relpersistence; + if (classform->relkind == RELKIND_MATVIEW && classform->relisivm) + removeImmv(relOid); + /* * Both RELKIND_RELATION and RELKIND_PARTITIONED_TABLE are OBJECT_TABLE, * but RemoveRelations() can only pass one relkind for a given relation. diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 560659f9568..d97c35eeef0 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -370,6 +370,7 @@ WaitLSN "Waiting to read or update shared Wait-for-LSN state." LogicalDecodingControl "Waiting to read or update logical decoding status information." DataChecksumsWorker "Waiting for data checksums worker." AioWorkerControl "Waiting to update AIO worker information." +IvmControl "Waiting for read or update IMMV information.." # # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE) diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 73bb7fbb430..88399513f87 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -12698,4 +12698,14 @@ proname => 'hashoid8extended', prorettype => 'int8', proargtypes => 'oid8 int8', prosrc => 'hashoid8extended' }, +# IVM +{ oid => '786', descr => 'ivm trigger (before)', + proname => 'IVM_immediate_before', provolatile => 'v', prorettype => 'trigger', + proargtypes => '', prosrc => 'IVM_immediate_before' }, +{ oid => '787', descr => 'ivm trigger (after)', + proname => 'IVM_immediate_maintenance', provolatile => 'v', prorettype => 'trigger', + proargtypes => '', prosrc => 'IVM_immediate_maintenance' }, +{ oid => '788', descr => 'ivm filetring ', + proname => 'ivm_visible_in_prestate', provolatile => 's', prorettype => 'bool', + proargtypes => 'oid tid oid', prosrc => 'ivm_visible_in_prestate' }, ] diff --git a/src/include/commands/createas.h b/src/include/commands/createas.h index f895c67c0c2..c286ebcd70e 100644 --- a/src/include/commands/createas.h +++ b/src/include/commands/createas.h @@ -16,6 +16,7 @@ #include "catalog/objectaddress.h" #include "nodes/params.h" +#include "nodes/pathnodes.h" #include "parser/parse_node.h" #include "tcop/dest.h" #include "utils/queryenvironment.h" @@ -25,6 +26,9 @@ extern ObjectAddress ExecCreateTableAs(ParseState *pstate, CreateTableAsStmt *st ParamListInfo params, QueryEnvironment *queryEnv, QueryCompletion *qc); +extern void CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid); +extern void CreateIndexOnIMMV(Query *query, Relation matviewRel); + extern int GetIntoRelEFlags(IntoClause *intoClause); extern DestReceiver *CreateIntoRelDestReceiver(IntoClause *intoClause); diff --git a/src/include/commands/matview.h b/src/include/commands/matview.h index 738c731c1a9..f1dbfc052c5 100644 --- a/src/include/commands/matview.h +++ b/src/include/commands/matview.h @@ -15,6 +15,7 @@ #define MATVIEW_H #include "catalog/objectaddress.h" +#include "fmgr.h" #include "nodes/params.h" #include "nodes/parsenodes.h" #include "tcop/dest.h" @@ -23,6 +24,8 @@ extern void SetMatViewPopulatedState(Relation relation, bool newstate); +extern void SetMatViewIVMState(Relation relation, bool newstate); + extern ObjectAddress ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString, QueryCompletion *qc); extern ObjectAddress RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, @@ -33,4 +36,12 @@ extern DestReceiver *CreateTransientRelDestReceiver(Oid transientoid); extern bool MatViewIncrementalMaintenanceIsEnabled(void); +extern Datum IVM_immediate_before(PG_FUNCTION_ARGS); +extern Datum IVM_immediate_maintenance(PG_FUNCTION_ARGS); +extern Datum IVM_visible_in_prestate(PG_FUNCTION_ARGS); +extern void AtAbort_IVM(SubTransactionId subtxid); +extern void AtPreCommit_IVM(void); +extern void removeImmv(Oid immv_oid); +extern bool isIvmName(const char *s); + #endif /* MATVIEW_H */ diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h index d7eb648bd27..756c63baf97 100644 --- a/src/include/storage/lwlocklist.h +++ b/src/include/storage/lwlocklist.h @@ -89,6 +89,7 @@ PG_LWLOCK(54, WaitLSN) PG_LWLOCK(55, LogicalDecodingControl) PG_LWLOCK(56, DataChecksumsWorker) PG_LWLOCK(57, AioWorkerControl) +PG_LWLOCK(58, IvmControl) /* * There also exist several built-in LWLock tranches. As with the predefined diff --git a/src/include/storage/subsystemlist.h b/src/include/storage/subsystemlist.h index 9ad619080be..481e15d00a1 100644 --- a/src/include/storage/subsystemlist.h +++ b/src/include/storage/subsystemlist.h @@ -88,3 +88,6 @@ PG_SHMEM_SUBSYSTEM(DataChecksumsShmemCallbacks) /* AIO subsystem. This delegates to the method-specific callbacks */ PG_SHMEM_SUBSYSTEM(AioShmemCallbacks) + +/* Incremental View Maintenance */ +PG_SHMEM_SUBSYSTEM(IvmShmemCallbacks) -- 2.43.0 --Multipart=_Wed__1_Jul_2026_00_04_01_+0900_OVSy2WWK_9aByzDJ Content-Type: text/x-diff; name="v38-0005-Add-Incremental-View-Maintenance-support-to-psql.patch" Content-Disposition: attachment; filename="v38-0005-Add-Incremental-View-Maintenance-support-to-psql.patch" Content-Transfer-Encoding: 7bit ^ permalink raw reply [nested|flat] 14+ messages in thread
end of thread, other threads:[~2026-05-29 09:06 UTC | newest] Thread overview: 14+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2022-04-28 10:01 RE: Logical replication timeout problem [email protected] <[email protected]> 2022-05-02 02:03 ` Masahiko Sawada <[email protected]> 2022-05-02 02:32 ` Amit Kapila <[email protected]> 2022-05-02 02:36 ` Masahiko Sawada <[email protected]> 2022-05-04 10:18 ` Amit Kapila <[email protected]> 2022-05-06 01:53 ` Masahiko Sawada <[email protected]> 2022-05-06 07:11 ` [email protected] <[email protected]> 2022-05-09 06:47 ` Amit Kapila <[email protected]> 2022-05-09 08:46 ` Masahiko Sawada <[email protected]> 2022-05-11 11:33 ` Amit Kapila <[email protected]> 2022-05-09 13:30 ` Euler Taveira <[email protected]> 2022-05-10 03:22 ` Amit Kapila <[email protected]> 2022-05-10 03:36 ` [email protected] <[email protected]> 2026-05-29 09:06 [PATCH v38 06/11] Add Incremental View Maintenance support Yugo Nagata <[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