On Mon, Jun 7, 2021 at 8:46 AM Dilip Kumar <dilipbal...@gmail.com> wrote:
> On Mon, 7 Jun 2021 at 8:30 AM, Amit Kapila <amit.kapil...@gmail.com>
> wrote:
>
>> On Wed, Jun 2, 2021 at 11:52 AM Amit Kapila <amit.kapil...@gmail.com>
>> wrote:
>> >
>> > On Wed, Jun 2, 2021 at 11:38 AM Dilip Kumar <dilipbal...@gmail.com>
>> wrote:
>> > >
>> > > On Wed, Jun 2, 2021 at 11:25 AM Amit Kapila <amit.kapil...@gmail.com>
>> wrote:
>> > > >
>> > > > I think the same relation case might not create a problem because it
>> > > > won't find the entry for it in the toast_hash, so it will return
>> from
>> > > > there but the other two problems will be there.
>> > >
>> > > Right
>> > >
>> > > So, one idea could be
>> > > > to just avoid those two cases (by simply adding return for those
>> > > > cases) and still we can rely on toast clean up on the next
>> > > > insert/update/delete. However, your approach looks more natural to
>> me
>> > > > as that will allow us to clean up memory in all cases instead of
>> > > > waiting till the transaction end. So, I still vote for going with
>> your
>> > > > patch's idea of cleaning at spec_abort but I am fine if you and
>> others
>> > > > decide not to process spec_abort message. What do you think? Tomas,
>> do
>> > > > you have any opinion on this matter?
>> > >
>> > > I agree that processing with spec abort looks more natural and ideally
>> > > the current code expects it to be getting cleaned after the change,
>> > > that's the reason we have those assertions and errors.
>> > >
>>
>> Okay, so, let's go with that approach. I have thought about whether it
>> creates any problem in back-branches but couldn't think of any
>> primarily because we are not going to send anything additional to
>> plugin/subscriber. Do you see any problems with back branches if we go
>> with this approach?
>
>
> I will check this and let you know.
>
>
>> > > OTOH I agree
>> > > that we can just return from those conditions because now we know that
>> > > with the current code those situations are possible. My vote is with
>> > > handling the spec abort option (Option1) because that looks more
>> > > natural way of handling these issues and we also don't have to clean
>> > > up the hash in "ReorderBufferReturnTXN" if no followup change after
>> > > spec abort.
>> > >
>> >
>> > Even, if we decide to go with spec_abort approach, it might be better
>> > to still keep the toastreset call in ReorderBufferReturnTXN so that it
>> > can be freed in case of error.
>> >
>>
>> Please take care of this as well.
>
>
> Ok
>
I have fixed all pending review comments and also added a test case which
is working fine. I haven't yet checked on the back branches. Let's
discuss if we think this patch looks fine then I can apply and test on the
back branches.
--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com
From 0b9c93398ef108a3d71cbac6f793a0314964aaa2 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilipkumar@localhost.localdomain>
Date: Tue, 1 Jun 2021 19:53:47 +0530
Subject: [PATCH v3] Bug fix for speculative abort
If speculative insert has a toast table insert then if that tuple
is not confirmed then the toast hash is not cleaned and that is
creating various problem like a) memory leak b) next insert is
using these uncleaned toast data for its insertion and other
error and assersion failure. So this patch handle that by
queuing the spec abort changes and cleaning up the toast hash
on spec abort. Currently, in this patch we are queuing up all
the spec abort changes, but as an optimization we can avoid
queuing the spec abort for toast tables but for that we need to
log that as a flag in WAL.
---
contrib/test_decoding/Makefile | 2 +-
.../test_decoding/expected/speculative_abort.out | 64 +++++++++++++++++++++
contrib/test_decoding/specs/speculative_abort.spec | 67 ++++++++++++++++++++++
src/backend/replication/logical/decode.c | 14 ++---
src/backend/replication/logical/reorderbuffer.c | 43 +++++++++++++-
src/include/replication/reorderbuffer.h | 1 +
6 files changed, 180 insertions(+), 11 deletions(-)
create mode 100644 contrib/test_decoding/expected/speculative_abort.out
create mode 100644 contrib/test_decoding/specs/speculative_abort.spec
diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile
index 9a31e0b..1cab935 100644
--- a/contrib/test_decoding/Makefile
+++ b/contrib/test_decoding/Makefile
@@ -8,7 +8,7 @@ REGRESS = ddl xact rewrite toast permissions decoding_in_xact \
spill slot truncate stream stats twophase twophase_stream
ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \
oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \
- twophase_snapshot
+ twophase_snapshot speculative_abort
REGRESS_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf
ISOLATION_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf
diff --git a/contrib/test_decoding/expected/speculative_abort.out b/contrib/test_decoding/expected/speculative_abort.out
new file mode 100644
index 0000000..d672deb
--- /dev/null
+++ b/contrib/test_decoding/expected/speculative_abort.out
@@ -0,0 +1,64 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s1_session s1_lock_s2 s1_lock_s3 s1_begin s1_insert_tbl1 s2_session s2_begin s2_insert_tbl1 s3_session s3_begin s3_insert_tbl1 s1_unlock_s2 s1_unlock_s3 s1_lock_s2 s1_abort s3_commit s1_unlock_s2 s2_insert_tbl2 s2_commit s1_get_changes
+data
+
+step s1_session: SET spec.session = 1;
+step s1_lock_s2: SELECT pg_advisory_lock(2);
+pg_advisory_lock
+
+
+step s1_lock_s3: SELECT pg_advisory_lock(2);
+pg_advisory_lock
+
+
+step s1_begin: BEGIN;
+step s1_insert_tbl1: INSERT INTO tbl1 VALUES(1, repeat('a', 4000)) ON CONFLICT DO NOTHING;
+step s2_session: SET spec.session = 2;
+step s2_begin: BEGIN;
+s2: NOTICE: 2acquiring advisory lock on 2
+step s2_insert_tbl1: INSERT INTO tbl1 VALUES(1, repeat('a', 4000)) ON CONFLICT DO NOTHING; <waiting ...>
+step s3_session: SET spec.session = 3;
+step s3_begin: BEGIN;
+s3: NOTICE: 3acquiring advisory lock on 3
+s3: NOTICE: 3acquiring advisory lock on 3
+step s3_insert_tbl1: INSERT INTO tbl1 VALUES(1, repeat('a', 4000)) ON CONFLICT DO NOTHING; <waiting ...>
+step s1_unlock_s2: SELECT pg_advisory_unlock(2);
+pg_advisory_unlock
+
+t
+step s1_unlock_s3: SELECT pg_advisory_unlock(2);
+pg_advisory_unlock
+
+t
+s2: NOTICE: 2acquiring advisory lock on 2
+step s1_lock_s2: SELECT pg_advisory_lock(2);
+pg_advisory_lock
+
+
+step s1_abort: ROLLBACK;
+s2: NOTICE: 2acquiring advisory lock on 2
+s3: NOTICE: 3acquiring advisory lock on 3
+step s3_insert_tbl1: <... completed>
+step s3_commit: COMMIT;
+step s1_unlock_s2: SELECT pg_advisory_unlock(2);
+pg_advisory_unlock
+
+t
+s2: NOTICE: 2acquiring advisory lock on 2
+s2: NOTICE: 2acquiring advisory lock on 2
+step s2_insert_tbl1: <... completed>
+step s2_insert_tbl2: INSERT INTO tbl2 VALUES(1);
+step s2_commit: COMMIT;
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+data
+
+BEGIN
+table public.tbl1: INSERT: a[integer]:1 b[text]:'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
+COMMIT
+BEGIN
+table public.tbl2: INSERT: a[integer]:1
+COMMIT
+?column?
+
+stop
diff --git a/contrib/test_decoding/specs/speculative_abort.spec b/contrib/test_decoding/specs/speculative_abort.spec
new file mode 100644
index 0000000..0e2ab29
--- /dev/null
+++ b/contrib/test_decoding/specs/speculative_abort.spec
@@ -0,0 +1,67 @@
+setup
+{
+ SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding');
+ DROP TABLE IF EXISTS tbl1;
+ CREATE TABLE tbl1 (a INT, b TEXT);
+ ALTER TABLE tbl1 ALTER COLUMN b SET STORAGE EXTERNAL;
+ CREATE TABLE tbl2 (a INT);
+
+ CREATE OR REPLACE FUNCTION blurt_and_lock(int) RETURNS int IMMUTABLE LANGUAGE plpgsql AS $$
+ BEGIN
+ -- depending on lock state, wait for lock 2 or 3
+ IF current_setting('spec.session')::int = 2 THEN
+ RAISE NOTICE '2acquiring advisory lock on 2';
+ PERFORM pg_advisory_lock(2);
+ PERFORM pg_advisory_unlock(2);
+ ELSIF current_setting('spec.session')::int = 3 THEN
+ RAISE NOTICE '3acquiring advisory lock on 3';
+ PERFORM pg_advisory_lock(3);
+ PERFORM pg_advisory_unlock(3);
+ END IF;
+ RETURN $1;
+ END;$$;
+
+ CREATE UNIQUE INDEX idx on tbl1(blurt_and_lock(a));
+
+ -- consume DDL
+ SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+}
+
+teardown
+{
+ DROP TABLE tbl1;
+ SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot');
+}
+
+session "s1"
+setup { SET synchronous_commit=on; }
+
+step "s1_lock_s2" { SELECT pg_advisory_lock(2); }
+step "s1_lock_s3" { SELECT pg_advisory_lock(2); }
+step "s1_session" { SET spec.session = 1; }
+step "s1_begin" { BEGIN; }
+step "s1_insert_tbl1" { INSERT INTO tbl1 VALUES(1, repeat('a', 4000)) ON CONFLICT DO NOTHING; }
+step "s1_abort" { ROLLBACK; }
+step "s1_unlock_s2" { SELECT pg_advisory_unlock(2); }
+step "s1_unlock_s3" { SELECT pg_advisory_unlock(2); }
+step "s1_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); }
+
+session "s2"
+setup { SET synchronous_commit=on; }
+
+step "s2_session" { SET spec.session = 2; }
+step "s2_begin" { BEGIN; }
+step "s2_insert_tbl1" { INSERT INTO tbl1 VALUES(1, repeat('a', 4000)) ON CONFLICT DO NOTHING; }
+step "s2_insert_tbl2" { INSERT INTO tbl2 VALUES(1); }
+step "s2_commit" { COMMIT; }
+
+session "s3"
+setup { SET synchronous_commit=on; }
+
+step "s3_session" { SET spec.session = 3; }
+step "s3_begin" { BEGIN; }
+step "s3_insert_tbl1" { INSERT INTO tbl1 VALUES(1, repeat('a', 4000)) ON CONFLICT DO NOTHING; }
+step "s3_commit" { COMMIT; }
+
+
+permutation "s1_session" "s1_lock_s2" "s1_lock_s3" "s1_begin" "s1_insert_tbl1" "s2_session" "s2_begin" "s2_insert_tbl1" "s3_session" "s3_begin" "s3_insert_tbl1" "s1_unlock_s2" "s1_unlock_s3" "s1_lock_s2" "s1_abort" "s3_commit" "s1_unlock_s2" "s2_insert_tbl2" "s2_commit" "s1_get_changes"
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 7067016..453efc5 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -1040,19 +1040,17 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
if (target_node.dbNode != ctx->slot->data.database)
return;
- /*
- * Super deletions are irrelevant for logical decoding, it's driven by the
- * confirmation records.
- */
- if (xlrec->flags & XLH_DELETE_IS_SUPER)
- return;
-
/* output plugin doesn't look for this origin, no need to queue */
if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
return;
change = ReorderBufferGetChange(ctx->reorder);
- change->action = REORDER_BUFFER_CHANGE_DELETE;
+
+ if (xlrec->flags & XLH_DELETE_IS_SUPER)
+ change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT;
+ else
+ change->action = REORDER_BUFFER_CHANGE_DELETE;
+
change->origin_id = XLogRecGetOrigin(r);
memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 2d9e127..dd95785 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -443,6 +443,9 @@ ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
txn->invalidations = NULL;
}
+ /* Reset the toast hash */
+ ReorderBufferToastReset(rb, txn);
+
pfree(txn);
}
@@ -520,6 +523,7 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change,
}
break;
case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
+ case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
break;
@@ -2211,8 +2215,8 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
change_done:
/*
- * Either speculative insertion was confirmed, or it was
- * unsuccessful and the record isn't needed anymore.
+ * If speculative insertion was confirmed, the record isn't
+ * needed anymore.
*/
if (specinsert != NULL)
{
@@ -2254,6 +2258,38 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
specinsert = change;
break;
+ case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
+
+ /*
+ * Abort for speculative insertion arrived. So cleanup the
+ * specinsert tuple and toast hash. If spec insert change
+ * is NULL then do nothing, this is possible because we
+ * have spec abort for each toast entry. So we just have
+ * to clean the specinsert and toast hash for the first
+ * spec abort for the main table and remaining changes for
+ * the tables can be ignored.
+ */
+ if (specinsert != NULL)
+ {
+ /*
+ * Clear the toast hash, we must clean the toast hash
+ * before we start with a completely new tuple,
+ * otherwise, while processing the new tuple it would
+ * create a confusion that whether we need to process
+ * these toast chunks or not.
+ */
+ Assert(change->data.tp.clear_toast_afterwards);
+ ReorderBufferToastReset(rb, txn);
+
+ /*
+ * If the speculative insertion was aborted, the record
+ * isn't needed anymore.
+ */
+ ReorderBufferReturnChange(rb, specinsert, true);
+ specinsert = NULL;
+ }
+ break;
+
case REORDER_BUFFER_CHANGE_TRUNCATE:
{
int i;
@@ -3754,6 +3790,7 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
break;
}
case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
+ case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
/* ReorderBufferChange contains everything important */
@@ -4017,6 +4054,7 @@ ReorderBufferChangeSize(ReorderBufferChange *change)
break;
}
case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
+ case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
/* ReorderBufferChange contains everything important */
@@ -4315,6 +4353,7 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
break;
}
case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
+ case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
break;
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 0c6e9d1..9ff0986 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -63,6 +63,7 @@ enum ReorderBufferChangeType
REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID,
REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT,
REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM,
+ REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT,
REORDER_BUFFER_CHANGE_TRUNCATE
};
--
1.8.3.1
