public inbox for [email protected]
help / color / mirror / Atom feedFrom: Drouvot, Bertrand <[email protected]>
To: Andres Freund <[email protected]>
Cc: [email protected]
Cc: Alvaro Herrera <[email protected]>
Cc: tushar <[email protected]>
Cc: [pgdg] Robert Haas <[email protected]>
Cc: Amit Khandekar <[email protected]>
Cc: Rahila Syed <[email protected]>
Cc: pgsql-hackers <[email protected]>
Subject: Re: [UNVERIFIED SENDER] Re: Minimal logical decoding on standbys
Date: Tue, 22 Jun 2021 12:38:34 +0200
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <CAFcNs+q9FvSW7gMqM__aeWOoDNqfcQCoB8LiO-X33fvqGO8mkQ@mail.gmail.com>
<[email protected]>
<CAFcNs+pyDR+iwYpTAp=cnkX7BgEbOnF6rrr8w=2inPAn4rQB-A@mail.gmail.com>
<[email protected]>
<CAFcNs+pQ0jBvUUH8XzE9ct64Doy-pJsfiT13StsaLpmviPAa8Q@mail.gmail.com>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
Hi Andres,
On 6/14/21 7:41 AM, Drouvot, Bertrand wrote:
> Hi Andres,
>
> On 4/8/21 5:47 AM, Andres Freund wrote:
>> Hi,
>>
>> On 2021-04-07 13:32:18 -0700, Andres Freund wrote:
>>> While working on this I found a, somewhat substantial, issue:
>>>
>>> When the primary is idle, on the standby logical decoding via walsender
>>> will typically not process the records until further WAL writes come in
>>> from the primary, or until a 10s lapsed.
>>>
>>> The problem is that WalSndWaitForWal() waits for the *replay* LSN to
>>> increase, but gets woken up by walreceiver when new WAL has been
>>> flushed. Which means that typically walsenders will get woken up at the
>>> same time that the startup process will be - which means that by the
>>> time the logical walsender checks GetXLogReplayRecPtr() it's unlikely
>>> that the startup process already replayed the record and updated
>>> XLogCtl->lastReplayedEndRecPtr.
>>>
>>> I think fixing this would require too invasive changes at this point. I
>>> think we might be able to live with 10s delay issue for one release,
>>> but
>>> it sure is ugly :(.
>> This is indeed pretty painful. It's a lot more regularly occuring if you
>> either have a slot disk, or you switch around the order of
>> WakeupRecovery() and WalSndWakeup() XLogWalRcvFlush().
>>
>> - There's about which timeline to use. If you use pg_recvlogical and you
>> restart the server, you'll see errors like:
>>
>> pg_recvlogical: error: unexpected termination of replication
>> stream: ERROR: requested WAL segment 000000000000000000000003 has
>> already been removed
>>
>> the real filename is 000000010000000000000003 - i.e. the timeline is
>> 0.
>>
>> This isn't too hard to fix, but definitely needs fixing.
>
> Thanks, nice catch!
>
> From what I have seen, we are not going through InitXLOGAccess() on a
> Standby and in some cases (like the one you mentioned)
> StartLogicalReplication() is called without IdentifySystem() being
> called previously: this lead to ThisTimeLineID still set to 0.
>
> I am proposing a fix in the attached v18 by adding a check in
> StartLogicalReplication() and ensuring that ThisTimeLineID is retrieved.
>
>>
>> - ResolveRecoveryConflictWithLogicalSlots() is racy - potentially
>> leading us to drop a slot that has been created since we signalled a
>> recovery conflict. See
>> https://www.postgresql.org/message-id/20210408020913.zzprrlvqyvlt5cyy%40alap3.anarazel.de
>> for some very similar issues.
>
> I have rewritten this part by following the same logic as the one used
> in 96540f80f8 (the commit linked to the thread you mentioned).
>
>>
>> - Given the precedent of max_slot_wal_keep_size, I think it's wrong to
>> just drop the logical slots. Instead we should just mark them as
>> invalid, like InvalidateObsoleteReplicationSlots().
>
> Makes fully sense and done that way in the attached patch.
>
> I am setting the slot's data.xmin and data.catalog_xmin as
> InvalidTransactionId to mark the slot(s) as invalid in case of conflict.
>
>> - There's no tests covering timeline switches, what happens if there's a
>> promotion if logical decoding is currently ongoing.
>
> I'll now work on the tests.
>
>>
>> - The way ResolveRecoveryConflictWithLogicalSlots() builds the error
>> message is not good (and I've complained about it before...).
>
> I changed it and made it more simple.
>
> I also removed the details around mentioning xmin or catalog xmin (as
> I am not sure of the added value and they are currently also not
> mentioned during standby recovery snapshot conflict).
>
>>
>> Unfortunately I think the things I have found are too many for me to
>> address within the given time. I'll send a version with a somewhat
>> polished set of the changes I made in the next few days...
>
> Thanks for the review and feedback.
>
> Please find enclosed v18 with the changes I worked on.
>
> I still need to have a look on the tests.
Please find enclosed v19 that also contains the changes related to your
TAP tests remarks, mainly:
- get rid of 024 and add more tests in 026 (025 has been used in the
meantime)
- test that logical decoding actually produces useful and correct results
- test standby promotion and logical decoding behavior once done
- useless "use" removal
- check_confl_logicalslot() function removal
- rewrote make_slot_active() to make use of poll_query_until() and timeout
- remove the useless eval()
- remove the "Catalog xmins should advance after standby logical slot
fetches the changes" test
One thing that's not clear to me is your remark "There's also no test
for a recovery conflict due to row removal": Don't you think that the
"vacuum full" conflict test is enough? if not, what kind of additional
tests would you like to see?
>
> There is also the 10s delay to work on, do you already have an idea on
> how we should handle it?
>
> Thanks
>
> Bertrand
>
Thanks
Bertrand
From c0eae773a83916aa8dd652a67d8ef542fec327b7 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:20:45 +0000
Subject: [PATCH v19 5/5] Doc changes describing details about logical
decoding.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
doc/src/sgml/logicaldecoding.sgml | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
100.0% doc/src/sgml/
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index 1765ea6c87..de67bdde8b 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -297,6 +297,27 @@ postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NU
may consume changes from a slot at any given time.
</para>
+ <para>
+ A logical replication slot can also be created on a hot standby. To prevent
+ <command>VACUUM</command> from removing required rows from the system
+ catalogs, <varname>hot_standby_feedback</varname> should be set on the
+ standby. In spite of that, if any required rows get removed, the slot gets
+ invalidated. It's highly recommended to use a physical slot between the primary
+ and the standby. Otherwise, hot_standby_feedback will work, but only while the
+ connection is alive (for example a node restart would break it). Existing
+ logical slots on standby also get invalidated if wal_level on primary is reduced to
+ less than 'logical'.
+ </para>
+
+ <para>
+ For a logical slot to be created, it builds a historic snapshot, for which
+ information of all the currently running transactions is essential. On
+ primary, this information is available, but on standby, this information
+ has to be obtained from primary. So, slot creation may wait for some
+ activity to happen on the primary. If the primary is idle, creating a
+ logical slot on standby may take a noticeable time.
+ </para>
+
<caution>
<para>
Replication slots persist across crashes and know nothing about the state
--
2.18.4
From 1e54c6ed8d75fd82cc4752f9fa078cdc54cd1f41 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:19:13 +0000
Subject: [PATCH v19 4/5] New TAP test for logical decoding on standby.
Author: Craig Ringer (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/test/perl/PostgresNode.pm | 37 ++
.../t/026_standby_logical_decoding.pl | 426 ++++++++++++++++++
2 files changed, 463 insertions(+)
7.2% src/test/perl/
92.7% src/test/recovery/t/
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 2027cbf43d..4984b2f6f4 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -2612,6 +2612,43 @@ sub pg_recvlogical_upto
=pod
+=item $node->create_logical_slot_on_standby(self, master, slot_name, dbname)
+
+Create logical replication slot on given standby
+
+=cut
+
+sub create_logical_slot_on_standby
+{
+ my ($self, $master, $slot_name, $dbname) = @_;
+ my ($stdout, $stderr);
+
+ my $handle;
+
+ $handle = IPC::Run::start(['pg_recvlogical', '-d', $self->connstr($dbname), '-P', 'test_decoding', '-S', $slot_name, '--create-slot'], '>', \$stdout, '2>', \$stderr);
+
+ # Once slot restart_lsn is created, the standby looks for xl_running_xacts
+ # WAL record from the restart_lsn onwards. So firstly, wait until the slot
+ # restart_lsn is evaluated.
+
+ $self->poll_query_until(
+ 'postgres', qq[
+ SELECT restart_lsn IS NOT NULL
+ FROM pg_catalog.pg_replication_slots WHERE slot_name = '$slot_name'
+ ]) or die "timed out waiting for logical slot to calculate its restart_lsn";
+
+ # Now arrange for the xl_running_xacts record for which pg_recvlogical
+ # is waiting.
+ $master->safe_psql('postgres', 'CHECKPOINT');
+
+ $handle->finish();
+
+ is($self->slot($slot_name)->{'slot_type'}, 'logical', $slot_name . ' on standby created')
+ or die "could not create slot" . $slot_name;
+}
+
+=pod
+
=back
=cut
diff --git a/src/test/recovery/t/026_standby_logical_decoding.pl b/src/test/recovery/t/026_standby_logical_decoding.pl
new file mode 100644
index 0000000000..1687c63932
--- /dev/null
+++ b/src/test/recovery/t/026_standby_logical_decoding.pl
@@ -0,0 +1,426 @@
+# logical decoding on standby : test logical decoding,
+# recovery conflict and standby promotion.
+
+use strict;
+use warnings;
+
+use PostgresNode;
+use Test::More tests => 32;
+use Time::HiRes qw(usleep);
+
+my ($stdin, $stdout, $stderr, $ret, $handle, $slot);
+
+my $node_primary = get_new_node('primary');
+my $node_standby = get_new_node('standby');
+
+# Name for the physical slot on primary
+my $primary_slotname = 'primary_physical';
+
+# return the size of logfile of $node in bytes
+sub get_log_size
+{
+ my ($node) = @_;
+
+ return (stat $node->logfile)[7];
+}
+
+# find $pat in logfile of $node after $off-th byte
+sub find_in_log
+{
+ my ($node, $pat, $off) = @_;
+
+ $off = 0 unless defined $off;
+ my $log = TestLib::slurp_file($node->logfile);
+ return 0 if (length($log) <= $off);
+
+ $log = substr($log, $off);
+
+ return $log =~ m/$pat/;
+}
+
+# Fetch xmin columns from slot's pg_replication_slots row, after waiting for
+# given boolean condition to be true to ensure we've reached a quiescent state.
+sub wait_for_xmins
+{
+ my ($node, $slotname, $check_expr) = @_;
+
+ $node->poll_query_until(
+ 'postgres', qq[
+ SELECT $check_expr
+ FROM pg_catalog.pg_replication_slots
+ WHERE slot_name = '$slotname';
+ ]) or die "Timed out waiting for slot xmins to advance";
+}
+
+# Create the required logical slots on standby.
+sub create_logical_slots
+{
+ $node_standby->create_logical_slot_on_standby($node_primary, 'inactiveslot', 'testdb');
+ $node_standby->create_logical_slot_on_standby($node_primary, 'activeslot', 'testdb');
+}
+
+# Acquire one of the standby logical slots created by create_logical_slots().
+# In case wait is true we are waiting for an active pid on the 'activeslot' slot.
+# If wait is not true it means we are testing a known failure scenario.
+sub make_slot_active
+{
+ my $wait = shift;
+ my $slot_user_handle;
+
+ print "starting pg_recvlogical\n";
+ $slot_user_handle = IPC::Run::start(['pg_recvlogical', '-d', $node_standby->connstr('testdb'), '-S', 'activeslot', '-f', '-', '--no-loop', '--start'], '>', \$stdout, '2>', \$stderr);
+
+ if ($wait)
+ # make sure activeslot is in use
+ {
+ $node_standby->poll_query_until('testdb',
+ "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'activeslot' AND active_pid IS NOT NULL)"
+ ) or die "slot never became active";
+ }
+
+ return $slot_user_handle;
+}
+
+# Check if all the slots on standby are dropped. These include the 'activeslot'
+# that was acquired by make_slot_active(), and the non-active 'inactiveslot'.
+sub check_slots_dropped
+{
+ my ($slot_user_handle) = @_;
+ my $return;
+
+ is($node_standby->slot('inactiveslot')->{'slot_type'}, '', 'inactiveslot on standby dropped');
+ is($node_standby->slot('activeslot')->{'slot_type'}, '', 'activeslot on standby dropped');
+
+ # our client should've terminated in response to the walsender error
+ $slot_user_handle->finish;
+ $return = $?;
+ cmp_ok($return, "!=", 0, "pg_recvlogical exited non-zero");
+ if ($return) {
+ like($stderr, qr/conflict with recovery/, 'slot have been dropped');
+ }
+
+ return 0;
+}
+
+
+########################
+# Initialize primary node
+########################
+
+$node_primary->init(allows_streaming => 1, has_archiving => 1);
+$node_primary->append_conf('postgresql.conf', q{
+wal_level = 'logical'
+max_replication_slots = 4
+max_wal_senders = 4
+log_min_messages = 'debug2'
+log_error_verbosity = verbose
+});
+$node_primary->dump_info;
+$node_primary->start;
+
+$node_primary->psql('postgres', q[CREATE DATABASE testdb]);
+
+$node_primary->safe_psql('testdb', qq[SELECT * FROM pg_create_physical_replication_slot('$primary_slotname');]);
+my $backup_name = 'b1';
+$node_primary->backup($backup_name);
+
+#######################
+# Initialize standby node
+#######################
+
+$node_standby->init_from_backup(
+ $node_primary, $backup_name,
+ has_streaming => 1,
+ has_restoring => 1);
+$node_standby->append_conf('postgresql.conf',
+ qq[primary_slot_name = '$primary_slotname']);
+$node_standby->start;
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+
+##################################################
+# Test that logical decoding on the standby
+# behaves correctly.
+##################################################
+
+create_logical_slots();
+
+$node_primary->safe_psql('testdb', qq[CREATE TABLE decoding_test(x integer, y text);]);
+$node_primary->safe_psql('testdb', qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,10) s;]);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $result = $node_standby->safe_psql('testdb',
+ qq[SELECT pg_logical_slot_get_changes('activeslot', NULL, NULL);]);
+
+# test if basic decoding works
+is(scalar(my @foobar = split /^/m, $result),
+ 14, 'Decoding produced 14 rows');
+
+# Insert some rows and verify that we get the same results from pg_recvlogical
+# and the SQL interface.
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
+);
+
+my $expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT};
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $stdout_sql = $node_standby->safe_psql('testdb',
+ qq[SELECT data FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
+);
+
+is($stdout_sql, $expected, 'got expected output from SQL decoding session');
+
+my $endpos = $node_standby->safe_psql('testdb',
+ "SELECT lsn FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
+);
+print "waiting to replay $endpos\n";
+
+# Insert some rows after $endpos, which we won't read.
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,50) s;]
+);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $stdout_recv = $node_standby->pg_recvlogical_upto(
+ 'testdb', 'activeslot', $endpos, 180,
+ 'include-xids' => '0',
+ 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, $expected,
+ 'got same expected output from pg_recvlogical decoding session');
+
+$node_standby->poll_query_until('testdb',
+ "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'activeslot' AND active_pid IS NULL)"
+) or die "slot never became inactive";
+
+$stdout_recv = $node_standby->pg_recvlogical_upto(
+ 'testdb', 'activeslot', $endpos, 180,
+ 'include-xids' => '0',
+ 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, '', 'pg_recvlogical acknowledged changes');
+
+$node_primary->safe_psql('postgres', 'CREATE DATABASE otherdb');
+
+is( $node_primary->psql(
+ 'otherdb',
+ "SELECT lsn FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
+ ),
+ 3,
+ 'replaying logical slot from another database fails');
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 1 : hot_standby_feedback off
+##################################################
+
+create_logical_slots();
+
+# One way to reproduce recovery conflict is to run VACUUM FULL with
+# hot_standby_feedback turned off on the standby.
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = off
+]);
+$node_standby->restart;
+# ensure walreceiver feedback off by waiting for expected xmin and
+# catalog_xmin on primary. Both should be NULL since hs_feedback is off
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NULL AND catalog_xmin IS NULL");
+
+$handle = make_slot_active(1);
+
+# This should trigger the conflict
+$node_primary->safe_psql('testdb', 'VACUUM FULL');
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery"),
+ 'inactiveslot slot invalidation is logged');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery"),
+ 'activeslot slot invalidation is logged');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# We are not able to read from the slot as it has been invalidated
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\""),
+ 'cannot read from logical replication slot');
+
+# Turn hot_standby_feedback back on
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = on
+]);
+$node_standby->restart;
+
+# ensure walreceiver feedback sent by waiting for expected xmin and
+# catalog_xmin on primary. With hot_standby_feedback on, xmin should advance,
+# but catalog_xmin should still remain NULL since there is no logical slot.
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NOT NULL AND catalog_xmin IS NULL");
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 2 : incorrect wal_level on primary.
+##################################################
+
+# get the position to search from in the standby logfile
+my $logstart = get_log_size($node_standby);
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+create_logical_slots();
+
+$handle = make_slot_active(1);
+
+# Make primary wal_level replica. This will trigger slot conflict.
+$node_primary->append_conf('postgresql.conf',q[
+wal_level = 'replica'
+]);
+$node_primary->restart;
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery", $logstart),
+ 'inactiveslot slot invalidation is logged');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery", $logstart),
+ 'activeslot slot invalidation is logged');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+ok( find_in_log(
+ $node_standby,
+ "logical decoding on standby requires wal_level >= logical on master", $logstart),
+ 'cannot start replication because wal_level < logical on master');
+
+# Restore primary wal_level
+$node_primary->append_conf('postgresql.conf',q[
+wal_level = 'logical'
+]);
+$node_primary->restart;
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# as the slot has been invalidated we should not be able to read
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\"", $logstart),
+ 'cannot read from logical replication slot');
+
+##################################################
+# DROP DATABASE should drops it's slots, including active slots.
+##################################################
+
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+create_logical_slots();
+$handle = make_slot_active(1);
+# Create a slot on a database that would not be dropped. This slot should not
+# get dropped.
+$node_standby->create_logical_slot_on_standby($node_primary, 'otherslot', 'postgres');
+
+# dropdb on the primary to verify slots are dropped on standby
+$node_primary->safe_psql('postgres', q[DROP DATABASE testdb]);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+is($node_standby->safe_psql('postgres',
+ q[SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = 'testdb')]), 'f',
+ 'database dropped on standby');
+
+check_slots_dropped($handle);
+
+is($node_standby->slot('otherslot')->{'slot_type'}, 'logical',
+ 'otherslot on standby not dropped');
+
+# Cleanup : manually drop the slot that was not dropped.
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('otherslot')]);
+
+##################################################
+# Test standby promotion and logical decoding bheavior
+# after the standby gets promoted.
+##################################################
+
+$node_primary->psql('postgres', q[CREATE DATABASE testdb]);
+$node_primary->safe_psql('testdb', qq[CREATE TABLE decoding_test(x integer, y text);]);
+
+# create the logical slots
+create_logical_slots();
+
+# Insert some rows before the promotion
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
+);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# promote
+$node_standby->promote;
+
+# insert some rows on promoted standby
+$node_standby->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,7) s;]
+);
+
+
+$expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT
+BEGIN
+table public.decoding_test: INSERT: x[integer]:5 y[text]:'5'
+table public.decoding_test: INSERT: x[integer]:6 y[text]:'6'
+table public.decoding_test: INSERT: x[integer]:7 y[text]:'7'
+COMMIT};
+
+# check that we are decoding pre and post promotion inserted rows
+$stdout_sql = $node_standby->safe_psql('testdb',
+ qq[SELECT data FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
+);
+
+is($stdout_sql, $expected, 'got expected output from SQL decoding session on promoted standby');
--
2.18.4
From b71ae587dcf987f231c823c4c2fd88f6229eef61 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:13:34 +0000
Subject: [PATCH v19 3/5] Allow logical decoding on standby.
Allow a logical slot to be created on standby. Restrict its usage
or its creation if wal_level on primary is less than logical.
During slot creation, it's restart_lsn is set to the last replayed
LSN. Effectively, a logical slot creation on standby waits for an
xl_running_xact record to arrive from primary. Conflicting slots
would be handled in next commits.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/backend/access/transam/xlog.c | 33 ++++++++++-
src/backend/access/transam/xlogfuncs.c | 2 +-
src/backend/access/transam/xlogutils.c | 2 +-
src/backend/postmaster/checkpointer.c | 4 +-
src/backend/replication/logical/decode.c | 22 +++++++-
src/backend/replication/logical/logical.c | 37 ++++++------
.../replication/logical/logicalfuncs.c | 2 +-
src/backend/replication/slot.c | 56 ++++++++++---------
src/backend/replication/slotfuncs.c | 2 +-
src/backend/replication/walreceiver.c | 4 +-
src/backend/replication/walreceiverfuncs.c | 2 +-
src/backend/replication/walsender.c | 26 ++++++---
src/include/access/xlog.h | 3 +-
13 files changed, 132 insertions(+), 63 deletions(-)
16.7% src/backend/access/transam/
32.7% src/backend/replication/logical/
45.2% src/backend/replication/
5.3% src/
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index e863caf3b2..48b43c9fd2 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -5078,6 +5078,17 @@ LocalProcessControlFile(bool reset)
ReadControlFile();
}
+/*
+ * Get the wal_level from the control file. For a standby, this value should be
+ * considered as its active wal_level, because it may be different from what
+ * was originally configured on standby.
+ */
+WalLevel
+GetActiveWalLevelOnStandby(void)
+{
+ return ControlFile->wal_level;
+}
+
/*
* Initialization of shared memory for XLOG
*/
@@ -9599,7 +9610,7 @@ CreateRestartPoint(int flags)
* whichever is later.
*/
receivePtr = GetWalRcvFlushRecPtr(NULL, NULL);
- replayPtr = GetXLogReplayRecPtr(&replayTLI);
+ replayPtr = GetXLogReplayRecPtr(&replayTLI, false);
endptr = (receivePtr < replayPtr) ? replayPtr : receivePtr;
KeepLogSeg(endptr, &_logSegNo);
InvalidateObsoleteReplicationSlots(_logSegNo);
@@ -11711,7 +11722,7 @@ register_persistent_abort_backup_handler(void)
* Exported to allow WALReceiver to read the pointer directly.
*/
XLogRecPtr
-GetXLogReplayRecPtr(TimeLineID *replayTLI)
+GetXLogReplayRecPtr(TimeLineID *replayTLI, bool avoid_header)
{
XLogRecPtr recptr;
TimeLineID tli;
@@ -11723,6 +11734,24 @@ GetXLogReplayRecPtr(TimeLineID *replayTLI)
if (replayTLI)
*replayTLI = tli;
+
+ /*
+ * Replay pointer may point one past the end of the record. If that
+ * is a XLOG page boundary, it will not be a valid LSN for the
+ * start of a record, so bump it up past the page header.
+ */
+ if (avoid_header && !XRecOffIsValid(recptr))
+ {
+ if (recptr % XLOG_BLCKSZ != 0)
+ elog(ERROR, "invalid replay pointer");
+
+ /* For the first page of a segment file, it's a long header */
+ if (XLogSegmentOffset(recptr, wal_segment_size) == 0)
+ recptr += SizeOfXLogLongPHD;
+ else
+ recptr += SizeOfXLogShortPHD;
+ }
+
return recptr;
}
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index b98deb72ec..a173f8d6fc 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -417,7 +417,7 @@ pg_last_wal_replay_lsn(PG_FUNCTION_ARGS)
{
XLogRecPtr recptr;
- recptr = GetXLogReplayRecPtr(NULL);
+ recptr = GetXLogReplayRecPtr(NULL, false);
if (recptr == 0)
PG_RETURN_NULL();
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index d17d660f46..81088baae3 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -850,7 +850,7 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
if (!RecoveryInProgress())
read_upto = GetFlushRecPtr();
else
- read_upto = GetXLogReplayRecPtr(&ThisTimeLineID);
+ read_upto = GetXLogReplayRecPtr(&ThisTimeLineID, false);
tli = ThisTimeLineID;
/*
diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c
index 75a95f3de7..11d8f1370c 100644
--- a/src/backend/postmaster/checkpointer.c
+++ b/src/backend/postmaster/checkpointer.c
@@ -437,7 +437,7 @@ CheckpointerMain(void)
*/
ckpt_active = true;
if (do_restartpoint)
- ckpt_start_recptr = GetXLogReplayRecPtr(NULL);
+ ckpt_start_recptr = GetXLogReplayRecPtr(NULL, false);
else
ckpt_start_recptr = GetInsertRecPtr();
ckpt_start_time = now;
@@ -801,7 +801,7 @@ IsCheckpointOnSchedule(double progress)
* value that was in effect when the WAL was generated).
*/
if (RecoveryInProgress())
- recptr = GetXLogReplayRecPtr(NULL);
+ recptr = GetXLogReplayRecPtr(NULL, false);
else
recptr = GetInsertRecPtr();
elapsed_xlogs = (((double) (recptr - ckpt_start_recptr)) /
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 453efc51e1..6684173a61 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -215,11 +215,31 @@ DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
* can restart from there.
*/
break;
+ case XLOG_PARAMETER_CHANGE:
+ {
+ xl_parameter_change *xlrec =
+ (xl_parameter_change *) XLogRecGetData(buf->record);
+
+ /*
+ * If wal_level on primary is reduced to less than logical, then we
+ * want to prevent existing logical slots from being used.
+ * Existing logical slots on standby get invalidated when this WAL
+ * record is replayed; and further, slot creation fails when the
+ * wal level is not sufficient; but all these operations are not
+ * synchronized, so a logical slot may creep in while the wal_level
+ * is being reduced. Hence this extra check.
+ */
+ if (xlrec->wal_level < WAL_LEVEL_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires "
+ "wal_level >= logical on master")));
+ break;
+ }
case XLOG_NOOP:
case XLOG_NEXTOID:
case XLOG_SWITCH:
case XLOG_BACKUP_END:
- case XLOG_PARAMETER_CHANGE:
case XLOG_RESTORE_POINT:
case XLOG_FPW_CHANGE:
case XLOG_FPI_FOR_HINT:
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index ffc6160e9f..3c46938972 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -119,23 +119,22 @@ CheckLogicalDecodingRequirements(void)
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("logical decoding requires a database connection")));
- /* ----
- * TODO: We got to change that someday soon...
- *
- * There's basically three things missing to allow this:
- * 1) We need to be able to correctly and quickly identify the timeline a
- * LSN belongs to
- * 2) We need to force hot_standby_feedback to be enabled at all times so
- * the primary cannot remove rows we need.
- * 3) support dropping replication slots referring to a database, in
- * dbase_redo. There can't be any active ones due to HS recovery
- * conflicts, so that should be relatively easy.
- * ----
- */
if (RecoveryInProgress())
- ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("logical decoding cannot be used while in recovery")));
+ {
+ /*
+ * This check may have race conditions, but whenever
+ * XLOG_PARAMETER_CHANGE indicates that wal_level has changed, we
+ * verify that there are no existing logical replication slots. And to
+ * avoid races around creating a new slot,
+ * CheckLogicalDecodingRequirements() is called once before creating
+ * the slot, and once when logical decoding is initially starting up.
+ */
+ if (GetActiveWalLevelOnStandby() < WAL_LEVEL_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires "
+ "wal_level >= logical on master")));
+ }
}
/*
@@ -330,6 +329,12 @@ CreateInitDecodingContext(const char *plugin,
LogicalDecodingContext *ctx;
MemoryContext old_context;
+ /*
+ * On standby, this check is also required while creating the slot. Check
+ * the comments in this function.
+ */
+ CheckLogicalDecodingRequirements();
+
/* shorter lines... */
slot = MyReplicationSlot;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 1d9400ea63..9069f3e50d 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -223,7 +223,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
if (!RecoveryInProgress())
end_of_wal = GetFlushRecPtr();
else
- end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID);
+ end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID, false);
ReplicationSlotAcquire(NameStr(*name), true);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index f9231229cc..4302d56539 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1087,37 +1087,28 @@ ReplicationSlotReserveWal(void)
/*
* For logical slots log a standby snapshot and start logical decoding
* at exactly that position. That allows the slot to start up more
- * quickly.
+ * quickly. But on a standby we cannot do WAL writes, so just use the
+ * replay pointer; effectively, an attempt to create a logical slot on
+ * standby will cause it to wait for an xl_running_xact record to be
+ * logged independently on the primary, so that a snapshot can be built
+ * using the record.
*
- * That's not needed (or indeed helpful) for physical slots as they'll
- * start replay at the last logged checkpoint anyway. Instead return
- * the location of the last redo LSN. While that slightly increases
- * the chance that we have to retry, it's where a base backup has to
- * start replay at.
+ * None of this is needed (or indeed helpful) for physical slots as
+ * they'll start replay at the last logged checkpoint anyway. Instead
+ * return the location of the last redo LSN. While that slightly
+ * increases the chance that we have to retry, it's where a base backup
+ * has to start replay at.
*/
- if (!RecoveryInProgress() && SlotIsLogical(slot))
- {
- XLogRecPtr flushptr;
-
- /* start at current insert position */
+ if (SlotIsPhysical(slot))
+ restart_lsn = GetRedoRecPtr();
+ else if (RecoveryInProgress())
+ restart_lsn = GetXLogReplayRecPtr(NULL, true);
+ else
restart_lsn = GetXLogInsertRecPtr();
- SpinLockAcquire(&slot->mutex);
- slot->data.restart_lsn = restart_lsn;
- SpinLockRelease(&slot->mutex);
-
- /* make sure we have enough information to start */
- flushptr = LogStandbySnapshot();
- /* and make sure it's fsynced to disk */
- XLogFlush(flushptr);
- }
- else
- {
- restart_lsn = GetRedoRecPtr();
- SpinLockAcquire(&slot->mutex);
- slot->data.restart_lsn = restart_lsn;
- SpinLockRelease(&slot->mutex);
- }
+ SpinLockAcquire(&slot->mutex);
+ slot->data.restart_lsn = restart_lsn;
+ SpinLockRelease(&slot->mutex);
/* prevent WAL removal as fast as possible */
ReplicationSlotsComputeRequiredLSN();
@@ -1133,6 +1124,17 @@ ReplicationSlotReserveWal(void)
if (XLogGetLastRemovedSegno() < segno)
break;
}
+
+ if (!RecoveryInProgress() && SlotIsLogical(slot))
+ {
+ XLogRecPtr flushptr;
+
+ /* make sure we have enough information to start */
+ flushptr = LogStandbySnapshot();
+
+ /* and make sure it's fsynced to disk */
+ XLogFlush(flushptr);
+ }
}
/*
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 31e74d3832..48d24442e2 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -636,7 +636,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
if (!RecoveryInProgress())
moveto = Min(moveto, GetFlushRecPtr());
else
- moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID));
+ moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID, false));
/* Acquire the slot so we "own" it */
ReplicationSlotAcquire(NameStr(*slotname), true);
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index faeea9f0cc..4bce8f192c 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -407,7 +407,7 @@ WalReceiverMain(void)
first_stream = false;
/* Initialize LogstreamResult and buffers for processing messages */
- LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
+ LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL, false);
initStringInfo(&reply_message);
initStringInfo(&incoming_message);
@@ -1075,7 +1075,7 @@ XLogWalRcvSendReply(bool force, bool requestReply)
/* Construct a new message */
writePtr = LogstreamResult.Write;
flushPtr = LogstreamResult.Flush;
- applyPtr = GetXLogReplayRecPtr(NULL);
+ applyPtr = GetXLogReplayRecPtr(NULL, false);
resetStringInfo(&reply_message);
pq_sendbyte(&reply_message, 'r');
diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c
index 6f0acbfdef..f264b71f73 100644
--- a/src/backend/replication/walreceiverfuncs.c
+++ b/src/backend/replication/walreceiverfuncs.c
@@ -372,7 +372,7 @@ GetReplicationApplyDelay(void)
receivePtr = walrcv->flushedUpto;
SpinLockRelease(&walrcv->mutex);
- replayPtr = GetXLogReplayRecPtr(NULL);
+ replayPtr = GetXLogReplayRecPtr(NULL, false);
if (receivePtr == replayPtr)
return 0;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 53e440c2ae..bc316cd28d 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1166,6 +1166,16 @@ StartLogicalReplication(StartReplicationCmd *cmd)
got_STOPPING = true;
}
+ /*
+ * In case of logical decoding on standby it may be that ThisTimeLineID
+ * is not set yet.
+ * Indeed we are not going through InitXLOGAccess on a Standby and
+ * it may also be that IdentifySystem has not been called yet.
+ * So let's get it through GetXLogReplayRecPtr().
+ */
+ if (ThisTimeLineID == 0)
+ (void) GetXLogReplayRecPtr(&ThisTimeLineID, false);
+
/*
* Create our decoding context, making it start at the previously ack'ed
* position.
@@ -1390,7 +1400,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr();
else
- RecentFlushPtr = GetXLogReplayRecPtr(NULL);
+ RecentFlushPtr = GetXLogReplayRecPtr(NULL, false);
for (;;)
{
@@ -1424,7 +1434,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr();
else
- RecentFlushPtr = GetXLogReplayRecPtr(NULL);
+ RecentFlushPtr = GetXLogReplayRecPtr(NULL, false);
/*
* If postmaster asked us to stop, don't wait anymore.
@@ -2890,10 +2900,12 @@ XLogSendLogical(void)
* If first time through in this session, initialize flushPtr. Otherwise,
* we only need to update flushPtr if EndRecPtr is past it.
*/
- if (flushPtr == InvalidXLogRecPtr)
- flushPtr = GetFlushRecPtr();
- else if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
- flushPtr = GetFlushRecPtr();
+ if (flushPtr == InvalidXLogRecPtr ||
+ logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
+ {
+ flushPtr = (am_cascading_walsender ?
+ GetStandbyFlushRecPtr() : GetFlushRecPtr());
+ }
/* If EndRecPtr is still past our flushPtr, it means we caught up. */
if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
@@ -2982,7 +2994,7 @@ GetStandbyFlushRecPtr(void)
*/
receivePtr = GetWalRcvFlushRecPtr(NULL, &receiveTLI);
- replayPtr = GetXLogReplayRecPtr(&replayTLI);
+ replayPtr = GetXLogReplayRecPtr(&replayTLI, false);
ThisTimeLineID = replayTLI;
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 77187c12be..c7fb52240e 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -316,7 +316,7 @@ extern bool HotStandbyActive(void);
extern bool HotStandbyActiveInReplay(void);
extern bool XLogInsertAllowed(void);
extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
-extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI);
+extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI, bool avoid_header);
extern XLogRecPtr GetXLogInsertRecPtr(void);
extern XLogRecPtr GetXLogWriteRecPtr(void);
extern RecoveryPauseState GetRecoveryPauseState(void);
@@ -333,6 +333,7 @@ extern Size XLOGShmemSize(void);
extern void XLOGShmemInit(void);
extern void BootStrapXLOG(void);
extern void LocalProcessControlFile(bool reset);
+extern WalLevel GetActiveWalLevelOnStandby(void);
extern void StartupXLOG(void);
extern void ShutdownXLOG(int code, Datum arg);
extern void InitXLOGAccess(void);
--
2.18.4
From 6142c34b7a30f468aac56f24a41e3275dae0540e Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:12:16 +0000
Subject: [PATCH v19 2/5] Handle logical slot conflicts on standby.
During WAL replay on standby, when slot conflict is identified,
invalidate such slots. Also do the same thing if wal_level on master
is reduced to below logical and there are existing logical slots
on standby. Introduce a new ProcSignalReason value for slot
conflict recovery. Arrange for a new pg_stat_get_activity field:
confl_logicalslot.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
doc/src/sgml/monitoring.sgml | 10 +
src/backend/access/gist/gistxlog.c | 4 +-
src/backend/access/hash/hash_xlog.c | 3 +-
src/backend/access/heap/heapam.c | 10 +-
src/backend/access/nbtree/nbtxlog.c | 4 +-
src/backend/access/spgist/spgxlog.c | 1 +
src/backend/access/transam/xlog.c | 13 ++
src/backend/catalog/system_views.sql | 3 +-
src/backend/postmaster/pgstat.c | 20 ++
.../replication/logical/logicalfuncs.c | 7 +-
src/backend/replication/slot.c | 209 ++++++++++++++++++
src/backend/replication/walsender.c | 8 +
src/backend/storage/ipc/procarray.c | 4 +
src/backend/storage/ipc/procsignal.c | 3 +
src/backend/storage/ipc/standby.c | 11 +-
src/backend/tcop/postgres.c | 22 ++
src/backend/utils/adt/pgstatfuncs.c | 16 ++
src/include/catalog/pg_proc.dat | 5 +
src/include/pgstat.h | 2 +
src/include/replication/slot.h | 3 +
src/include/storage/procsignal.h | 1 +
src/include/storage/standby.h | 4 +-
src/test/regress/expected/rules.out | 3 +-
23 files changed, 351 insertions(+), 15 deletions(-)
4.0% src/backend/access/heap/
4.7% src/backend/access/transam/
5.1% src/backend/access/
4.3% src/backend/postmaster/
3.5% src/backend/replication/logical/
52.3% src/backend/replication/
5.6% src/backend/storage/ipc/
7.1% src/backend/tcop/
3.5% src/backend/
6.3% src/include/
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index dcbb10fb6f..e9e1efa1b8 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -4029,6 +4029,16 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
deadlocks
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>confl_logicalslot</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of queries in this database that have been canceled due to
+ logical slots
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 46aee6f2a9..5963e639d8 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -195,7 +195,8 @@ gistRedoDeleteRecord(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid,
+ xldata->onCatalogTable, rnode);
}
if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
@@ -395,6 +396,7 @@ gistRedoPageReuse(XLogReaderState *record)
*/
if (InHotStandby)
ResolveRecoveryConflictWithSnapshotFullXid(xlrec->latestRemovedFullXid,
+ xlrec->onCatalogTable,
xlrec->node);
}
diff --git a/src/backend/access/hash/hash_xlog.c b/src/backend/access/hash/hash_xlog.c
index af35a991fc..f91d4c36e2 100644
--- a/src/backend/access/hash/hash_xlog.c
+++ b/src/backend/access/hash/hash_xlog.c
@@ -1002,7 +1002,8 @@ hash_xlog_vacuum_one_page(XLogReaderState *record)
RelFileNode rnode;
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid,
+ xldata->onCatalogTable, rnode);
}
action = XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &buffer);
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 4019a2122e..75ca5f79f4 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -8419,7 +8419,8 @@ heap_xlog_prune(XLogReaderState *record)
* no queries running for which the removed tuples are still visible.
*/
if (InHotStandby)
- ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
/*
* If we have a full-page image, restore it (using a cleanup lock) and
@@ -8587,7 +8588,9 @@ heap_xlog_visible(XLogReaderState *record)
* rather than killing the transaction outright.
*/
if (InHotStandby)
- ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid,
+ xlrec->onCatalogTable,
+ rnode);
/*
* Read the heap page, if it still exists. If the heap file has dropped or
@@ -8724,7 +8727,8 @@ heap_xlog_freeze_page(XLogReaderState *record)
TransactionIdRetreat(latestRemovedXid);
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
}
if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
diff --git a/src/backend/access/nbtree/nbtxlog.c b/src/backend/access/nbtree/nbtxlog.c
index c2e920f159..004e8b7dd8 100644
--- a/src/backend/access/nbtree/nbtxlog.c
+++ b/src/backend/access/nbtree/nbtxlog.c
@@ -669,7 +669,8 @@ btree_xlog_delete(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
}
/*
@@ -1007,6 +1008,7 @@ btree_xlog_reuse_page(XLogReaderState *record)
if (InHotStandby)
ResolveRecoveryConflictWithSnapshotFullXid(xlrec->latestRemovedFullXid,
+ xlrec->onCatalogTable,
xlrec->node);
}
diff --git a/src/backend/access/spgist/spgxlog.c b/src/backend/access/spgist/spgxlog.c
index 3dfd2aa317..add4da4e74 100644
--- a/src/backend/access/spgist/spgxlog.c
+++ b/src/backend/access/spgist/spgxlog.c
@@ -881,6 +881,7 @@ spgRedoVacuumRedirect(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &node, NULL, NULL);
ResolveRecoveryConflictWithSnapshot(xldata->newestRedirectXid,
+ xldata->onCatalogTable,
node);
}
}
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 1b3a3d9bea..e863caf3b2 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -10358,6 +10358,19 @@ xlog_redo(XLogReaderState *record)
/* Update our copy of the parameters in pg_control */
memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_parameter_change));
+ /*
+ * Invalidate logical slots if we are in hot standby and the primary does not
+ * have a WAL level sufficient for logical decoding. No need to search
+ * for potentially conflicting logically slots if standby is running
+ * with wal_level lower than logical, because in that case, we would
+ * have either disallowed creation of logical slots or invalidated existing
+ * ones.
+ */
+ if (InRecovery && InHotStandby &&
+ xlrec.wal_level < WAL_LEVEL_LOGICAL &&
+ wal_level >= WAL_LEVEL_LOGICAL)
+ InvalidateConflictingLogicalReplicationSlots(InvalidOid,InvalidTransactionId);
+
LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
ControlFile->MaxConnections = xlrec.MaxConnections;
ControlFile->max_worker_processes = xlrec.max_worker_processes;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 999d984068..e4b7830a1e 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1020,7 +1020,8 @@ CREATE VIEW pg_stat_database_conflicts AS
pg_stat_get_db_conflict_lock(D.oid) AS confl_lock,
pg_stat_get_db_conflict_snapshot(D.oid) AS confl_snapshot,
pg_stat_get_db_conflict_bufferpin(D.oid) AS confl_bufferpin,
- pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock
+ pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock,
+ pg_stat_get_db_conflict_logicalslot(D.oid) AS confl_logicalslot
FROM pg_database D;
CREATE VIEW pg_stat_user_functions AS
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index b0d07c0e0b..38fa5f4809 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -1863,6 +1863,22 @@ pgstat_report_replslot_drop(const char *slotname)
pgstat_send(&msg, sizeof(PgStat_MsgReplSlot));
}
+/* ----------
+ * pgstat_report_replslot_conflict()
+ * Tell the collector about a logical slot being conflicting
+ * with recovery.
+ * ----------
+ */
+void
+pgstat_report_replslot_conflict(Oid dboid)
+{
+ PgStat_MsgRecoveryConflict msg;
+ pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RECOVERYCONFLICT);
+ msg.m_databaseid = dboid;
+ msg.m_reason = PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT;
+ pgstat_send(&msg, sizeof(msg));
+}
+
/* ----------
* pgstat_ping() -
*
@@ -3489,6 +3505,7 @@ reset_dbentry_counters(PgStat_StatDBEntry *dbentry)
dbentry->n_conflict_tablespace = 0;
dbentry->n_conflict_lock = 0;
dbentry->n_conflict_snapshot = 0;
+ dbentry->n_conflict_logicalslot = 0;
dbentry->n_conflict_bufferpin = 0;
dbentry->n_conflict_startup_deadlock = 0;
dbentry->n_temp_files = 0;
@@ -5425,6 +5442,9 @@ pgstat_recv_recoveryconflict(PgStat_MsgRecoveryConflict *msg, int len)
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
dbentry->n_conflict_snapshot++;
break;
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ dbentry->n_conflict_logicalslot++;
+ break;
case PROCSIG_RECOVERY_CONFLICT_BUFFERPIN:
dbentry->n_conflict_bufferpin++;
break;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 1f38c5b33e..1d9400ea63 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -241,11 +241,14 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
/*
* After the sanity checks in CreateDecodingContext, make sure the
- * restart_lsn is valid. Avoid "cannot get changes" wording in this
+ * restart_lsn is valid or both xmin and catalog_xmin are valid.
+ * Avoid "cannot get changes" wording in this
* errmsg because that'd be confusingly ambiguous about no changes
* being available.
*/
- if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
+ if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn)
+ || (!TransactionIdIsValid(MyReplicationSlot->data.xmin)
+ && !TransactionIdIsValid(MyReplicationSlot->data.catalog_xmin)))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("can no longer get changes from replication slot \"%s\"",
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 8c18b4ed05..f9231229cc 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1317,6 +1317,215 @@ restart:
LWLockRelease(ReplicationSlotControlLock);
}
+/*
+ * Helper for InvalidateConflictingLogicalReplicationSlot -- acquires the given slot
+ * and mark it invalid, if necessary and possible.
+ *
+ * Returns whether ReplicationSlotControlLock was released in the interim (and
+ * in that case we're not holding the lock at return, otherwise we are).
+ *
+ * This is inherently racy, because we release the LWLock
+ * for syscalls, so caller must restart if we return true.
+ */
+static bool
+InvalidatePossiblyConflictingLogicalReplicationSlot(ReplicationSlot *s, TransactionId xid)
+{
+ int last_signaled_pid = 0;
+ bool released_lock = false;
+
+ for (;;)
+ {
+ TransactionId slot_xmin;
+ TransactionId slot_catalog_xmin;
+ NameData slotname;
+ int active_pid = 0;
+
+ Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ if (!s->in_use)
+ {
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+ break;
+ }
+
+ /*
+ * Check if the slot needs to be invalidated. If it needs to be
+ * invalidated, and is not currently acquired, acquire it and mark it
+ * as having been invalidated. We do this with the spinlock held to
+ * avoid race conditions -- for example the xmin(s) could move forward
+ * , or the slot could be dropped.
+ */
+ SpinLockAcquire(&s->mutex);
+
+ slot_xmin = s->data.xmin;
+ slot_catalog_xmin = s->data.catalog_xmin;
+
+ /*
+ * If the slot is already invalid or is not conflicting, we don't need to
+ * do anything.
+ */
+
+ /* slot has been invalidated */
+ if ((!TransactionIdIsValid(slot_xmin) && !TransactionIdIsValid(slot_catalog_xmin))
+ ||
+ /*
+ * we are not forcing for invalidation because the xid is valid
+ * and this is a non conflicting slot
+ */
+ (TransactionIdIsValid(xid) && !(
+ (TransactionIdIsValid(slot_xmin) && TransactionIdPrecedesOrEquals(slot_xmin, xid))
+ ||
+ (TransactionIdIsValid(slot_catalog_xmin) && TransactionIdPrecedesOrEquals(slot_catalog_xmin, xid))
+ ))
+ )
+ {
+ SpinLockRelease(&s->mutex);
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+ break;
+ }
+
+ slotname = s->data.name;
+ active_pid = s->active_pid;
+
+ /*
+ * If the slot can be acquired, do so and mark it invalidated
+ * immediately. Otherwise we'll signal the owning process, below, and
+ * retry.
+ */
+ if (active_pid == 0)
+ {
+ MyReplicationSlot = s;
+ s->active_pid = MyProcPid;
+ s->data.xmin = InvalidTransactionId;
+ s->data.catalog_xmin = InvalidTransactionId;
+ }
+
+ SpinLockRelease(&s->mutex);
+
+ if (active_pid != 0)
+ {
+ /*
+ * Prepare the sleep on the slot's condition variable before
+ * releasing the lock, to close a possible race condition if the
+ * slot is released before the sleep below.
+ */
+
+ ConditionVariablePrepareToSleep(&s->active_cv);
+
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ /*
+ * Signal to terminate the process that owns the slot, if we
+ * haven't already signalled it. (Avoidance of repeated
+ * signalling is the only reason for there to be a loop in this
+ * routine; otherwise we could rely on caller's restart loop.)
+ *
+ * There is the race condition that other process may own the slot
+ * after its current owner process is terminated and before this
+ * process owns it. To handle that, we signal only if the PID of
+ * the owning process has changed from the previous time. (This
+ * logic assumes that the same PID is not reused very quickly.)
+ */
+ if (last_signaled_pid != active_pid)
+ {
+ ereport(LOG,
+ (errmsg("terminating process %d because replication slot \"%s\" conflicts with recovery",
+ active_pid, NameStr(slotname))));
+
+ (void) kill(active_pid, SIGTERM);
+ last_signaled_pid = active_pid;
+ }
+
+ /* Wait until the slot is released. */
+ ConditionVariableSleep(&s->active_cv,
+ WAIT_EVENT_REPLICATION_SLOT_DROP);
+
+ /*
+ * Re-acquire lock and start over; we expect to invalidate the
+ * slot next time (unless another process acquires the slot in the
+ * meantime).
+ */
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ continue;
+ }
+ else
+ {
+ /*
+ * We hold the slot now and have already invalidated it; flush it
+ * to ensure that state persists.
+ *
+ * Don't want to hold ReplicationSlotControlLock across file
+ * system operations, so release it now but be sure to tell caller
+ * to restart from scratch.
+ */
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ /* Make sure the invalidated state persists across server restart */
+ ReplicationSlotMarkDirty();
+ ReplicationSlotSave();
+ ReplicationSlotRelease();
+ pgstat_report_replslot_conflict(s->data.database);
+
+ ereport(LOG,
+ (errmsg("invalidating slot \"%s\" because it conflicts with recovery", NameStr(slotname))));
+
+ /* done with this slot for now */
+ break;
+ }
+ }
+
+ Assert(!released_lock == LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ return released_lock;
+}
+
+/*
+ * Resolve recovery conflicts with logical slots.
+ *
+ * When xid is valid, it means that we are about to remove rows older than xid.
+ * Therefore we need to invalidate slots that depend on seeing those rows.
+ * When xid is invalid, invalidate all logical slots. This is required when the
+ * master wal_level is set back to replica, so existing logical slots need to
+ * be invalidated.
+ */
+void
+InvalidateConflictingLogicalReplicationSlots(Oid dboid, TransactionId xid)
+{
+
+ Assert(max_replication_slots >= 0);
+
+ if (max_replication_slots == 0)
+ return;
+restart:
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ for (int i = 0; i < max_replication_slots; i++)
+ {
+ ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+ if (!s->in_use)
+ continue;
+
+ /* We are only dealing with *logical* slot conflicts. */
+ if (!SlotIsLogical(s))
+ continue;
+
+ /* not our database and we don't want all the database, skip */
+ if ((s->data.database != InvalidOid && s->data.database != dboid) && TransactionIdIsValid(xid))
+ continue;
+
+ if (InvalidatePossiblyConflictingLogicalReplicationSlot(s, xid))
+ {
+ /* if the lock was released, we need to restart from scratch */
+ goto restart;
+ }
+ }
+ LWLockRelease(ReplicationSlotControlLock);
+}
+
/*
* Flush all replication slots to disk.
*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3224536356..53e440c2ae 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1139,6 +1139,14 @@ StartLogicalReplication(StartReplicationCmd *cmd)
ReplicationSlotAcquire(cmd->slotname, true);
+ if (!TransactionIdIsValid(MyReplicationSlot->data.xmin)
+ && !TransactionIdIsValid(MyReplicationSlot->data.catalog_xmin))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot read from logical replication slot \"%s\"",
+ cmd->slotname),
+ errdetail("This slot has been invalidated because it was conflicting with recovery.")));
+
if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index e4c008e443..bdc9ba1fc0 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -3401,6 +3401,10 @@ SignalVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode,
GET_VXID_FROM_PGPROC(procvxid, *proc);
+ /*
+ * Note: vxid.localTransactionId can be invalid, which means the
+ * request is to signal the pid that is not running a transaction.
+ */
if (procvxid.backendId == vxid.backendId &&
procvxid.localTransactionId == vxid.localTransactionId)
{
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index defb75aa26..315d5a1e33 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -673,6 +673,9 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT);
+ if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT))
+ RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT);
+
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK);
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 553b6e5460..b67e79c55a 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -23,6 +23,7 @@
#include "access/xloginsert.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "replication/slot.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
@@ -440,7 +441,8 @@ ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlist,
}
void
-ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode node)
+ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid,
+ bool onCatalogTable, RelFileNode node)
{
VirtualTransactionId *backends;
@@ -465,6 +467,9 @@ ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode
PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
WAIT_EVENT_RECOVERY_CONFLICT_SNAPSHOT,
true);
+
+ if (onCatalogTable)
+ InvalidateConflictingLogicalReplicationSlots(node.dbNode, latestRemovedXid);
}
/*
@@ -473,7 +478,7 @@ ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode
*/
void
ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXid,
- RelFileNode node)
+ bool onCatalogTable, RelFileNode node)
{
/*
* ResolveRecoveryConflictWithSnapshot operates on 32-bit TransactionIds,
@@ -491,7 +496,7 @@ ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXi
TransactionId latestRemovedXid;
latestRemovedXid = XidFromFullTransactionId(latestRemovedFullXid);
- ResolveRecoveryConflictWithSnapshot(latestRemovedXid, node);
+ ResolveRecoveryConflictWithSnapshot(latestRemovedXid, onCatalogTable, node);
}
}
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 8cea10c901..477c96df9d 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -2467,6 +2467,9 @@ errdetail_recovery_conflict(void)
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
errdetail("User query might have needed to see row versions that must be removed.");
break;
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ errdetail("User was using the logical slot that must be dropped.");
+ break;
case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK:
errdetail("User transaction caused buffer deadlock with recovery.");
break;
@@ -3037,6 +3040,25 @@ RecoveryConflictInterrupt(ProcSignalReason reason)
case PROCSIG_RECOVERY_CONFLICT_LOCK:
case PROCSIG_RECOVERY_CONFLICT_TABLESPACE:
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ /*
+ * For conflicts that require a logical slot to be invalidated, the
+ * requirement is for the signal receiver to release the slot,
+ * so that it could be invalidated by the signal sender. So for
+ * normal backends, the transaction should be aborted, just
+ * like for other recovery conflicts. But if it's walsender on
+ * standby, then it has to be killed so as to release an
+ * acquired logical slot.
+ */
+ if (am_cascading_walsender &&
+ reason == PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT &&
+ MyReplicationSlot && SlotIsLogical(MyReplicationSlot))
+ {
+ RecoveryConflictPending = true;
+ QueryCancelPending = true;
+ InterruptPending = true;
+ break;
+ }
/*
* If we aren't in a transaction any longer then ignore.
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 14056f5347..e64d2666d0 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -1500,6 +1500,21 @@ pg_stat_get_db_conflict_snapshot(PG_FUNCTION_ARGS)
PG_RETURN_INT64(result);
}
+Datum
+pg_stat_get_db_conflict_logicalslot(PG_FUNCTION_ARGS)
+{
+ Oid dbid = PG_GETARG_OID(0);
+ int64 result;
+ PgStat_StatDBEntry *dbentry;
+
+ if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (dbentry->n_conflict_logicalslot);
+
+ PG_RETURN_INT64(result);
+}
+
Datum
pg_stat_get_db_conflict_bufferpin(PG_FUNCTION_ARGS)
{
@@ -1543,6 +1558,7 @@ pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS)
result = (int64) (dbentry->n_conflict_tablespace +
dbentry->n_conflict_lock +
dbentry->n_conflict_snapshot +
+ dbentry->n_conflict_logicalslot +
dbentry->n_conflict_bufferpin +
dbentry->n_conflict_startup_deadlock);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index fde251fa4f..52a6366042 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5434,6 +5434,11 @@
proname => 'pg_stat_get_db_conflict_snapshot', provolatile => 's',
proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
prosrc => 'pg_stat_get_db_conflict_snapshot' },
+{ oid => '4544',
+ descr => 'statistics: recovery conflicts in database caused by logical replication slot',
+ proname => 'pg_stat_get_db_conflict_logicalslot', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+ prosrc => 'pg_stat_get_db_conflict_logicalslot' },
{ oid => '3068',
descr => 'statistics: recovery conflicts in database caused by shared buffer pin',
proname => 'pg_stat_get_db_conflict_bufferpin', provolatile => 's',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 9612c0a6c2..52623b4be9 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -743,6 +743,7 @@ typedef struct PgStat_StatDBEntry
PgStat_Counter n_conflict_tablespace;
PgStat_Counter n_conflict_lock;
PgStat_Counter n_conflict_snapshot;
+ PgStat_Counter n_conflict_logicalslot;
PgStat_Counter n_conflict_bufferpin;
PgStat_Counter n_conflict_startup_deadlock;
PgStat_Counter n_temp_files;
@@ -1011,6 +1012,7 @@ extern void pgstat_report_checksum_failure(void);
extern void pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat);
extern void pgstat_report_replslot_create(const char *slotname);
extern void pgstat_report_replslot_drop(const char *slotname);
+extern void pgstat_report_replslot_conflict(Oid dbOid);
extern void pgstat_initialize(void);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 2eb7e3a530..bd292e5e14 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -215,6 +215,7 @@ extern XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void);
extern bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive);
extern void ReplicationSlotsDropDBSlots(Oid dboid);
extern void InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno);
+extern void InvalidateConflictingLogicalReplicationSlots(Oid dboid, TransactionId xid);
extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, int szslot);
extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
@@ -224,4 +225,6 @@ extern void CheckPointReplicationSlots(void);
extern void CheckSlotRequirements(void);
+extern void ResolveRecoveryConflictWithLogicalSlots(Oid dboid, TransactionId xid, char *reason);
+
#endif /* SLOT_H */
diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h
index eec186be2e..f86b070dbc 100644
--- a/src/include/storage/procsignal.h
+++ b/src/include/storage/procsignal.h
@@ -41,6 +41,7 @@ typedef enum
PROCSIG_RECOVERY_CONFLICT_TABLESPACE,
PROCSIG_RECOVERY_CONFLICT_LOCK,
PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
+ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT,
PROCSIG_RECOVERY_CONFLICT_BUFFERPIN,
PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK,
diff --git a/src/include/storage/standby.h b/src/include/storage/standby.h
index 38fd85a431..3ba1882216 100644
--- a/src/include/storage/standby.h
+++ b/src/include/storage/standby.h
@@ -30,9 +30,9 @@ extern void InitRecoveryTransactionEnvironment(void);
extern void ShutdownRecoveryTransactionEnvironment(void);
extern void ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid,
- RelFileNode node);
+ bool onCatalogTable, RelFileNode node);
extern void ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXid,
- RelFileNode node);
+ bool onCatalogTable, RelFileNode node);
extern void ResolveRecoveryConflictWithTablespace(Oid tsid);
extern void ResolveRecoveryConflictWithDatabase(Oid dbid);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index e5ab11275d..66f28b649d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1871,7 +1871,8 @@ pg_stat_database_conflicts| SELECT d.oid AS datid,
pg_stat_get_db_conflict_lock(d.oid) AS confl_lock,
pg_stat_get_db_conflict_snapshot(d.oid) AS confl_snapshot,
pg_stat_get_db_conflict_bufferpin(d.oid) AS confl_bufferpin,
- pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock
+ pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock,
+ pg_stat_get_db_conflict_logicalslot(d.oid) AS confl_logicalslot
FROM pg_database d;
pg_stat_gssapi| SELECT s.pid,
s.gss_auth AS gss_authenticated,
--
2.18.4
From e685cf4ef639dbc92992b66d2a33010b4a2cde6f Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:04:15 +0000
Subject: [PATCH v19 1/5] Add info in WAL records in preparation for logical
slot conflict handling.
When a WAL replay on standby indicates that a catalog table tuple is
to be deleted by an xid that is greater than a logical slot's
catalog_xmin, then that means the slot's catalog_xmin conflicts with
the xid, and we need to handle the conflict. While subsequent commits
will do the actual conflict handling, this commit adds a new field
onCatalogTable in such WAL records, that is true for catalog tables,
so as to arrange for conflict handling.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/backend/access/gist/gist.c | 2 +-
src/backend/access/gist/gistbuild.c | 2 +-
src/backend/access/gist/gistutil.c | 4 ++--
src/backend/access/gist/gistxlog.c | 4 +++-
src/backend/access/hash/hashinsert.c | 1 +
src/backend/access/heap/heapam.c | 4 +++-
src/backend/access/heap/pruneheap.c | 1 +
src/backend/access/heap/visibilitymap.c | 2 +-
src/backend/access/nbtree/nbtpage.c | 12 +++++++++---
src/backend/access/spgist/spgvacuum.c | 8 ++++++++
src/backend/utils/cache/lsyscache.c | 15 +++++++++++++++
src/include/access/gist_private.h | 6 +++---
src/include/access/gistxlog.h | 3 ++-
src/include/access/hash_xlog.h | 1 +
src/include/access/heapam_xlog.h | 5 ++++-
src/include/access/nbtxlog.h | 2 ++
src/include/access/spgxlog.h | 1 +
src/include/utils/lsyscache.h | 1 +
src/include/utils/rel.h | 9 +++++++++
19 files changed, 68 insertions(+), 15 deletions(-)
17.9% src/backend/access/gist/
13.1% src/backend/access/heap/
14.5% src/backend/access/nbtree/
8.9% src/backend/access/spgist/
7.2% src/backend/utils/cache/
19.6% src/include/access/
16.6% src/include/utils/
diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c
index 0683f42c25..b6e6340c3c 100644
--- a/src/backend/access/gist/gist.c
+++ b/src/backend/access/gist/gist.c
@@ -348,7 +348,7 @@ gistplacetopage(Relation rel, Size freespace, GISTSTATE *giststate,
for (; ptr; ptr = ptr->next)
{
/* Allocate new page */
- ptr->buffer = gistNewBuffer(rel);
+ ptr->buffer = gistNewBuffer(heapRel, rel);
GISTInitBuffer(ptr->buffer, (is_leaf) ? F_LEAF : 0);
ptr->page = BufferGetPage(ptr->buffer);
ptr->block.blkno = BufferGetBlockNumber(ptr->buffer);
diff --git a/src/backend/access/gist/gistbuild.c b/src/backend/access/gist/gistbuild.c
index f46a42197c..80949bd4db 100644
--- a/src/backend/access/gist/gistbuild.c
+++ b/src/backend/access/gist/gistbuild.c
@@ -290,7 +290,7 @@ gistbuild(Relation heap, Relation index, IndexInfo *indexInfo)
Page page;
/* initialize the root page */
- buffer = gistNewBuffer(index);
+ buffer = gistNewBuffer(heap, index);
Assert(BufferGetBlockNumber(buffer) == GIST_ROOT_BLKNO);
page = BufferGetPage(buffer);
diff --git a/src/backend/access/gist/gistutil.c b/src/backend/access/gist/gistutil.c
index 8dcd53c457..8ba6178a5f 100644
--- a/src/backend/access/gist/gistutil.c
+++ b/src/backend/access/gist/gistutil.c
@@ -821,7 +821,7 @@ gistcheckpage(Relation rel, Buffer buf)
* Caller is responsible for initializing the page by calling GISTInitBuffer
*/
Buffer
-gistNewBuffer(Relation r)
+gistNewBuffer(Relation heapRel, Relation r)
{
Buffer buffer;
bool needLock;
@@ -865,7 +865,7 @@ gistNewBuffer(Relation r)
* page's deleteXid.
*/
if (XLogStandbyInfoActive() && RelationNeedsWAL(r))
- gistXLogPageReuse(r, blkno, GistPageGetDeleteXid(page));
+ gistXLogPageReuse(heapRel, r, blkno, GistPageGetDeleteXid(page));
return buffer;
}
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 6464cb9281..46aee6f2a9 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -596,7 +596,8 @@ gistXLogAssignLSN(void)
* Write XLOG record about reuse of a deleted page.
*/
void
-gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemovedXid)
+gistXLogPageReuse(Relation heapRel, Relation rel,
+ BlockNumber blkno, FullTransactionId latestRemovedXid)
{
gistxlogPageReuse xlrec_reuse;
@@ -607,6 +608,7 @@ gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemov
*/
/* XLOG stuff */
+ xlrec_reuse.onCatalogTable = RelationIsAccessibleInLogicalDecoding(heapRel);
xlrec_reuse.node = rel->rd_node;
xlrec_reuse.block = blkno;
xlrec_reuse.latestRemovedFullXid = latestRemovedXid;
diff --git a/src/backend/access/hash/hashinsert.c b/src/backend/access/hash/hashinsert.c
index d254a00b6a..ce223b2c19 100644
--- a/src/backend/access/hash/hashinsert.c
+++ b/src/backend/access/hash/hashinsert.c
@@ -398,6 +398,7 @@ _hash_vacuum_one_page(Relation rel, Relation hrel, Buffer metabuf, Buffer buf)
xl_hash_vacuum_one_page xlrec;
XLogRecPtr recptr;
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(hrel);
xlrec.latestRemovedXid = latestRemovedXid;
xlrec.ntuples = ndeletable;
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 2433998f39..4019a2122e 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -7941,6 +7941,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid,
/* nor when there are no tuples to freeze */
Assert(ntuples > 0);
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(reln);
xlrec.cutoff_xid = cutoff_xid;
xlrec.ntuples = ntuples;
@@ -7971,7 +7972,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid,
* heap_buffer, if necessary.
*/
XLogRecPtr
-log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
+log_heap_visible(Relation rel, Buffer heap_buffer, Buffer vm_buffer,
TransactionId cutoff_xid, uint8 vmflags)
{
xl_heap_visible xlrec;
@@ -7981,6 +7982,7 @@ log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
Assert(BufferIsValid(heap_buffer));
Assert(BufferIsValid(vm_buffer));
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(rel);
xlrec.cutoff_xid = cutoff_xid;
xlrec.flags = vmflags;
XLogBeginInsert();
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index 15ca1b304a..0590b7053c 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -323,6 +323,7 @@ heap_page_prune(Relation relation, Buffer buffer,
xlrec.latestRemovedXid = prstate.latestRemovedXid;
xlrec.nredirected = prstate.nredirected;
xlrec.ndead = prstate.ndead;
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(relation);
XLogBeginInsert();
XLogRegisterData((char *) &xlrec, SizeOfHeapPrune);
diff --git a/src/backend/access/heap/visibilitymap.c b/src/backend/access/heap/visibilitymap.c
index e198df65d8..6e89a08c52 100644
--- a/src/backend/access/heap/visibilitymap.c
+++ b/src/backend/access/heap/visibilitymap.c
@@ -282,7 +282,7 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
if (XLogRecPtrIsInvalid(recptr))
{
Assert(!InRecovery);
- recptr = log_heap_visible(rel->rd_node, heapBuf, vmBuf,
+ recptr = log_heap_visible(rel, heapBuf, vmBuf,
cutoff_xid, flags);
/*
diff --git a/src/backend/access/nbtree/nbtpage.c b/src/backend/access/nbtree/nbtpage.c
index ebec8fa5b8..2d27a3f974 100644
--- a/src/backend/access/nbtree/nbtpage.c
+++ b/src/backend/access/nbtree/nbtpage.c
@@ -36,6 +36,7 @@
#include "utils/memdebug.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
+#include "utils/lsyscache.h"
static BTMetaPageData *_bt_getmeta(Relation rel, Buffer metabuf);
static void _bt_log_reuse_page(Relation rel, BlockNumber blkno,
@@ -43,7 +44,8 @@ static void _bt_log_reuse_page(Relation rel, BlockNumber blkno,
static void _bt_delitems_delete(Relation rel, Buffer buf,
TransactionId latestRemovedXid,
OffsetNumber *deletable, int ndeletable,
- BTVacuumPosting *updatable, int nupdatable);
+ BTVacuumPosting *updatable, int nupdatable,
+ Relation heapRel);
static char *_bt_delitems_update(BTVacuumPosting *updatable, int nupdatable,
OffsetNumber *updatedoffsets,
Size *updatedbuflen, bool needswal);
@@ -836,6 +838,7 @@ _bt_log_reuse_page(Relation rel, BlockNumber blkno, FullTransactionId safexid)
*/
/* XLOG stuff */
+ xlrec_reuse.onCatalogTable = get_rel_logical_catalog(rel->rd_index->indrelid);
xlrec_reuse.node = rel->rd_node;
xlrec_reuse.block = blkno;
xlrec_reuse.latestRemovedFullXid = safexid;
@@ -1296,7 +1299,8 @@ _bt_delitems_vacuum(Relation rel, Buffer buf,
static void
_bt_delitems_delete(Relation rel, Buffer buf, TransactionId latestRemovedXid,
OffsetNumber *deletable, int ndeletable,
- BTVacuumPosting *updatable, int nupdatable)
+ BTVacuumPosting *updatable, int nupdatable,
+ Relation heapRel)
{
Page page = BufferGetPage(buf);
BTPageOpaque opaque;
@@ -1358,6 +1362,8 @@ _bt_delitems_delete(Relation rel, Buffer buf, TransactionId latestRemovedXid,
XLogRecPtr recptr;
xl_btree_delete xlrec_delete;
+ xlrec_delete.onCatalogTable =
+ RelationIsAccessibleInLogicalDecoding(heapRel);
xlrec_delete.latestRemovedXid = latestRemovedXid;
xlrec_delete.ndeleted = ndeletable;
xlrec_delete.nupdated = nupdatable;
@@ -1685,7 +1691,7 @@ _bt_delitems_delete_check(Relation rel, Buffer buf, Relation heapRel,
/* Physically delete tuples (or TIDs) using deletable (or updatable) */
_bt_delitems_delete(rel, buf, latestRemovedXid, deletable, ndeletable,
- updatable, nupdatable);
+ updatable, nupdatable, heapRel);
/* be tidy */
for (int i = 0; i < nupdatable; i++)
diff --git a/src/backend/access/spgist/spgvacuum.c b/src/backend/access/spgist/spgvacuum.c
index 76fb0374c4..3186885d14 100644
--- a/src/backend/access/spgist/spgvacuum.c
+++ b/src/backend/access/spgist/spgvacuum.c
@@ -27,6 +27,7 @@
#include "storage/indexfsm.h"
#include "storage/lmgr.h"
#include "utils/snapmgr.h"
+#include "utils/lsyscache.h"
/* Entry in pending-list of TIDs we need to revisit */
@@ -503,6 +504,13 @@ vacuumRedirectAndPlaceholder(Relation index, Buffer buffer)
spgxlogVacuumRedirect xlrec;
GlobalVisState *vistest;
+ /*
+ * There is no chance of endless recursion even when we are doing catalog
+ * acceses here; because, spgist is never used for catalogs. Check
+ * comments in RelationIsAccessibleInLogicalDecoding().
+ */
+ xlrec.onCatalogTable = get_rel_logical_catalog(index->rd_index->indrelid);
+
xlrec.nToPlaceholder = 0;
xlrec.newestRedirectXid = InvalidTransactionId;
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index 6bba5f8ec4..28b1f961f4 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -18,6 +18,7 @@
#include "access/hash.h"
#include "access/htup_details.h"
#include "access/nbtree.h"
+#include "access/table.h"
#include "bootstrap/bootstrap.h"
#include "catalog/namespace.h"
#include "catalog/pg_am.h"
@@ -2062,6 +2063,20 @@ get_rel_persistence(Oid relid)
return result;
}
+bool
+get_rel_logical_catalog(Oid relid)
+{
+ bool res;
+ Relation rel;
+
+ /* assume previously locked */
+ rel = table_open(relid, NoLock);
+ res = RelationIsAccessibleInLogicalDecoding(rel);
+ table_close(rel, NoLock);
+
+ return res;
+}
+
/* ---------- TRANSFORM CACHE ---------- */
diff --git a/src/include/access/gist_private.h b/src/include/access/gist_private.h
index 553d364e2d..a0f4015556 100644
--- a/src/include/access/gist_private.h
+++ b/src/include/access/gist_private.h
@@ -440,8 +440,8 @@ extern XLogRecPtr gistXLogPageDelete(Buffer buffer,
FullTransactionId xid, Buffer parentBuffer,
OffsetNumber downlinkOffset);
-extern void gistXLogPageReuse(Relation rel, BlockNumber blkno,
- FullTransactionId latestRemovedXid);
+extern void gistXLogPageReuse(Relation heapRel, Relation rel,
+ BlockNumber blkno, FullTransactionId latestRemovedXid);
extern XLogRecPtr gistXLogUpdate(Buffer buffer,
OffsetNumber *todelete, int ntodelete,
@@ -485,7 +485,7 @@ extern bool gistproperty(Oid index_oid, int attno,
extern bool gistfitpage(IndexTuple *itvec, int len);
extern bool gistnospace(Page page, IndexTuple *itvec, int len, OffsetNumber todelete, Size freespace);
extern void gistcheckpage(Relation rel, Buffer buf);
-extern Buffer gistNewBuffer(Relation r);
+extern Buffer gistNewBuffer(Relation heapRel, Relation r);
extern bool gistPageRecyclable(Page page);
extern void gistfillbuffer(Page page, IndexTuple *itup, int len,
OffsetNumber off);
diff --git a/src/include/access/gistxlog.h b/src/include/access/gistxlog.h
index fd5144f258..73999ddc70 100644
--- a/src/include/access/gistxlog.h
+++ b/src/include/access/gistxlog.h
@@ -49,9 +49,9 @@ typedef struct gistxlogPageUpdate
*/
typedef struct gistxlogDelete
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 ntodelete; /* number of deleted offsets */
-
/*
* In payload of blk 0 : todelete OffsetNumbers
*/
@@ -97,6 +97,7 @@ typedef struct gistxlogPageDelete
*/
typedef struct gistxlogPageReuse
{
+ bool onCatalogTable;
RelFileNode node;
BlockNumber block;
FullTransactionId latestRemovedFullXid;
diff --git a/src/include/access/hash_xlog.h b/src/include/access/hash_xlog.h
index 4353a32dbb..94c3292c1e 100644
--- a/src/include/access/hash_xlog.h
+++ b/src/include/access/hash_xlog.h
@@ -250,6 +250,7 @@ typedef struct xl_hash_init_bitmap_page
*/
typedef struct xl_hash_vacuum_one_page
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
int ntuples;
diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h
index 27db48184e..eba48b0aee 100644
--- a/src/include/access/heapam_xlog.h
+++ b/src/include/access/heapam_xlog.h
@@ -242,6 +242,7 @@ typedef struct xl_heap_update
*/
typedef struct xl_heap_prune
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 nredirected;
uint16 ndead;
@@ -338,6 +339,7 @@ typedef struct xl_heap_freeze_tuple
*/
typedef struct xl_heap_freeze_page
{
+ bool onCatalogTable;
TransactionId cutoff_xid;
uint16 ntuples;
} xl_heap_freeze_page;
@@ -352,6 +354,7 @@ typedef struct xl_heap_freeze_page
*/
typedef struct xl_heap_visible
{
+ bool onCatalogTable;
TransactionId cutoff_xid;
uint8 flags;
} xl_heap_visible;
@@ -413,7 +416,7 @@ extern bool heap_prepare_freeze_tuple(HeapTupleHeader tuple,
bool *totally_frozen);
extern void heap_execute_freeze_tuple(HeapTupleHeader tuple,
xl_heap_freeze_tuple *xlrec_tp);
-extern XLogRecPtr log_heap_visible(RelFileNode rnode, Buffer heap_buffer,
+extern XLogRecPtr log_heap_visible(Relation rel, Buffer heap_buffer,
Buffer vm_buffer, TransactionId cutoff_xid, uint8 flags);
#endif /* HEAPAM_XLOG_H */
diff --git a/src/include/access/nbtxlog.h b/src/include/access/nbtxlog.h
index 0f7731856b..b15aa47f1b 100644
--- a/src/include/access/nbtxlog.h
+++ b/src/include/access/nbtxlog.h
@@ -185,6 +185,7 @@ typedef struct xl_btree_dedup
*/
typedef struct xl_btree_reuse_page
{
+ bool onCatalogTable;
RelFileNode node;
BlockNumber block;
FullTransactionId latestRemovedFullXid;
@@ -232,6 +233,7 @@ typedef struct xl_btree_vacuum
typedef struct xl_btree_delete
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 ndeleted;
uint16 nupdated;
diff --git a/src/include/access/spgxlog.h b/src/include/access/spgxlog.h
index 69405b5750..06b91f4d04 100644
--- a/src/include/access/spgxlog.h
+++ b/src/include/access/spgxlog.h
@@ -237,6 +237,7 @@ typedef struct spgxlogVacuumRoot
typedef struct spgxlogVacuumRedirect
{
+ bool onCatalogTable;
uint16 nToPlaceholder; /* number of redirects to make placeholders */
OffsetNumber firstPlaceholder; /* first placeholder tuple to remove */
TransactionId newestRedirectXid; /* newest XID of removed redirects */
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index 77871aaefc..e2a5efed30 100644
--- a/src/include/utils/lsyscache.h
+++ b/src/include/utils/lsyscache.h
@@ -139,6 +139,7 @@ extern char get_rel_relkind(Oid relid);
extern bool get_rel_relispartition(Oid relid);
extern Oid get_rel_tablespace(Oid relid);
extern char get_rel_persistence(Oid relid);
+extern bool get_rel_logical_catalog(Oid relid);
extern Oid get_transform_fromsql(Oid typid, Oid langid, List *trftypes);
extern Oid get_transform_tosql(Oid typid, Oid langid, List *trftypes);
extern bool get_typisdefined(Oid typid);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 77d176a934..b488bb9618 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -26,6 +26,7 @@
#include "storage/relfilenode.h"
#include "utils/relcache.h"
#include "utils/reltrigger.h"
+#include "catalog/catalog.h"
/*
@@ -365,6 +366,9 @@ typedef struct StdRdOptions
* RelationIsUsedAsCatalogTable
* Returns whether the relation should be treated as a catalog table
* from the pov of logical decoding. Note multiple eval of argument!
+ * This definition should not invoke anything that performs catalog
+ * access; otherwise it may cause infinite recursion. Check the comments
+ * in RelationIsAccessibleInLogicalDecoding() for details.
*/
#define RelationIsUsedAsCatalogTable(relation) \
((relation)->rd_options && \
@@ -642,6 +646,11 @@ typedef struct ViewOptions
* RelationIsAccessibleInLogicalDecoding
* True if we need to log enough information to have access via
* decoding snapshot.
+ * This definition should not invoke anything that performs catalog
+ * access. Otherwise, e.g. logging a WAL entry for catalog relation may
+ * invoke this function, which will in turn do catalog access, which may
+ * in turn cause another similar WAL entry to be logged, leading to
+ * infinite recursion.
*/
#define RelationIsAccessibleInLogicalDecoding(relation) \
(XLogLogicalInfoActive() && \
--
2.18.4
Attachments:
[text/plain] v19-0005-Doc-changes-describing-details-about-logical-dec.patch (2.1K, ../[email protected]/2-v19-0005-Doc-changes-describing-details-about-logical-dec.patch)
download | inline diff:
From c0eae773a83916aa8dd652a67d8ef542fec327b7 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:20:45 +0000
Subject: [PATCH v19 5/5] Doc changes describing details about logical
decoding.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
doc/src/sgml/logicaldecoding.sgml | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
100.0% doc/src/sgml/
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index 1765ea6c87..de67bdde8b 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -297,6 +297,27 @@ postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NU
may consume changes from a slot at any given time.
</para>
+ <para>
+ A logical replication slot can also be created on a hot standby. To prevent
+ <command>VACUUM</command> from removing required rows from the system
+ catalogs, <varname>hot_standby_feedback</varname> should be set on the
+ standby. In spite of that, if any required rows get removed, the slot gets
+ invalidated. It's highly recommended to use a physical slot between the primary
+ and the standby. Otherwise, hot_standby_feedback will work, but only while the
+ connection is alive (for example a node restart would break it). Existing
+ logical slots on standby also get invalidated if wal_level on primary is reduced to
+ less than 'logical'.
+ </para>
+
+ <para>
+ For a logical slot to be created, it builds a historic snapshot, for which
+ information of all the currently running transactions is essential. On
+ primary, this information is available, but on standby, this information
+ has to be obtained from primary. So, slot creation may wait for some
+ activity to happen on the primary. If the primary is idle, creating a
+ logical slot on standby may take a noticeable time.
+ </para>
+
<caution>
<para>
Replication slots persist across crashes and know nothing about the state
--
2.18.4
[text/plain] v19-0004-New-TAP-test-for-logical-decoding-on-standby.patch (16.9K, ../[email protected]/3-v19-0004-New-TAP-test-for-logical-decoding-on-standby.patch)
download | inline diff:
From 1e54c6ed8d75fd82cc4752f9fa078cdc54cd1f41 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:19:13 +0000
Subject: [PATCH v19 4/5] New TAP test for logical decoding on standby.
Author: Craig Ringer (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/test/perl/PostgresNode.pm | 37 ++
.../t/026_standby_logical_decoding.pl | 426 ++++++++++++++++++
2 files changed, 463 insertions(+)
7.2% src/test/perl/
92.7% src/test/recovery/t/
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 2027cbf43d..4984b2f6f4 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -2612,6 +2612,43 @@ sub pg_recvlogical_upto
=pod
+=item $node->create_logical_slot_on_standby(self, master, slot_name, dbname)
+
+Create logical replication slot on given standby
+
+=cut
+
+sub create_logical_slot_on_standby
+{
+ my ($self, $master, $slot_name, $dbname) = @_;
+ my ($stdout, $stderr);
+
+ my $handle;
+
+ $handle = IPC::Run::start(['pg_recvlogical', '-d', $self->connstr($dbname), '-P', 'test_decoding', '-S', $slot_name, '--create-slot'], '>', \$stdout, '2>', \$stderr);
+
+ # Once slot restart_lsn is created, the standby looks for xl_running_xacts
+ # WAL record from the restart_lsn onwards. So firstly, wait until the slot
+ # restart_lsn is evaluated.
+
+ $self->poll_query_until(
+ 'postgres', qq[
+ SELECT restart_lsn IS NOT NULL
+ FROM pg_catalog.pg_replication_slots WHERE slot_name = '$slot_name'
+ ]) or die "timed out waiting for logical slot to calculate its restart_lsn";
+
+ # Now arrange for the xl_running_xacts record for which pg_recvlogical
+ # is waiting.
+ $master->safe_psql('postgres', 'CHECKPOINT');
+
+ $handle->finish();
+
+ is($self->slot($slot_name)->{'slot_type'}, 'logical', $slot_name . ' on standby created')
+ or die "could not create slot" . $slot_name;
+}
+
+=pod
+
=back
=cut
diff --git a/src/test/recovery/t/026_standby_logical_decoding.pl b/src/test/recovery/t/026_standby_logical_decoding.pl
new file mode 100644
index 0000000000..1687c63932
--- /dev/null
+++ b/src/test/recovery/t/026_standby_logical_decoding.pl
@@ -0,0 +1,426 @@
+# logical decoding on standby : test logical decoding,
+# recovery conflict and standby promotion.
+
+use strict;
+use warnings;
+
+use PostgresNode;
+use Test::More tests => 32;
+use Time::HiRes qw(usleep);
+
+my ($stdin, $stdout, $stderr, $ret, $handle, $slot);
+
+my $node_primary = get_new_node('primary');
+my $node_standby = get_new_node('standby');
+
+# Name for the physical slot on primary
+my $primary_slotname = 'primary_physical';
+
+# return the size of logfile of $node in bytes
+sub get_log_size
+{
+ my ($node) = @_;
+
+ return (stat $node->logfile)[7];
+}
+
+# find $pat in logfile of $node after $off-th byte
+sub find_in_log
+{
+ my ($node, $pat, $off) = @_;
+
+ $off = 0 unless defined $off;
+ my $log = TestLib::slurp_file($node->logfile);
+ return 0 if (length($log) <= $off);
+
+ $log = substr($log, $off);
+
+ return $log =~ m/$pat/;
+}
+
+# Fetch xmin columns from slot's pg_replication_slots row, after waiting for
+# given boolean condition to be true to ensure we've reached a quiescent state.
+sub wait_for_xmins
+{
+ my ($node, $slotname, $check_expr) = @_;
+
+ $node->poll_query_until(
+ 'postgres', qq[
+ SELECT $check_expr
+ FROM pg_catalog.pg_replication_slots
+ WHERE slot_name = '$slotname';
+ ]) or die "Timed out waiting for slot xmins to advance";
+}
+
+# Create the required logical slots on standby.
+sub create_logical_slots
+{
+ $node_standby->create_logical_slot_on_standby($node_primary, 'inactiveslot', 'testdb');
+ $node_standby->create_logical_slot_on_standby($node_primary, 'activeslot', 'testdb');
+}
+
+# Acquire one of the standby logical slots created by create_logical_slots().
+# In case wait is true we are waiting for an active pid on the 'activeslot' slot.
+# If wait is not true it means we are testing a known failure scenario.
+sub make_slot_active
+{
+ my $wait = shift;
+ my $slot_user_handle;
+
+ print "starting pg_recvlogical\n";
+ $slot_user_handle = IPC::Run::start(['pg_recvlogical', '-d', $node_standby->connstr('testdb'), '-S', 'activeslot', '-f', '-', '--no-loop', '--start'], '>', \$stdout, '2>', \$stderr);
+
+ if ($wait)
+ # make sure activeslot is in use
+ {
+ $node_standby->poll_query_until('testdb',
+ "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'activeslot' AND active_pid IS NOT NULL)"
+ ) or die "slot never became active";
+ }
+
+ return $slot_user_handle;
+}
+
+# Check if all the slots on standby are dropped. These include the 'activeslot'
+# that was acquired by make_slot_active(), and the non-active 'inactiveslot'.
+sub check_slots_dropped
+{
+ my ($slot_user_handle) = @_;
+ my $return;
+
+ is($node_standby->slot('inactiveslot')->{'slot_type'}, '', 'inactiveslot on standby dropped');
+ is($node_standby->slot('activeslot')->{'slot_type'}, '', 'activeslot on standby dropped');
+
+ # our client should've terminated in response to the walsender error
+ $slot_user_handle->finish;
+ $return = $?;
+ cmp_ok($return, "!=", 0, "pg_recvlogical exited non-zero");
+ if ($return) {
+ like($stderr, qr/conflict with recovery/, 'slot have been dropped');
+ }
+
+ return 0;
+}
+
+
+########################
+# Initialize primary node
+########################
+
+$node_primary->init(allows_streaming => 1, has_archiving => 1);
+$node_primary->append_conf('postgresql.conf', q{
+wal_level = 'logical'
+max_replication_slots = 4
+max_wal_senders = 4
+log_min_messages = 'debug2'
+log_error_verbosity = verbose
+});
+$node_primary->dump_info;
+$node_primary->start;
+
+$node_primary->psql('postgres', q[CREATE DATABASE testdb]);
+
+$node_primary->safe_psql('testdb', qq[SELECT * FROM pg_create_physical_replication_slot('$primary_slotname');]);
+my $backup_name = 'b1';
+$node_primary->backup($backup_name);
+
+#######################
+# Initialize standby node
+#######################
+
+$node_standby->init_from_backup(
+ $node_primary, $backup_name,
+ has_streaming => 1,
+ has_restoring => 1);
+$node_standby->append_conf('postgresql.conf',
+ qq[primary_slot_name = '$primary_slotname']);
+$node_standby->start;
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+
+##################################################
+# Test that logical decoding on the standby
+# behaves correctly.
+##################################################
+
+create_logical_slots();
+
+$node_primary->safe_psql('testdb', qq[CREATE TABLE decoding_test(x integer, y text);]);
+$node_primary->safe_psql('testdb', qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,10) s;]);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $result = $node_standby->safe_psql('testdb',
+ qq[SELECT pg_logical_slot_get_changes('activeslot', NULL, NULL);]);
+
+# test if basic decoding works
+is(scalar(my @foobar = split /^/m, $result),
+ 14, 'Decoding produced 14 rows');
+
+# Insert some rows and verify that we get the same results from pg_recvlogical
+# and the SQL interface.
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
+);
+
+my $expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT};
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $stdout_sql = $node_standby->safe_psql('testdb',
+ qq[SELECT data FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
+);
+
+is($stdout_sql, $expected, 'got expected output from SQL decoding session');
+
+my $endpos = $node_standby->safe_psql('testdb',
+ "SELECT lsn FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
+);
+print "waiting to replay $endpos\n";
+
+# Insert some rows after $endpos, which we won't read.
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,50) s;]
+);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+my $stdout_recv = $node_standby->pg_recvlogical_upto(
+ 'testdb', 'activeslot', $endpos, 180,
+ 'include-xids' => '0',
+ 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, $expected,
+ 'got same expected output from pg_recvlogical decoding session');
+
+$node_standby->poll_query_until('testdb',
+ "SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'activeslot' AND active_pid IS NULL)"
+) or die "slot never became inactive";
+
+$stdout_recv = $node_standby->pg_recvlogical_upto(
+ 'testdb', 'activeslot', $endpos, 180,
+ 'include-xids' => '0',
+ 'skip-empty-xacts' => '1');
+chomp($stdout_recv);
+is($stdout_recv, '', 'pg_recvlogical acknowledged changes');
+
+$node_primary->safe_psql('postgres', 'CREATE DATABASE otherdb');
+
+is( $node_primary->psql(
+ 'otherdb',
+ "SELECT lsn FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL) ORDER BY lsn DESC LIMIT 1;"
+ ),
+ 3,
+ 'replaying logical slot from another database fails');
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 1 : hot_standby_feedback off
+##################################################
+
+create_logical_slots();
+
+# One way to reproduce recovery conflict is to run VACUUM FULL with
+# hot_standby_feedback turned off on the standby.
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = off
+]);
+$node_standby->restart;
+# ensure walreceiver feedback off by waiting for expected xmin and
+# catalog_xmin on primary. Both should be NULL since hs_feedback is off
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NULL AND catalog_xmin IS NULL");
+
+$handle = make_slot_active(1);
+
+# This should trigger the conflict
+$node_primary->safe_psql('testdb', 'VACUUM FULL');
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery"),
+ 'inactiveslot slot invalidation is logged');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery"),
+ 'activeslot slot invalidation is logged');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# We are not able to read from the slot as it has been invalidated
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\""),
+ 'cannot read from logical replication slot');
+
+# Turn hot_standby_feedback back on
+$node_standby->append_conf('postgresql.conf',q[
+hot_standby_feedback = on
+]);
+$node_standby->restart;
+
+# ensure walreceiver feedback sent by waiting for expected xmin and
+# catalog_xmin on primary. With hot_standby_feedback on, xmin should advance,
+# but catalog_xmin should still remain NULL since there is no logical slot.
+wait_for_xmins($node_primary, $primary_slotname,
+ "xmin IS NOT NULL AND catalog_xmin IS NULL");
+
+##################################################
+# Recovery conflict: Invalidate conflicting slots, including in-use slots
+# Scenario 2 : incorrect wal_level on primary.
+##################################################
+
+# get the position to search from in the standby logfile
+my $logstart = get_log_size($node_standby);
+
+# drop the logical slots
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+
+create_logical_slots();
+
+$handle = make_slot_active(1);
+
+# Make primary wal_level replica. This will trigger slot conflict.
+$node_primary->append_conf('postgresql.conf',q[
+wal_level = 'replica'
+]);
+$node_primary->restart;
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# message should be issued
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"inactiveslot\" because it conflicts with recovery", $logstart),
+ 'inactiveslot slot invalidation is logged');
+
+ok( find_in_log(
+ $node_standby,
+ "invalidating slot \"activeslot\" because it conflicts with recovery", $logstart),
+ 'activeslot slot invalidation is logged');
+
+# Verify that pg_stat_database_conflicts.confl_logicalslot has been updated
+ok( $node_standby->poll_query_until(
+ 'postgres',
+ "select (confl_logicalslot = 2) from pg_stat_database_conflicts where datname = 'testdb'", 't'),
+ 'confl_logicalslot updated') or die "Timed out waiting confl_logicalslot to be updated";
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+ok( find_in_log(
+ $node_standby,
+ "logical decoding on standby requires wal_level >= logical on master", $logstart),
+ 'cannot start replication because wal_level < logical on master');
+
+# Restore primary wal_level
+$node_primary->append_conf('postgresql.conf',q[
+wal_level = 'logical'
+]);
+$node_primary->restart;
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+$handle = make_slot_active(0);
+usleep(100_000);
+
+# as the slot has been invalidated we should not be able to read
+ok( find_in_log(
+ $node_standby,
+ "cannot read from logical replication slot \"activeslot\"", $logstart),
+ 'cannot read from logical replication slot');
+
+##################################################
+# DROP DATABASE should drops it's slots, including active slots.
+##################################################
+
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('inactiveslot')]);
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('activeslot')]);
+create_logical_slots();
+$handle = make_slot_active(1);
+# Create a slot on a database that would not be dropped. This slot should not
+# get dropped.
+$node_standby->create_logical_slot_on_standby($node_primary, 'otherslot', 'postgres');
+
+# dropdb on the primary to verify slots are dropped on standby
+$node_primary->safe_psql('postgres', q[DROP DATABASE testdb]);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+is($node_standby->safe_psql('postgres',
+ q[SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = 'testdb')]), 'f',
+ 'database dropped on standby');
+
+check_slots_dropped($handle);
+
+is($node_standby->slot('otherslot')->{'slot_type'}, 'logical',
+ 'otherslot on standby not dropped');
+
+# Cleanup : manually drop the slot that was not dropped.
+$node_standby->psql('postgres', q[SELECT pg_drop_replication_slot('otherslot')]);
+
+##################################################
+# Test standby promotion and logical decoding bheavior
+# after the standby gets promoted.
+##################################################
+
+$node_primary->psql('postgres', q[CREATE DATABASE testdb]);
+$node_primary->safe_psql('testdb', qq[CREATE TABLE decoding_test(x integer, y text);]);
+
+# create the logical slots
+create_logical_slots();
+
+# Insert some rows before the promotion
+$node_primary->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,4) s;]
+);
+
+$node_primary->wait_for_catchup($node_standby, 'replay', $node_primary->lsn('flush'));
+
+# promote
+$node_standby->promote;
+
+# insert some rows on promoted standby
+$node_standby->safe_psql('testdb',
+ qq[INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(5,7) s;]
+);
+
+
+$expected = q{BEGIN
+table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'
+table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'
+table public.decoding_test: INSERT: x[integer]:3 y[text]:'3'
+table public.decoding_test: INSERT: x[integer]:4 y[text]:'4'
+COMMIT
+BEGIN
+table public.decoding_test: INSERT: x[integer]:5 y[text]:'5'
+table public.decoding_test: INSERT: x[integer]:6 y[text]:'6'
+table public.decoding_test: INSERT: x[integer]:7 y[text]:'7'
+COMMIT};
+
+# check that we are decoding pre and post promotion inserted rows
+$stdout_sql = $node_standby->safe_psql('testdb',
+ qq[SELECT data FROM pg_logical_slot_peek_changes('activeslot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');]
+);
+
+is($stdout_sql, $expected, 'got expected output from SQL decoding session on promoted standby');
--
2.18.4
[text/plain] v19-0003-Allow-logical-decoding-on-standby.patch (17.2K, ../[email protected]/4-v19-0003-Allow-logical-decoding-on-standby.patch)
download | inline diff:
From b71ae587dcf987f231c823c4c2fd88f6229eef61 Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:13:34 +0000
Subject: [PATCH v19 3/5] Allow logical decoding on standby.
Allow a logical slot to be created on standby. Restrict its usage
or its creation if wal_level on primary is less than logical.
During slot creation, it's restart_lsn is set to the last replayed
LSN. Effectively, a logical slot creation on standby waits for an
xl_running_xact record to arrive from primary. Conflicting slots
would be handled in next commits.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/backend/access/transam/xlog.c | 33 ++++++++++-
src/backend/access/transam/xlogfuncs.c | 2 +-
src/backend/access/transam/xlogutils.c | 2 +-
src/backend/postmaster/checkpointer.c | 4 +-
src/backend/replication/logical/decode.c | 22 +++++++-
src/backend/replication/logical/logical.c | 37 ++++++------
.../replication/logical/logicalfuncs.c | 2 +-
src/backend/replication/slot.c | 56 ++++++++++---------
src/backend/replication/slotfuncs.c | 2 +-
src/backend/replication/walreceiver.c | 4 +-
src/backend/replication/walreceiverfuncs.c | 2 +-
src/backend/replication/walsender.c | 26 ++++++---
src/include/access/xlog.h | 3 +-
13 files changed, 132 insertions(+), 63 deletions(-)
16.7% src/backend/access/transam/
32.7% src/backend/replication/logical/
45.2% src/backend/replication/
5.3% src/
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index e863caf3b2..48b43c9fd2 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -5078,6 +5078,17 @@ LocalProcessControlFile(bool reset)
ReadControlFile();
}
+/*
+ * Get the wal_level from the control file. For a standby, this value should be
+ * considered as its active wal_level, because it may be different from what
+ * was originally configured on standby.
+ */
+WalLevel
+GetActiveWalLevelOnStandby(void)
+{
+ return ControlFile->wal_level;
+}
+
/*
* Initialization of shared memory for XLOG
*/
@@ -9599,7 +9610,7 @@ CreateRestartPoint(int flags)
* whichever is later.
*/
receivePtr = GetWalRcvFlushRecPtr(NULL, NULL);
- replayPtr = GetXLogReplayRecPtr(&replayTLI);
+ replayPtr = GetXLogReplayRecPtr(&replayTLI, false);
endptr = (receivePtr < replayPtr) ? replayPtr : receivePtr;
KeepLogSeg(endptr, &_logSegNo);
InvalidateObsoleteReplicationSlots(_logSegNo);
@@ -11711,7 +11722,7 @@ register_persistent_abort_backup_handler(void)
* Exported to allow WALReceiver to read the pointer directly.
*/
XLogRecPtr
-GetXLogReplayRecPtr(TimeLineID *replayTLI)
+GetXLogReplayRecPtr(TimeLineID *replayTLI, bool avoid_header)
{
XLogRecPtr recptr;
TimeLineID tli;
@@ -11723,6 +11734,24 @@ GetXLogReplayRecPtr(TimeLineID *replayTLI)
if (replayTLI)
*replayTLI = tli;
+
+ /*
+ * Replay pointer may point one past the end of the record. If that
+ * is a XLOG page boundary, it will not be a valid LSN for the
+ * start of a record, so bump it up past the page header.
+ */
+ if (avoid_header && !XRecOffIsValid(recptr))
+ {
+ if (recptr % XLOG_BLCKSZ != 0)
+ elog(ERROR, "invalid replay pointer");
+
+ /* For the first page of a segment file, it's a long header */
+ if (XLogSegmentOffset(recptr, wal_segment_size) == 0)
+ recptr += SizeOfXLogLongPHD;
+ else
+ recptr += SizeOfXLogShortPHD;
+ }
+
return recptr;
}
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index b98deb72ec..a173f8d6fc 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -417,7 +417,7 @@ pg_last_wal_replay_lsn(PG_FUNCTION_ARGS)
{
XLogRecPtr recptr;
- recptr = GetXLogReplayRecPtr(NULL);
+ recptr = GetXLogReplayRecPtr(NULL, false);
if (recptr == 0)
PG_RETURN_NULL();
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index d17d660f46..81088baae3 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -850,7 +850,7 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
if (!RecoveryInProgress())
read_upto = GetFlushRecPtr();
else
- read_upto = GetXLogReplayRecPtr(&ThisTimeLineID);
+ read_upto = GetXLogReplayRecPtr(&ThisTimeLineID, false);
tli = ThisTimeLineID;
/*
diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c
index 75a95f3de7..11d8f1370c 100644
--- a/src/backend/postmaster/checkpointer.c
+++ b/src/backend/postmaster/checkpointer.c
@@ -437,7 +437,7 @@ CheckpointerMain(void)
*/
ckpt_active = true;
if (do_restartpoint)
- ckpt_start_recptr = GetXLogReplayRecPtr(NULL);
+ ckpt_start_recptr = GetXLogReplayRecPtr(NULL, false);
else
ckpt_start_recptr = GetInsertRecPtr();
ckpt_start_time = now;
@@ -801,7 +801,7 @@ IsCheckpointOnSchedule(double progress)
* value that was in effect when the WAL was generated).
*/
if (RecoveryInProgress())
- recptr = GetXLogReplayRecPtr(NULL);
+ recptr = GetXLogReplayRecPtr(NULL, false);
else
recptr = GetInsertRecPtr();
elapsed_xlogs = (((double) (recptr - ckpt_start_recptr)) /
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 453efc51e1..6684173a61 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -215,11 +215,31 @@ DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
* can restart from there.
*/
break;
+ case XLOG_PARAMETER_CHANGE:
+ {
+ xl_parameter_change *xlrec =
+ (xl_parameter_change *) XLogRecGetData(buf->record);
+
+ /*
+ * If wal_level on primary is reduced to less than logical, then we
+ * want to prevent existing logical slots from being used.
+ * Existing logical slots on standby get invalidated when this WAL
+ * record is replayed; and further, slot creation fails when the
+ * wal level is not sufficient; but all these operations are not
+ * synchronized, so a logical slot may creep in while the wal_level
+ * is being reduced. Hence this extra check.
+ */
+ if (xlrec->wal_level < WAL_LEVEL_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires "
+ "wal_level >= logical on master")));
+ break;
+ }
case XLOG_NOOP:
case XLOG_NEXTOID:
case XLOG_SWITCH:
case XLOG_BACKUP_END:
- case XLOG_PARAMETER_CHANGE:
case XLOG_RESTORE_POINT:
case XLOG_FPW_CHANGE:
case XLOG_FPI_FOR_HINT:
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index ffc6160e9f..3c46938972 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -119,23 +119,22 @@ CheckLogicalDecodingRequirements(void)
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("logical decoding requires a database connection")));
- /* ----
- * TODO: We got to change that someday soon...
- *
- * There's basically three things missing to allow this:
- * 1) We need to be able to correctly and quickly identify the timeline a
- * LSN belongs to
- * 2) We need to force hot_standby_feedback to be enabled at all times so
- * the primary cannot remove rows we need.
- * 3) support dropping replication slots referring to a database, in
- * dbase_redo. There can't be any active ones due to HS recovery
- * conflicts, so that should be relatively easy.
- * ----
- */
if (RecoveryInProgress())
- ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("logical decoding cannot be used while in recovery")));
+ {
+ /*
+ * This check may have race conditions, but whenever
+ * XLOG_PARAMETER_CHANGE indicates that wal_level has changed, we
+ * verify that there are no existing logical replication slots. And to
+ * avoid races around creating a new slot,
+ * CheckLogicalDecodingRequirements() is called once before creating
+ * the slot, and once when logical decoding is initially starting up.
+ */
+ if (GetActiveWalLevelOnStandby() < WAL_LEVEL_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires "
+ "wal_level >= logical on master")));
+ }
}
/*
@@ -330,6 +329,12 @@ CreateInitDecodingContext(const char *plugin,
LogicalDecodingContext *ctx;
MemoryContext old_context;
+ /*
+ * On standby, this check is also required while creating the slot. Check
+ * the comments in this function.
+ */
+ CheckLogicalDecodingRequirements();
+
/* shorter lines... */
slot = MyReplicationSlot;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 1d9400ea63..9069f3e50d 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -223,7 +223,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
if (!RecoveryInProgress())
end_of_wal = GetFlushRecPtr();
else
- end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID);
+ end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID, false);
ReplicationSlotAcquire(NameStr(*name), true);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index f9231229cc..4302d56539 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1087,37 +1087,28 @@ ReplicationSlotReserveWal(void)
/*
* For logical slots log a standby snapshot and start logical decoding
* at exactly that position. That allows the slot to start up more
- * quickly.
+ * quickly. But on a standby we cannot do WAL writes, so just use the
+ * replay pointer; effectively, an attempt to create a logical slot on
+ * standby will cause it to wait for an xl_running_xact record to be
+ * logged independently on the primary, so that a snapshot can be built
+ * using the record.
*
- * That's not needed (or indeed helpful) for physical slots as they'll
- * start replay at the last logged checkpoint anyway. Instead return
- * the location of the last redo LSN. While that slightly increases
- * the chance that we have to retry, it's where a base backup has to
- * start replay at.
+ * None of this is needed (or indeed helpful) for physical slots as
+ * they'll start replay at the last logged checkpoint anyway. Instead
+ * return the location of the last redo LSN. While that slightly
+ * increases the chance that we have to retry, it's where a base backup
+ * has to start replay at.
*/
- if (!RecoveryInProgress() && SlotIsLogical(slot))
- {
- XLogRecPtr flushptr;
-
- /* start at current insert position */
+ if (SlotIsPhysical(slot))
+ restart_lsn = GetRedoRecPtr();
+ else if (RecoveryInProgress())
+ restart_lsn = GetXLogReplayRecPtr(NULL, true);
+ else
restart_lsn = GetXLogInsertRecPtr();
- SpinLockAcquire(&slot->mutex);
- slot->data.restart_lsn = restart_lsn;
- SpinLockRelease(&slot->mutex);
-
- /* make sure we have enough information to start */
- flushptr = LogStandbySnapshot();
- /* and make sure it's fsynced to disk */
- XLogFlush(flushptr);
- }
- else
- {
- restart_lsn = GetRedoRecPtr();
- SpinLockAcquire(&slot->mutex);
- slot->data.restart_lsn = restart_lsn;
- SpinLockRelease(&slot->mutex);
- }
+ SpinLockAcquire(&slot->mutex);
+ slot->data.restart_lsn = restart_lsn;
+ SpinLockRelease(&slot->mutex);
/* prevent WAL removal as fast as possible */
ReplicationSlotsComputeRequiredLSN();
@@ -1133,6 +1124,17 @@ ReplicationSlotReserveWal(void)
if (XLogGetLastRemovedSegno() < segno)
break;
}
+
+ if (!RecoveryInProgress() && SlotIsLogical(slot))
+ {
+ XLogRecPtr flushptr;
+
+ /* make sure we have enough information to start */
+ flushptr = LogStandbySnapshot();
+
+ /* and make sure it's fsynced to disk */
+ XLogFlush(flushptr);
+ }
}
/*
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 31e74d3832..48d24442e2 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -636,7 +636,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
if (!RecoveryInProgress())
moveto = Min(moveto, GetFlushRecPtr());
else
- moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID));
+ moveto = Min(moveto, GetXLogReplayRecPtr(&ThisTimeLineID, false));
/* Acquire the slot so we "own" it */
ReplicationSlotAcquire(NameStr(*slotname), true);
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index faeea9f0cc..4bce8f192c 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -407,7 +407,7 @@ WalReceiverMain(void)
first_stream = false;
/* Initialize LogstreamResult and buffers for processing messages */
- LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
+ LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL, false);
initStringInfo(&reply_message);
initStringInfo(&incoming_message);
@@ -1075,7 +1075,7 @@ XLogWalRcvSendReply(bool force, bool requestReply)
/* Construct a new message */
writePtr = LogstreamResult.Write;
flushPtr = LogstreamResult.Flush;
- applyPtr = GetXLogReplayRecPtr(NULL);
+ applyPtr = GetXLogReplayRecPtr(NULL, false);
resetStringInfo(&reply_message);
pq_sendbyte(&reply_message, 'r');
diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c
index 6f0acbfdef..f264b71f73 100644
--- a/src/backend/replication/walreceiverfuncs.c
+++ b/src/backend/replication/walreceiverfuncs.c
@@ -372,7 +372,7 @@ GetReplicationApplyDelay(void)
receivePtr = walrcv->flushedUpto;
SpinLockRelease(&walrcv->mutex);
- replayPtr = GetXLogReplayRecPtr(NULL);
+ replayPtr = GetXLogReplayRecPtr(NULL, false);
if (receivePtr == replayPtr)
return 0;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 53e440c2ae..bc316cd28d 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1166,6 +1166,16 @@ StartLogicalReplication(StartReplicationCmd *cmd)
got_STOPPING = true;
}
+ /*
+ * In case of logical decoding on standby it may be that ThisTimeLineID
+ * is not set yet.
+ * Indeed we are not going through InitXLOGAccess on a Standby and
+ * it may also be that IdentifySystem has not been called yet.
+ * So let's get it through GetXLogReplayRecPtr().
+ */
+ if (ThisTimeLineID == 0)
+ (void) GetXLogReplayRecPtr(&ThisTimeLineID, false);
+
/*
* Create our decoding context, making it start at the previously ack'ed
* position.
@@ -1390,7 +1400,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr();
else
- RecentFlushPtr = GetXLogReplayRecPtr(NULL);
+ RecentFlushPtr = GetXLogReplayRecPtr(NULL, false);
for (;;)
{
@@ -1424,7 +1434,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr();
else
- RecentFlushPtr = GetXLogReplayRecPtr(NULL);
+ RecentFlushPtr = GetXLogReplayRecPtr(NULL, false);
/*
* If postmaster asked us to stop, don't wait anymore.
@@ -2890,10 +2900,12 @@ XLogSendLogical(void)
* If first time through in this session, initialize flushPtr. Otherwise,
* we only need to update flushPtr if EndRecPtr is past it.
*/
- if (flushPtr == InvalidXLogRecPtr)
- flushPtr = GetFlushRecPtr();
- else if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
- flushPtr = GetFlushRecPtr();
+ if (flushPtr == InvalidXLogRecPtr ||
+ logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
+ {
+ flushPtr = (am_cascading_walsender ?
+ GetStandbyFlushRecPtr() : GetFlushRecPtr());
+ }
/* If EndRecPtr is still past our flushPtr, it means we caught up. */
if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
@@ -2982,7 +2994,7 @@ GetStandbyFlushRecPtr(void)
*/
receivePtr = GetWalRcvFlushRecPtr(NULL, &receiveTLI);
- replayPtr = GetXLogReplayRecPtr(&replayTLI);
+ replayPtr = GetXLogReplayRecPtr(&replayTLI, false);
ThisTimeLineID = replayTLI;
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 77187c12be..c7fb52240e 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -316,7 +316,7 @@ extern bool HotStandbyActive(void);
extern bool HotStandbyActiveInReplay(void);
extern bool XLogInsertAllowed(void);
extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
-extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI);
+extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI, bool avoid_header);
extern XLogRecPtr GetXLogInsertRecPtr(void);
extern XLogRecPtr GetXLogWriteRecPtr(void);
extern RecoveryPauseState GetRecoveryPauseState(void);
@@ -333,6 +333,7 @@ extern Size XLOGShmemSize(void);
extern void XLOGShmemInit(void);
extern void BootStrapXLOG(void);
extern void LocalProcessControlFile(bool reset);
+extern WalLevel GetActiveWalLevelOnStandby(void);
extern void StartupXLOG(void);
extern void ShutdownXLOG(int code, Datum arg);
extern void InitXLOGAccess(void);
--
2.18.4
[text/plain] v19-0002-Handle-logical-slot-conflicts-on-standby.patch (29.3K, ../[email protected]/5-v19-0002-Handle-logical-slot-conflicts-on-standby.patch)
download | inline diff:
From 6142c34b7a30f468aac56f24a41e3275dae0540e Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:12:16 +0000
Subject: [PATCH v19 2/5] Handle logical slot conflicts on standby.
During WAL replay on standby, when slot conflict is identified,
invalidate such slots. Also do the same thing if wal_level on master
is reduced to below logical and there are existing logical slots
on standby. Introduce a new ProcSignalReason value for slot
conflict recovery. Arrange for a new pg_stat_get_activity field:
confl_logicalslot.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
doc/src/sgml/monitoring.sgml | 10 +
src/backend/access/gist/gistxlog.c | 4 +-
src/backend/access/hash/hash_xlog.c | 3 +-
src/backend/access/heap/heapam.c | 10 +-
src/backend/access/nbtree/nbtxlog.c | 4 +-
src/backend/access/spgist/spgxlog.c | 1 +
src/backend/access/transam/xlog.c | 13 ++
src/backend/catalog/system_views.sql | 3 +-
src/backend/postmaster/pgstat.c | 20 ++
.../replication/logical/logicalfuncs.c | 7 +-
src/backend/replication/slot.c | 209 ++++++++++++++++++
src/backend/replication/walsender.c | 8 +
src/backend/storage/ipc/procarray.c | 4 +
src/backend/storage/ipc/procsignal.c | 3 +
src/backend/storage/ipc/standby.c | 11 +-
src/backend/tcop/postgres.c | 22 ++
src/backend/utils/adt/pgstatfuncs.c | 16 ++
src/include/catalog/pg_proc.dat | 5 +
src/include/pgstat.h | 2 +
src/include/replication/slot.h | 3 +
src/include/storage/procsignal.h | 1 +
src/include/storage/standby.h | 4 +-
src/test/regress/expected/rules.out | 3 +-
23 files changed, 351 insertions(+), 15 deletions(-)
4.0% src/backend/access/heap/
4.7% src/backend/access/transam/
5.1% src/backend/access/
4.3% src/backend/postmaster/
3.5% src/backend/replication/logical/
52.3% src/backend/replication/
5.6% src/backend/storage/ipc/
7.1% src/backend/tcop/
3.5% src/backend/
6.3% src/include/
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index dcbb10fb6f..e9e1efa1b8 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -4029,6 +4029,16 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
deadlocks
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>confl_logicalslot</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of queries in this database that have been canceled due to
+ logical slots
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 46aee6f2a9..5963e639d8 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -195,7 +195,8 @@ gistRedoDeleteRecord(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid,
+ xldata->onCatalogTable, rnode);
}
if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
@@ -395,6 +396,7 @@ gistRedoPageReuse(XLogReaderState *record)
*/
if (InHotStandby)
ResolveRecoveryConflictWithSnapshotFullXid(xlrec->latestRemovedFullXid,
+ xlrec->onCatalogTable,
xlrec->node);
}
diff --git a/src/backend/access/hash/hash_xlog.c b/src/backend/access/hash/hash_xlog.c
index af35a991fc..f91d4c36e2 100644
--- a/src/backend/access/hash/hash_xlog.c
+++ b/src/backend/access/hash/hash_xlog.c
@@ -1002,7 +1002,8 @@ hash_xlog_vacuum_one_page(XLogReaderState *record)
RelFileNode rnode;
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xldata->latestRemovedXid,
+ xldata->onCatalogTable, rnode);
}
action = XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &buffer);
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 4019a2122e..75ca5f79f4 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -8419,7 +8419,8 @@ heap_xlog_prune(XLogReaderState *record)
* no queries running for which the removed tuples are still visible.
*/
if (InHotStandby)
- ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
/*
* If we have a full-page image, restore it (using a cleanup lock) and
@@ -8587,7 +8588,9 @@ heap_xlog_visible(XLogReaderState *record)
* rather than killing the transaction outright.
*/
if (InHotStandby)
- ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->cutoff_xid,
+ xlrec->onCatalogTable,
+ rnode);
/*
* Read the heap page, if it still exists. If the heap file has dropped or
@@ -8724,7 +8727,8 @@ heap_xlog_freeze_page(XLogReaderState *record)
TransactionIdRetreat(latestRemovedXid);
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
}
if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
diff --git a/src/backend/access/nbtree/nbtxlog.c b/src/backend/access/nbtree/nbtxlog.c
index c2e920f159..004e8b7dd8 100644
--- a/src/backend/access/nbtree/nbtxlog.c
+++ b/src/backend/access/nbtree/nbtxlog.c
@@ -669,7 +669,8 @@ btree_xlog_delete(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &rnode, NULL, NULL);
- ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid, rnode);
+ ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
+ xlrec->onCatalogTable, rnode);
}
/*
@@ -1007,6 +1008,7 @@ btree_xlog_reuse_page(XLogReaderState *record)
if (InHotStandby)
ResolveRecoveryConflictWithSnapshotFullXid(xlrec->latestRemovedFullXid,
+ xlrec->onCatalogTable,
xlrec->node);
}
diff --git a/src/backend/access/spgist/spgxlog.c b/src/backend/access/spgist/spgxlog.c
index 3dfd2aa317..add4da4e74 100644
--- a/src/backend/access/spgist/spgxlog.c
+++ b/src/backend/access/spgist/spgxlog.c
@@ -881,6 +881,7 @@ spgRedoVacuumRedirect(XLogReaderState *record)
XLogRecGetBlockTag(record, 0, &node, NULL, NULL);
ResolveRecoveryConflictWithSnapshot(xldata->newestRedirectXid,
+ xldata->onCatalogTable,
node);
}
}
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 1b3a3d9bea..e863caf3b2 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -10358,6 +10358,19 @@ xlog_redo(XLogReaderState *record)
/* Update our copy of the parameters in pg_control */
memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_parameter_change));
+ /*
+ * Invalidate logical slots if we are in hot standby and the primary does not
+ * have a WAL level sufficient for logical decoding. No need to search
+ * for potentially conflicting logically slots if standby is running
+ * with wal_level lower than logical, because in that case, we would
+ * have either disallowed creation of logical slots or invalidated existing
+ * ones.
+ */
+ if (InRecovery && InHotStandby &&
+ xlrec.wal_level < WAL_LEVEL_LOGICAL &&
+ wal_level >= WAL_LEVEL_LOGICAL)
+ InvalidateConflictingLogicalReplicationSlots(InvalidOid,InvalidTransactionId);
+
LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
ControlFile->MaxConnections = xlrec.MaxConnections;
ControlFile->max_worker_processes = xlrec.max_worker_processes;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 999d984068..e4b7830a1e 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1020,7 +1020,8 @@ CREATE VIEW pg_stat_database_conflicts AS
pg_stat_get_db_conflict_lock(D.oid) AS confl_lock,
pg_stat_get_db_conflict_snapshot(D.oid) AS confl_snapshot,
pg_stat_get_db_conflict_bufferpin(D.oid) AS confl_bufferpin,
- pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock
+ pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock,
+ pg_stat_get_db_conflict_logicalslot(D.oid) AS confl_logicalslot
FROM pg_database D;
CREATE VIEW pg_stat_user_functions AS
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index b0d07c0e0b..38fa5f4809 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -1863,6 +1863,22 @@ pgstat_report_replslot_drop(const char *slotname)
pgstat_send(&msg, sizeof(PgStat_MsgReplSlot));
}
+/* ----------
+ * pgstat_report_replslot_conflict()
+ * Tell the collector about a logical slot being conflicting
+ * with recovery.
+ * ----------
+ */
+void
+pgstat_report_replslot_conflict(Oid dboid)
+{
+ PgStat_MsgRecoveryConflict msg;
+ pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_RECOVERYCONFLICT);
+ msg.m_databaseid = dboid;
+ msg.m_reason = PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT;
+ pgstat_send(&msg, sizeof(msg));
+}
+
/* ----------
* pgstat_ping() -
*
@@ -3489,6 +3505,7 @@ reset_dbentry_counters(PgStat_StatDBEntry *dbentry)
dbentry->n_conflict_tablespace = 0;
dbentry->n_conflict_lock = 0;
dbentry->n_conflict_snapshot = 0;
+ dbentry->n_conflict_logicalslot = 0;
dbentry->n_conflict_bufferpin = 0;
dbentry->n_conflict_startup_deadlock = 0;
dbentry->n_temp_files = 0;
@@ -5425,6 +5442,9 @@ pgstat_recv_recoveryconflict(PgStat_MsgRecoveryConflict *msg, int len)
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
dbentry->n_conflict_snapshot++;
break;
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ dbentry->n_conflict_logicalslot++;
+ break;
case PROCSIG_RECOVERY_CONFLICT_BUFFERPIN:
dbentry->n_conflict_bufferpin++;
break;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 1f38c5b33e..1d9400ea63 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -241,11 +241,14 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
/*
* After the sanity checks in CreateDecodingContext, make sure the
- * restart_lsn is valid. Avoid "cannot get changes" wording in this
+ * restart_lsn is valid or both xmin and catalog_xmin are valid.
+ * Avoid "cannot get changes" wording in this
* errmsg because that'd be confusingly ambiguous about no changes
* being available.
*/
- if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
+ if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn)
+ || (!TransactionIdIsValid(MyReplicationSlot->data.xmin)
+ && !TransactionIdIsValid(MyReplicationSlot->data.catalog_xmin)))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("can no longer get changes from replication slot \"%s\"",
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 8c18b4ed05..f9231229cc 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1317,6 +1317,215 @@ restart:
LWLockRelease(ReplicationSlotControlLock);
}
+/*
+ * Helper for InvalidateConflictingLogicalReplicationSlot -- acquires the given slot
+ * and mark it invalid, if necessary and possible.
+ *
+ * Returns whether ReplicationSlotControlLock was released in the interim (and
+ * in that case we're not holding the lock at return, otherwise we are).
+ *
+ * This is inherently racy, because we release the LWLock
+ * for syscalls, so caller must restart if we return true.
+ */
+static bool
+InvalidatePossiblyConflictingLogicalReplicationSlot(ReplicationSlot *s, TransactionId xid)
+{
+ int last_signaled_pid = 0;
+ bool released_lock = false;
+
+ for (;;)
+ {
+ TransactionId slot_xmin;
+ TransactionId slot_catalog_xmin;
+ NameData slotname;
+ int active_pid = 0;
+
+ Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ if (!s->in_use)
+ {
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+ break;
+ }
+
+ /*
+ * Check if the slot needs to be invalidated. If it needs to be
+ * invalidated, and is not currently acquired, acquire it and mark it
+ * as having been invalidated. We do this with the spinlock held to
+ * avoid race conditions -- for example the xmin(s) could move forward
+ * , or the slot could be dropped.
+ */
+ SpinLockAcquire(&s->mutex);
+
+ slot_xmin = s->data.xmin;
+ slot_catalog_xmin = s->data.catalog_xmin;
+
+ /*
+ * If the slot is already invalid or is not conflicting, we don't need to
+ * do anything.
+ */
+
+ /* slot has been invalidated */
+ if ((!TransactionIdIsValid(slot_xmin) && !TransactionIdIsValid(slot_catalog_xmin))
+ ||
+ /*
+ * we are not forcing for invalidation because the xid is valid
+ * and this is a non conflicting slot
+ */
+ (TransactionIdIsValid(xid) && !(
+ (TransactionIdIsValid(slot_xmin) && TransactionIdPrecedesOrEquals(slot_xmin, xid))
+ ||
+ (TransactionIdIsValid(slot_catalog_xmin) && TransactionIdPrecedesOrEquals(slot_catalog_xmin, xid))
+ ))
+ )
+ {
+ SpinLockRelease(&s->mutex);
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+ break;
+ }
+
+ slotname = s->data.name;
+ active_pid = s->active_pid;
+
+ /*
+ * If the slot can be acquired, do so and mark it invalidated
+ * immediately. Otherwise we'll signal the owning process, below, and
+ * retry.
+ */
+ if (active_pid == 0)
+ {
+ MyReplicationSlot = s;
+ s->active_pid = MyProcPid;
+ s->data.xmin = InvalidTransactionId;
+ s->data.catalog_xmin = InvalidTransactionId;
+ }
+
+ SpinLockRelease(&s->mutex);
+
+ if (active_pid != 0)
+ {
+ /*
+ * Prepare the sleep on the slot's condition variable before
+ * releasing the lock, to close a possible race condition if the
+ * slot is released before the sleep below.
+ */
+
+ ConditionVariablePrepareToSleep(&s->active_cv);
+
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ /*
+ * Signal to terminate the process that owns the slot, if we
+ * haven't already signalled it. (Avoidance of repeated
+ * signalling is the only reason for there to be a loop in this
+ * routine; otherwise we could rely on caller's restart loop.)
+ *
+ * There is the race condition that other process may own the slot
+ * after its current owner process is terminated and before this
+ * process owns it. To handle that, we signal only if the PID of
+ * the owning process has changed from the previous time. (This
+ * logic assumes that the same PID is not reused very quickly.)
+ */
+ if (last_signaled_pid != active_pid)
+ {
+ ereport(LOG,
+ (errmsg("terminating process %d because replication slot \"%s\" conflicts with recovery",
+ active_pid, NameStr(slotname))));
+
+ (void) kill(active_pid, SIGTERM);
+ last_signaled_pid = active_pid;
+ }
+
+ /* Wait until the slot is released. */
+ ConditionVariableSleep(&s->active_cv,
+ WAIT_EVENT_REPLICATION_SLOT_DROP);
+
+ /*
+ * Re-acquire lock and start over; we expect to invalidate the
+ * slot next time (unless another process acquires the slot in the
+ * meantime).
+ */
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ continue;
+ }
+ else
+ {
+ /*
+ * We hold the slot now and have already invalidated it; flush it
+ * to ensure that state persists.
+ *
+ * Don't want to hold ReplicationSlotControlLock across file
+ * system operations, so release it now but be sure to tell caller
+ * to restart from scratch.
+ */
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ /* Make sure the invalidated state persists across server restart */
+ ReplicationSlotMarkDirty();
+ ReplicationSlotSave();
+ ReplicationSlotRelease();
+ pgstat_report_replslot_conflict(s->data.database);
+
+ ereport(LOG,
+ (errmsg("invalidating slot \"%s\" because it conflicts with recovery", NameStr(slotname))));
+
+ /* done with this slot for now */
+ break;
+ }
+ }
+
+ Assert(!released_lock == LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ return released_lock;
+}
+
+/*
+ * Resolve recovery conflicts with logical slots.
+ *
+ * When xid is valid, it means that we are about to remove rows older than xid.
+ * Therefore we need to invalidate slots that depend on seeing those rows.
+ * When xid is invalid, invalidate all logical slots. This is required when the
+ * master wal_level is set back to replica, so existing logical slots need to
+ * be invalidated.
+ */
+void
+InvalidateConflictingLogicalReplicationSlots(Oid dboid, TransactionId xid)
+{
+
+ Assert(max_replication_slots >= 0);
+
+ if (max_replication_slots == 0)
+ return;
+restart:
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ for (int i = 0; i < max_replication_slots; i++)
+ {
+ ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+ if (!s->in_use)
+ continue;
+
+ /* We are only dealing with *logical* slot conflicts. */
+ if (!SlotIsLogical(s))
+ continue;
+
+ /* not our database and we don't want all the database, skip */
+ if ((s->data.database != InvalidOid && s->data.database != dboid) && TransactionIdIsValid(xid))
+ continue;
+
+ if (InvalidatePossiblyConflictingLogicalReplicationSlot(s, xid))
+ {
+ /* if the lock was released, we need to restart from scratch */
+ goto restart;
+ }
+ }
+ LWLockRelease(ReplicationSlotControlLock);
+}
+
/*
* Flush all replication slots to disk.
*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3224536356..53e440c2ae 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1139,6 +1139,14 @@ StartLogicalReplication(StartReplicationCmd *cmd)
ReplicationSlotAcquire(cmd->slotname, true);
+ if (!TransactionIdIsValid(MyReplicationSlot->data.xmin)
+ && !TransactionIdIsValid(MyReplicationSlot->data.catalog_xmin))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot read from logical replication slot \"%s\"",
+ cmd->slotname),
+ errdetail("This slot has been invalidated because it was conflicting with recovery.")));
+
if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index e4c008e443..bdc9ba1fc0 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -3401,6 +3401,10 @@ SignalVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode,
GET_VXID_FROM_PGPROC(procvxid, *proc);
+ /*
+ * Note: vxid.localTransactionId can be invalid, which means the
+ * request is to signal the pid that is not running a transaction.
+ */
if (procvxid.backendId == vxid.backendId &&
procvxid.localTransactionId == vxid.localTransactionId)
{
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index defb75aa26..315d5a1e33 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -673,6 +673,9 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT);
+ if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT))
+ RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT);
+
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK);
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 553b6e5460..b67e79c55a 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -23,6 +23,7 @@
#include "access/xloginsert.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "replication/slot.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
@@ -440,7 +441,8 @@ ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlist,
}
void
-ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode node)
+ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid,
+ bool onCatalogTable, RelFileNode node)
{
VirtualTransactionId *backends;
@@ -465,6 +467,9 @@ ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode
PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
WAIT_EVENT_RECOVERY_CONFLICT_SNAPSHOT,
true);
+
+ if (onCatalogTable)
+ InvalidateConflictingLogicalReplicationSlots(node.dbNode, latestRemovedXid);
}
/*
@@ -473,7 +478,7 @@ ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode
*/
void
ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXid,
- RelFileNode node)
+ bool onCatalogTable, RelFileNode node)
{
/*
* ResolveRecoveryConflictWithSnapshot operates on 32-bit TransactionIds,
@@ -491,7 +496,7 @@ ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXi
TransactionId latestRemovedXid;
latestRemovedXid = XidFromFullTransactionId(latestRemovedFullXid);
- ResolveRecoveryConflictWithSnapshot(latestRemovedXid, node);
+ ResolveRecoveryConflictWithSnapshot(latestRemovedXid, onCatalogTable, node);
}
}
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 8cea10c901..477c96df9d 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -2467,6 +2467,9 @@ errdetail_recovery_conflict(void)
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
errdetail("User query might have needed to see row versions that must be removed.");
break;
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ errdetail("User was using the logical slot that must be dropped.");
+ break;
case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK:
errdetail("User transaction caused buffer deadlock with recovery.");
break;
@@ -3037,6 +3040,25 @@ RecoveryConflictInterrupt(ProcSignalReason reason)
case PROCSIG_RECOVERY_CONFLICT_LOCK:
case PROCSIG_RECOVERY_CONFLICT_TABLESPACE:
case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
+ case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
+ /*
+ * For conflicts that require a logical slot to be invalidated, the
+ * requirement is for the signal receiver to release the slot,
+ * so that it could be invalidated by the signal sender. So for
+ * normal backends, the transaction should be aborted, just
+ * like for other recovery conflicts. But if it's walsender on
+ * standby, then it has to be killed so as to release an
+ * acquired logical slot.
+ */
+ if (am_cascading_walsender &&
+ reason == PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT &&
+ MyReplicationSlot && SlotIsLogical(MyReplicationSlot))
+ {
+ RecoveryConflictPending = true;
+ QueryCancelPending = true;
+ InterruptPending = true;
+ break;
+ }
/*
* If we aren't in a transaction any longer then ignore.
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 14056f5347..e64d2666d0 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -1500,6 +1500,21 @@ pg_stat_get_db_conflict_snapshot(PG_FUNCTION_ARGS)
PG_RETURN_INT64(result);
}
+Datum
+pg_stat_get_db_conflict_logicalslot(PG_FUNCTION_ARGS)
+{
+ Oid dbid = PG_GETARG_OID(0);
+ int64 result;
+ PgStat_StatDBEntry *dbentry;
+
+ if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (dbentry->n_conflict_logicalslot);
+
+ PG_RETURN_INT64(result);
+}
+
Datum
pg_stat_get_db_conflict_bufferpin(PG_FUNCTION_ARGS)
{
@@ -1543,6 +1558,7 @@ pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS)
result = (int64) (dbentry->n_conflict_tablespace +
dbentry->n_conflict_lock +
dbentry->n_conflict_snapshot +
+ dbentry->n_conflict_logicalslot +
dbentry->n_conflict_bufferpin +
dbentry->n_conflict_startup_deadlock);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index fde251fa4f..52a6366042 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5434,6 +5434,11 @@
proname => 'pg_stat_get_db_conflict_snapshot', provolatile => 's',
proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
prosrc => 'pg_stat_get_db_conflict_snapshot' },
+{ oid => '4544',
+ descr => 'statistics: recovery conflicts in database caused by logical replication slot',
+ proname => 'pg_stat_get_db_conflict_logicalslot', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
+ prosrc => 'pg_stat_get_db_conflict_logicalslot' },
{ oid => '3068',
descr => 'statistics: recovery conflicts in database caused by shared buffer pin',
proname => 'pg_stat_get_db_conflict_bufferpin', provolatile => 's',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 9612c0a6c2..52623b4be9 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -743,6 +743,7 @@ typedef struct PgStat_StatDBEntry
PgStat_Counter n_conflict_tablespace;
PgStat_Counter n_conflict_lock;
PgStat_Counter n_conflict_snapshot;
+ PgStat_Counter n_conflict_logicalslot;
PgStat_Counter n_conflict_bufferpin;
PgStat_Counter n_conflict_startup_deadlock;
PgStat_Counter n_temp_files;
@@ -1011,6 +1012,7 @@ extern void pgstat_report_checksum_failure(void);
extern void pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat);
extern void pgstat_report_replslot_create(const char *slotname);
extern void pgstat_report_replslot_drop(const char *slotname);
+extern void pgstat_report_replslot_conflict(Oid dbOid);
extern void pgstat_initialize(void);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 2eb7e3a530..bd292e5e14 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -215,6 +215,7 @@ extern XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void);
extern bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive);
extern void ReplicationSlotsDropDBSlots(Oid dboid);
extern void InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno);
+extern void InvalidateConflictingLogicalReplicationSlots(Oid dboid, TransactionId xid);
extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, int szslot);
extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
@@ -224,4 +225,6 @@ extern void CheckPointReplicationSlots(void);
extern void CheckSlotRequirements(void);
+extern void ResolveRecoveryConflictWithLogicalSlots(Oid dboid, TransactionId xid, char *reason);
+
#endif /* SLOT_H */
diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h
index eec186be2e..f86b070dbc 100644
--- a/src/include/storage/procsignal.h
+++ b/src/include/storage/procsignal.h
@@ -41,6 +41,7 @@ typedef enum
PROCSIG_RECOVERY_CONFLICT_TABLESPACE,
PROCSIG_RECOVERY_CONFLICT_LOCK,
PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
+ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT,
PROCSIG_RECOVERY_CONFLICT_BUFFERPIN,
PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK,
diff --git a/src/include/storage/standby.h b/src/include/storage/standby.h
index 38fd85a431..3ba1882216 100644
--- a/src/include/storage/standby.h
+++ b/src/include/storage/standby.h
@@ -30,9 +30,9 @@ extern void InitRecoveryTransactionEnvironment(void);
extern void ShutdownRecoveryTransactionEnvironment(void);
extern void ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid,
- RelFileNode node);
+ bool onCatalogTable, RelFileNode node);
extern void ResolveRecoveryConflictWithSnapshotFullXid(FullTransactionId latestRemovedFullXid,
- RelFileNode node);
+ bool onCatalogTable, RelFileNode node);
extern void ResolveRecoveryConflictWithTablespace(Oid tsid);
extern void ResolveRecoveryConflictWithDatabase(Oid dbid);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index e5ab11275d..66f28b649d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1871,7 +1871,8 @@ pg_stat_database_conflicts| SELECT d.oid AS datid,
pg_stat_get_db_conflict_lock(d.oid) AS confl_lock,
pg_stat_get_db_conflict_snapshot(d.oid) AS confl_snapshot,
pg_stat_get_db_conflict_bufferpin(d.oid) AS confl_bufferpin,
- pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock
+ pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock,
+ pg_stat_get_db_conflict_logicalslot(d.oid) AS confl_logicalslot
FROM pg_database d;
pg_stat_gssapi| SELECT s.pid,
s.gss_auth AS gss_authenticated,
--
2.18.4
[text/plain] v19-0001-Add-info-in-WAL-records-in-preparation-for-logic.patch (18.0K, ../[email protected]/6-v19-0001-Add-info-in-WAL-records-in-preparation-for-logic.patch)
download | inline diff:
From e685cf4ef639dbc92992b66d2a33010b4a2cde6f Mon Sep 17 00:00:00 2001
From: bdrouvotAWS <[email protected]>
Date: Tue, 22 Jun 2021 08:04:15 +0000
Subject: [PATCH v19 1/5] Add info in WAL records in preparation for logical
slot conflict handling.
When a WAL replay on standby indicates that a catalog table tuple is
to be deleted by an xid that is greater than a logical slot's
catalog_xmin, then that means the slot's catalog_xmin conflicts with
the xid, and we need to handle the conflict. While subsequent commits
will do the actual conflict handling, this commit adds a new field
onCatalogTable in such WAL records, that is true for catalog tables,
so as to arrange for conflict handling.
Author: Andres Freund (in an older version), Amit Khandekar, Bertrand Drouvot
Reviewed-By: Bertrand Drouvot, Andres Freund, Robert Haas, Fabrizio de Royes Mello
---
src/backend/access/gist/gist.c | 2 +-
src/backend/access/gist/gistbuild.c | 2 +-
src/backend/access/gist/gistutil.c | 4 ++--
src/backend/access/gist/gistxlog.c | 4 +++-
src/backend/access/hash/hashinsert.c | 1 +
src/backend/access/heap/heapam.c | 4 +++-
src/backend/access/heap/pruneheap.c | 1 +
src/backend/access/heap/visibilitymap.c | 2 +-
src/backend/access/nbtree/nbtpage.c | 12 +++++++++---
src/backend/access/spgist/spgvacuum.c | 8 ++++++++
src/backend/utils/cache/lsyscache.c | 15 +++++++++++++++
src/include/access/gist_private.h | 6 +++---
src/include/access/gistxlog.h | 3 ++-
src/include/access/hash_xlog.h | 1 +
src/include/access/heapam_xlog.h | 5 ++++-
src/include/access/nbtxlog.h | 2 ++
src/include/access/spgxlog.h | 1 +
src/include/utils/lsyscache.h | 1 +
src/include/utils/rel.h | 9 +++++++++
19 files changed, 68 insertions(+), 15 deletions(-)
17.9% src/backend/access/gist/
13.1% src/backend/access/heap/
14.5% src/backend/access/nbtree/
8.9% src/backend/access/spgist/
7.2% src/backend/utils/cache/
19.6% src/include/access/
16.6% src/include/utils/
diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c
index 0683f42c25..b6e6340c3c 100644
--- a/src/backend/access/gist/gist.c
+++ b/src/backend/access/gist/gist.c
@@ -348,7 +348,7 @@ gistplacetopage(Relation rel, Size freespace, GISTSTATE *giststate,
for (; ptr; ptr = ptr->next)
{
/* Allocate new page */
- ptr->buffer = gistNewBuffer(rel);
+ ptr->buffer = gistNewBuffer(heapRel, rel);
GISTInitBuffer(ptr->buffer, (is_leaf) ? F_LEAF : 0);
ptr->page = BufferGetPage(ptr->buffer);
ptr->block.blkno = BufferGetBlockNumber(ptr->buffer);
diff --git a/src/backend/access/gist/gistbuild.c b/src/backend/access/gist/gistbuild.c
index f46a42197c..80949bd4db 100644
--- a/src/backend/access/gist/gistbuild.c
+++ b/src/backend/access/gist/gistbuild.c
@@ -290,7 +290,7 @@ gistbuild(Relation heap, Relation index, IndexInfo *indexInfo)
Page page;
/* initialize the root page */
- buffer = gistNewBuffer(index);
+ buffer = gistNewBuffer(heap, index);
Assert(BufferGetBlockNumber(buffer) == GIST_ROOT_BLKNO);
page = BufferGetPage(buffer);
diff --git a/src/backend/access/gist/gistutil.c b/src/backend/access/gist/gistutil.c
index 8dcd53c457..8ba6178a5f 100644
--- a/src/backend/access/gist/gistutil.c
+++ b/src/backend/access/gist/gistutil.c
@@ -821,7 +821,7 @@ gistcheckpage(Relation rel, Buffer buf)
* Caller is responsible for initializing the page by calling GISTInitBuffer
*/
Buffer
-gistNewBuffer(Relation r)
+gistNewBuffer(Relation heapRel, Relation r)
{
Buffer buffer;
bool needLock;
@@ -865,7 +865,7 @@ gistNewBuffer(Relation r)
* page's deleteXid.
*/
if (XLogStandbyInfoActive() && RelationNeedsWAL(r))
- gistXLogPageReuse(r, blkno, GistPageGetDeleteXid(page));
+ gistXLogPageReuse(heapRel, r, blkno, GistPageGetDeleteXid(page));
return buffer;
}
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 6464cb9281..46aee6f2a9 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -596,7 +596,8 @@ gistXLogAssignLSN(void)
* Write XLOG record about reuse of a deleted page.
*/
void
-gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemovedXid)
+gistXLogPageReuse(Relation heapRel, Relation rel,
+ BlockNumber blkno, FullTransactionId latestRemovedXid)
{
gistxlogPageReuse xlrec_reuse;
@@ -607,6 +608,7 @@ gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemov
*/
/* XLOG stuff */
+ xlrec_reuse.onCatalogTable = RelationIsAccessibleInLogicalDecoding(heapRel);
xlrec_reuse.node = rel->rd_node;
xlrec_reuse.block = blkno;
xlrec_reuse.latestRemovedFullXid = latestRemovedXid;
diff --git a/src/backend/access/hash/hashinsert.c b/src/backend/access/hash/hashinsert.c
index d254a00b6a..ce223b2c19 100644
--- a/src/backend/access/hash/hashinsert.c
+++ b/src/backend/access/hash/hashinsert.c
@@ -398,6 +398,7 @@ _hash_vacuum_one_page(Relation rel, Relation hrel, Buffer metabuf, Buffer buf)
xl_hash_vacuum_one_page xlrec;
XLogRecPtr recptr;
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(hrel);
xlrec.latestRemovedXid = latestRemovedXid;
xlrec.ntuples = ndeletable;
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 2433998f39..4019a2122e 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -7941,6 +7941,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid,
/* nor when there are no tuples to freeze */
Assert(ntuples > 0);
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(reln);
xlrec.cutoff_xid = cutoff_xid;
xlrec.ntuples = ntuples;
@@ -7971,7 +7972,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid,
* heap_buffer, if necessary.
*/
XLogRecPtr
-log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
+log_heap_visible(Relation rel, Buffer heap_buffer, Buffer vm_buffer,
TransactionId cutoff_xid, uint8 vmflags)
{
xl_heap_visible xlrec;
@@ -7981,6 +7982,7 @@ log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
Assert(BufferIsValid(heap_buffer));
Assert(BufferIsValid(vm_buffer));
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(rel);
xlrec.cutoff_xid = cutoff_xid;
xlrec.flags = vmflags;
XLogBeginInsert();
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index 15ca1b304a..0590b7053c 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -323,6 +323,7 @@ heap_page_prune(Relation relation, Buffer buffer,
xlrec.latestRemovedXid = prstate.latestRemovedXid;
xlrec.nredirected = prstate.nredirected;
xlrec.ndead = prstate.ndead;
+ xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(relation);
XLogBeginInsert();
XLogRegisterData((char *) &xlrec, SizeOfHeapPrune);
diff --git a/src/backend/access/heap/visibilitymap.c b/src/backend/access/heap/visibilitymap.c
index e198df65d8..6e89a08c52 100644
--- a/src/backend/access/heap/visibilitymap.c
+++ b/src/backend/access/heap/visibilitymap.c
@@ -282,7 +282,7 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
if (XLogRecPtrIsInvalid(recptr))
{
Assert(!InRecovery);
- recptr = log_heap_visible(rel->rd_node, heapBuf, vmBuf,
+ recptr = log_heap_visible(rel, heapBuf, vmBuf,
cutoff_xid, flags);
/*
diff --git a/src/backend/access/nbtree/nbtpage.c b/src/backend/access/nbtree/nbtpage.c
index ebec8fa5b8..2d27a3f974 100644
--- a/src/backend/access/nbtree/nbtpage.c
+++ b/src/backend/access/nbtree/nbtpage.c
@@ -36,6 +36,7 @@
#include "utils/memdebug.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
+#include "utils/lsyscache.h"
static BTMetaPageData *_bt_getmeta(Relation rel, Buffer metabuf);
static void _bt_log_reuse_page(Relation rel, BlockNumber blkno,
@@ -43,7 +44,8 @@ static void _bt_log_reuse_page(Relation rel, BlockNumber blkno,
static void _bt_delitems_delete(Relation rel, Buffer buf,
TransactionId latestRemovedXid,
OffsetNumber *deletable, int ndeletable,
- BTVacuumPosting *updatable, int nupdatable);
+ BTVacuumPosting *updatable, int nupdatable,
+ Relation heapRel);
static char *_bt_delitems_update(BTVacuumPosting *updatable, int nupdatable,
OffsetNumber *updatedoffsets,
Size *updatedbuflen, bool needswal);
@@ -836,6 +838,7 @@ _bt_log_reuse_page(Relation rel, BlockNumber blkno, FullTransactionId safexid)
*/
/* XLOG stuff */
+ xlrec_reuse.onCatalogTable = get_rel_logical_catalog(rel->rd_index->indrelid);
xlrec_reuse.node = rel->rd_node;
xlrec_reuse.block = blkno;
xlrec_reuse.latestRemovedFullXid = safexid;
@@ -1296,7 +1299,8 @@ _bt_delitems_vacuum(Relation rel, Buffer buf,
static void
_bt_delitems_delete(Relation rel, Buffer buf, TransactionId latestRemovedXid,
OffsetNumber *deletable, int ndeletable,
- BTVacuumPosting *updatable, int nupdatable)
+ BTVacuumPosting *updatable, int nupdatable,
+ Relation heapRel)
{
Page page = BufferGetPage(buf);
BTPageOpaque opaque;
@@ -1358,6 +1362,8 @@ _bt_delitems_delete(Relation rel, Buffer buf, TransactionId latestRemovedXid,
XLogRecPtr recptr;
xl_btree_delete xlrec_delete;
+ xlrec_delete.onCatalogTable =
+ RelationIsAccessibleInLogicalDecoding(heapRel);
xlrec_delete.latestRemovedXid = latestRemovedXid;
xlrec_delete.ndeleted = ndeletable;
xlrec_delete.nupdated = nupdatable;
@@ -1685,7 +1691,7 @@ _bt_delitems_delete_check(Relation rel, Buffer buf, Relation heapRel,
/* Physically delete tuples (or TIDs) using deletable (or updatable) */
_bt_delitems_delete(rel, buf, latestRemovedXid, deletable, ndeletable,
- updatable, nupdatable);
+ updatable, nupdatable, heapRel);
/* be tidy */
for (int i = 0; i < nupdatable; i++)
diff --git a/src/backend/access/spgist/spgvacuum.c b/src/backend/access/spgist/spgvacuum.c
index 76fb0374c4..3186885d14 100644
--- a/src/backend/access/spgist/spgvacuum.c
+++ b/src/backend/access/spgist/spgvacuum.c
@@ -27,6 +27,7 @@
#include "storage/indexfsm.h"
#include "storage/lmgr.h"
#include "utils/snapmgr.h"
+#include "utils/lsyscache.h"
/* Entry in pending-list of TIDs we need to revisit */
@@ -503,6 +504,13 @@ vacuumRedirectAndPlaceholder(Relation index, Buffer buffer)
spgxlogVacuumRedirect xlrec;
GlobalVisState *vistest;
+ /*
+ * There is no chance of endless recursion even when we are doing catalog
+ * acceses here; because, spgist is never used for catalogs. Check
+ * comments in RelationIsAccessibleInLogicalDecoding().
+ */
+ xlrec.onCatalogTable = get_rel_logical_catalog(index->rd_index->indrelid);
+
xlrec.nToPlaceholder = 0;
xlrec.newestRedirectXid = InvalidTransactionId;
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index 6bba5f8ec4..28b1f961f4 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -18,6 +18,7 @@
#include "access/hash.h"
#include "access/htup_details.h"
#include "access/nbtree.h"
+#include "access/table.h"
#include "bootstrap/bootstrap.h"
#include "catalog/namespace.h"
#include "catalog/pg_am.h"
@@ -2062,6 +2063,20 @@ get_rel_persistence(Oid relid)
return result;
}
+bool
+get_rel_logical_catalog(Oid relid)
+{
+ bool res;
+ Relation rel;
+
+ /* assume previously locked */
+ rel = table_open(relid, NoLock);
+ res = RelationIsAccessibleInLogicalDecoding(rel);
+ table_close(rel, NoLock);
+
+ return res;
+}
+
/* ---------- TRANSFORM CACHE ---------- */
diff --git a/src/include/access/gist_private.h b/src/include/access/gist_private.h
index 553d364e2d..a0f4015556 100644
--- a/src/include/access/gist_private.h
+++ b/src/include/access/gist_private.h
@@ -440,8 +440,8 @@ extern XLogRecPtr gistXLogPageDelete(Buffer buffer,
FullTransactionId xid, Buffer parentBuffer,
OffsetNumber downlinkOffset);
-extern void gistXLogPageReuse(Relation rel, BlockNumber blkno,
- FullTransactionId latestRemovedXid);
+extern void gistXLogPageReuse(Relation heapRel, Relation rel,
+ BlockNumber blkno, FullTransactionId latestRemovedXid);
extern XLogRecPtr gistXLogUpdate(Buffer buffer,
OffsetNumber *todelete, int ntodelete,
@@ -485,7 +485,7 @@ extern bool gistproperty(Oid index_oid, int attno,
extern bool gistfitpage(IndexTuple *itvec, int len);
extern bool gistnospace(Page page, IndexTuple *itvec, int len, OffsetNumber todelete, Size freespace);
extern void gistcheckpage(Relation rel, Buffer buf);
-extern Buffer gistNewBuffer(Relation r);
+extern Buffer gistNewBuffer(Relation heapRel, Relation r);
extern bool gistPageRecyclable(Page page);
extern void gistfillbuffer(Page page, IndexTuple *itup, int len,
OffsetNumber off);
diff --git a/src/include/access/gistxlog.h b/src/include/access/gistxlog.h
index fd5144f258..73999ddc70 100644
--- a/src/include/access/gistxlog.h
+++ b/src/include/access/gistxlog.h
@@ -49,9 +49,9 @@ typedef struct gistxlogPageUpdate
*/
typedef struct gistxlogDelete
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 ntodelete; /* number of deleted offsets */
-
/*
* In payload of blk 0 : todelete OffsetNumbers
*/
@@ -97,6 +97,7 @@ typedef struct gistxlogPageDelete
*/
typedef struct gistxlogPageReuse
{
+ bool onCatalogTable;
RelFileNode node;
BlockNumber block;
FullTransactionId latestRemovedFullXid;
diff --git a/src/include/access/hash_xlog.h b/src/include/access/hash_xlog.h
index 4353a32dbb..94c3292c1e 100644
--- a/src/include/access/hash_xlog.h
+++ b/src/include/access/hash_xlog.h
@@ -250,6 +250,7 @@ typedef struct xl_hash_init_bitmap_page
*/
typedef struct xl_hash_vacuum_one_page
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
int ntuples;
diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h
index 27db48184e..eba48b0aee 100644
--- a/src/include/access/heapam_xlog.h
+++ b/src/include/access/heapam_xlog.h
@@ -242,6 +242,7 @@ typedef struct xl_heap_update
*/
typedef struct xl_heap_prune
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 nredirected;
uint16 ndead;
@@ -338,6 +339,7 @@ typedef struct xl_heap_freeze_tuple
*/
typedef struct xl_heap_freeze_page
{
+ bool onCatalogTable;
TransactionId cutoff_xid;
uint16 ntuples;
} xl_heap_freeze_page;
@@ -352,6 +354,7 @@ typedef struct xl_heap_freeze_page
*/
typedef struct xl_heap_visible
{
+ bool onCatalogTable;
TransactionId cutoff_xid;
uint8 flags;
} xl_heap_visible;
@@ -413,7 +416,7 @@ extern bool heap_prepare_freeze_tuple(HeapTupleHeader tuple,
bool *totally_frozen);
extern void heap_execute_freeze_tuple(HeapTupleHeader tuple,
xl_heap_freeze_tuple *xlrec_tp);
-extern XLogRecPtr log_heap_visible(RelFileNode rnode, Buffer heap_buffer,
+extern XLogRecPtr log_heap_visible(Relation rel, Buffer heap_buffer,
Buffer vm_buffer, TransactionId cutoff_xid, uint8 flags);
#endif /* HEAPAM_XLOG_H */
diff --git a/src/include/access/nbtxlog.h b/src/include/access/nbtxlog.h
index 0f7731856b..b15aa47f1b 100644
--- a/src/include/access/nbtxlog.h
+++ b/src/include/access/nbtxlog.h
@@ -185,6 +185,7 @@ typedef struct xl_btree_dedup
*/
typedef struct xl_btree_reuse_page
{
+ bool onCatalogTable;
RelFileNode node;
BlockNumber block;
FullTransactionId latestRemovedFullXid;
@@ -232,6 +233,7 @@ typedef struct xl_btree_vacuum
typedef struct xl_btree_delete
{
+ bool onCatalogTable;
TransactionId latestRemovedXid;
uint16 ndeleted;
uint16 nupdated;
diff --git a/src/include/access/spgxlog.h b/src/include/access/spgxlog.h
index 69405b5750..06b91f4d04 100644
--- a/src/include/access/spgxlog.h
+++ b/src/include/access/spgxlog.h
@@ -237,6 +237,7 @@ typedef struct spgxlogVacuumRoot
typedef struct spgxlogVacuumRedirect
{
+ bool onCatalogTable;
uint16 nToPlaceholder; /* number of redirects to make placeholders */
OffsetNumber firstPlaceholder; /* first placeholder tuple to remove */
TransactionId newestRedirectXid; /* newest XID of removed redirects */
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index 77871aaefc..e2a5efed30 100644
--- a/src/include/utils/lsyscache.h
+++ b/src/include/utils/lsyscache.h
@@ -139,6 +139,7 @@ extern char get_rel_relkind(Oid relid);
extern bool get_rel_relispartition(Oid relid);
extern Oid get_rel_tablespace(Oid relid);
extern char get_rel_persistence(Oid relid);
+extern bool get_rel_logical_catalog(Oid relid);
extern Oid get_transform_fromsql(Oid typid, Oid langid, List *trftypes);
extern Oid get_transform_tosql(Oid typid, Oid langid, List *trftypes);
extern bool get_typisdefined(Oid typid);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 77d176a934..b488bb9618 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -26,6 +26,7 @@
#include "storage/relfilenode.h"
#include "utils/relcache.h"
#include "utils/reltrigger.h"
+#include "catalog/catalog.h"
/*
@@ -365,6 +366,9 @@ typedef struct StdRdOptions
* RelationIsUsedAsCatalogTable
* Returns whether the relation should be treated as a catalog table
* from the pov of logical decoding. Note multiple eval of argument!
+ * This definition should not invoke anything that performs catalog
+ * access; otherwise it may cause infinite recursion. Check the comments
+ * in RelationIsAccessibleInLogicalDecoding() for details.
*/
#define RelationIsUsedAsCatalogTable(relation) \
((relation)->rd_options && \
@@ -642,6 +646,11 @@ typedef struct ViewOptions
* RelationIsAccessibleInLogicalDecoding
* True if we need to log enough information to have access via
* decoding snapshot.
+ * This definition should not invoke anything that performs catalog
+ * access. Otherwise, e.g. logging a WAL entry for catalog relation may
+ * invoke this function, which will in turn do catalog access, which may
+ * in turn cause another similar WAL entry to be logged, leading to
+ * infinite recursion.
*/
#define RelationIsAccessibleInLogicalDecoding(relation) \
(XLogLogicalInfoActive() && \
--
2.18.4
view thread (196+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: [UNVERIFIED SENDER] Re: Minimal logical decoding on standbys
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox