logical decoding: Correctly free speculative insertion

The error path in ReorderBufferProcessTXN was not freeing
(reorderbuffer.c's representation of) a speculative insertion record
correctly.  In assert-enabled builds, this leads to an assertion
failure.  In production builds, I see no effect; there may be a small
transient leak, but in an improbable code path such as this, such a leak
is not of any significance.  For users running with assertions enabled,
the crash is annoying.

Fix by having ReorderBufferProcessTXN() free the speculative insert
ahead of freeing the rest of the transaction, and no longer try to
handle that insert as a separate argument to ReorderBufferResetTXN().

This code came in with commit 7259736a6e5b (14-era).  Backpatch all the
way back.

In branches 14-16, also backpatch the assertion that originally fails in
the problem scenario, which was added by dbed2e36625d (originally
backpatched to 17), that at the end of ReorderBufferReturnTXN() the
in-memory size of the transaction is zero.

Author: Vishal Prasanna <vishal.g@zohocorp.com>
Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Backpatch-through: 14
Discussion: https://postgr.es/m/19c7623e882.4080fd5426212.311756747309556767@zohocorp.com

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 59efa73930fa5630d409e5edb31c06b2bab2079e..b190b4712bbb508fb1728c0149ddf4f058d335af 100644 (file)
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -478,6 +478,9 @@ ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
        /* Reset the toast hash */
        ReorderBufferToastReset(rb, txn);
 
+       /* All changes must be deallocated */
+       Assert(txn->size == 0);
+
        pfree(txn);
 }
 
@@ -2033,8 +2036,7 @@ static void
 ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                                          Snapshot snapshot_now,
                                          CommandId command_id,
-                                         XLogRecPtr last_lsn,
-                                         ReorderBufferChange *specinsert)
+                                         XLogRecPtr last_lsn)
 {
        /* Discard the changes that we just streamed */
        ReorderBufferTruncateTXN(rb, txn, rbtxn_prepared(txn));
@@ -2042,13 +2044,6 @@ ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
        /* Free all resources allocated for toast reconstruction */
        ReorderBufferToastReset(rb, txn);
 
-       /* Return the spec insert change if it is not NULL */
-       if (specinsert != NULL)
-       {
-               ReorderBufferReturnChange(rb, specinsert, true);
-               specinsert = NULL;
-       }
-
        /*
         * For the streaming case, stop the stream and remember the command ID and
         * snapshot for the streaming run.
@@ -2307,7 +2302,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                                         * CheckTableNotInUse() and locking.
                                         */
 
-                                       /* clear out a pending (and thus failed) speculation */
+                                       /* clear out a pending (= failed) speculative insertion */
                                        if (specinsert != NULL)
                                        {
                                                ReorderBufferReturnChange(rb, specinsert, true);
@@ -2592,6 +2587,13 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                if (using_subtxn)
                        RollbackAndReleaseCurrentSubTransaction();
 
+               /* Free the specinsert change before freeing the ReorderBufferTXN */
+               if (specinsert != NULL)
+               {
+                       ReorderBufferReturnChange(rb, specinsert, true);
+                       specinsert = NULL;
+               }
+
                /*
                 * The error code ERRCODE_TRANSACTION_ROLLBACK indicates a concurrent
                 * abort of the (sub)transaction we are streaming or preparing. We
@@ -2618,8 +2620,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 
                        /* Reset the TXN so that it is allowed to stream remaining data. */
                        ReorderBufferResetTXN(rb, txn, snapshot_now,
-                                                                 command_id, prev_lsn,
-                                                                 specinsert);
+                                                                 command_id, prev_lsn);
                }
                else
                {

diff --git a/src/test/subscription/t/100_bugs.pl b/src/test/subscription/t/100_bugs.pl
index 061eb2a32c1ed262f8b4ec231b2ec7ed6e9ea864..eee297efa2b472da6d3d283dde48fefd8169c752 100644 (file)
--- a/src/test/subscription/t/100_bugs.pl
+++ b/src/test/subscription/t/100_bugs.pl
@@ -448,4 +448,49 @@ $node_publisher->safe_psql('postgres', "DROP DATABASE regress_db");
 
 $node_publisher->stop('fast');
 
+# https://postgr.es/m/19c7623e882.4080fd5426212.311756747309556767%40zohocorp.com
+
+# The bug was that when an ERROR was raised while processing an INSERT ... ON
+# CONFLICT statement, the decoded change misses to be free'd. This can cause an
+# assertion failure if enabled.
+
+$node_publisher->rotate_logfile();
+$node_publisher->start();
+
+# Create a publication with the zero-division row filter. It always throws an
+# ERROR before publishing changes, when the filter is evaluated.
+$node_publisher->safe_psql(
+       'postgres', qq(
+       CREATE TABLE tab_upsert (a INT PRIMARY KEY, b INT);
+       CREATE PUBLICATION pub_rowfilter_error FOR TABLE tab_upsert WHERE ((a / 0) > 0);
+       SELECT * FROM pg_create_logical_replication_slot('upsert_slot', 'pgoutput');
+       INSERT INTO tab_upsert (a, b) VALUES (1, 1)
+               ON CONFLICT(a) DO UPDATE SET b = excluded.b;
+));
+
+# Decode the changes with a publication whose row filter causes a
+# division by zero error, and verify that the logical decoder doesn't crash.
+($ret, $stdout, $stderr) = $node_publisher->psql(
+       'postgres', qq(
+       SELECT *
+       FROM pg_logical_slot_peek_binary_changes(
+               'upsert_slot',
+               NULL,
+               NULL,
+               'proto_version', '1',
+               'publication_names', 'pub_rowfilter_error'
+       );
+));
+
+ok( $stderr =~ qr/division by zero/,
+       'peek logical changes with row filter causing division by zero throws error'
+);
+
+# Clean up
+$node_publisher->safe_psql('postgres', "SELECT pg_drop_replication_slot('upsert_slot')");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION pub_rowfilter_error");
+$node_publisher->safe_psql('postgres', "DROP TABLE tab_upsert");
+
+$node_publisher->stop('fast');
+
 done_testing();