From 594c4dcb609434fa417a0871212400bede4af44a Mon Sep 17 00:00:00 2001 From: Tomas Vondra Date: Sat, 29 Jul 2023 23:45:35 +0200 Subject: [PATCH 8/8] try adding fake transaction with sequence changes --- .../replication/logical/reorderbuffer.c | 194 ++++++++++-------- src/include/replication/reorderbuffer.h | 5 + 2 files changed, 111 insertions(+), 88 deletions(-) diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 83ac3a6a4c..e68da32a92 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -411,6 +411,7 @@ ReorderBufferAllocate(void) dlist_init(&buffer->toplevel_by_lsn); dlist_init(&buffer->txns_by_base_snapshot_lsn); + dlist_init(&buffer->sequence_changes); dclist_init(&buffer->catchange_txns); /* @@ -1231,99 +1232,25 @@ ReorderBufferQueueSequence(ReorderBuffer *rb, TransactionId xid, } else { - /* - * This increment is for a sequence that was not created in any - * running transaction, so we treat it as non-transactional and - * just send it to the output plugin directly. - */ - ReorderBufferTXN *txn = NULL; - volatile Snapshot snapshot_now = snapshot; - bool using_subtxn; - - /* non-transactional changes require a valid snapshot */ - Assert(snapshot_now); - - /* Make sure the sequence is not in any of the hash tables */ - Assert(!TransactionIdIsValid(ReorderBufferSequenceGetXid(rb, rlocator))); - - if (xid != InvalidTransactionId) - txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true); - - /* setup snapshot to allow catalog access */ - SetupHistoricSnapshot(snapshot_now, NULL); - - /* - * Decoding needs access to syscaches et al., which in turn use - * heavyweight locks and such. Thus we need to have enough state around to - * keep track of those. The easiest way is to simply use a transaction - * internally. That also allows us to easily enforce that nothing writes - * to the database by checking for xid assignments. - * - * When we're called via the SQL SRF there's already a transaction - * started, so start an explicit subtransaction there. - */ - using_subtxn = IsTransactionOrTransactionBlock(); - - PG_TRY(); - { - Relation relation; - HeapTuple tuple; - Form_pg_sequence_data seq; - Oid reloid; - int64 value; - - if (using_subtxn) - BeginInternalSubTransaction("sequence"); - else - StartTransactionCommand(); - - reloid = RelidByRelfilenumber(rlocator.spcOid, rlocator.relNumber); - - if (reloid == InvalidOid) - elog(ERROR, "could not map filenode \"%s\" to relation OID", - relpathperm(rlocator, - MAIN_FORKNUM)); - - relation = RelationIdGetRelation(reloid); - - if (!RelationIsValid(relation)) - elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")", - reloid, - relpathperm(rlocator, MAIN_FORKNUM)); - - tuple = &tuplebuf->tuple; - seq = (Form_pg_sequence_data) GETSTRUCT(tuple); - - /* - * Calculate the first value of the next batch (at which point we - * generate and decode another WAL record. - */ - value = seq->last_value; - value += (seq->is_called) ? seq->log_cnt : 0; - - rb->sequence(rb, txn, lsn, relation, transactional, value); + MemoryContext oldcontext; + ReorderBufferChange *change; - RelationClose(relation); + /* allocate and queue the transactional sequence change */ + oldcontext = MemoryContextSwitchTo(rb->context); - TeardownHistoricSnapshot(false); + change = ReorderBufferGetChange(rb); - AbortCurrentTransaction(); + change->action = REORDER_BUFFER_CHANGE_SEQUENCE; + change->origin_id = origin_id; + change->lsn = lsn; - if (using_subtxn) - RollbackAndReleaseCurrentSubTransaction(); - } - PG_CATCH(); - { - TeardownHistoricSnapshot(true); + memcpy(&change->data.sequence.locator, &rlocator, sizeof(RelFileLocator)); - AbortCurrentTransaction(); + change->data.sequence.tuple = tuplebuf; - if (using_subtxn) - RollbackAndReleaseCurrentSubTransaction(); + dlist_push_tail(&rb->sequence_changes, &change->node); - PG_RE_THROW(); - } - PG_END_TRY(); + MemoryContextSwitchTo(oldcontext); } } @@ -2209,8 +2136,11 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) dclist_delete_from(&rb->catchange_txns, &txn->catchange_node); /* now remove reference from buffer */ - hash_search(rb->by_txn, &txn->xid, HASH_REMOVE, &found); - Assert(found); + if (TransactionIdIsValid(txn->xid)) /* XXX fake sequence TXN has invalid XID */ + { + hash_search(rb->by_txn, &txn->xid, HASH_REMOVE, &found); + Assert(found); + } /* remove entries spilled to disk */ if (rbtxn_is_serialized(txn)) @@ -3372,6 +3302,81 @@ ReorderBufferReplay(ReorderBufferTXN *txn, command_id, false); } +/* + * Create a "fake" substransaction injecting the non-transactional sequence + * changes that happened until this commit. + * + * XXX This has issues with snapshots. Firstly, the "parent" transaction may + * be empty withoug a base_snapshot, so that ReorderBufferReplay() will treat + * is as empty transaction - and won't apply even the sequence change. + * + * Secondly, the sequence changes originate from other transactions, so it's + * not clear if we can simply transfer snapshots in some way. Maybe we can? + * sequence_decode() knows the *current* snapshot, but can we keep it for + * later, somehow? Or can we maybe just use some "fresh" snapshot (as these + * non-transactional changes should not change sequence-related catalogs). + */ +static void +ReorderBufferAddSequenceTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr end_lsn) +{ + dlist_mutable_iter miter; + ReorderBufferTXN *subtxn = NULL; + + dlist_foreach_modify(miter, &rb->sequence_changes) + { + ReorderBufferChange *change = dlist_container(ReorderBufferChange, node, + miter.cur); + + /* changes are ordered by LSN, so stop once we exceed end_lsn */ + if (change->lsn > end_lsn) + break; + + if (subtxn == NULL) + { + subtxn = ReorderBufferGetTXN(rb); + + subtxn->first_lsn = change->lsn; + + /* similar to ReorderBufferAssignChild, but we don't have XIDs */ + subtxn->txn_flags |= RBTXN_IS_SUBXACT; + subtxn->toplevel_xid = txn->xid; + + Assert(subtxn->nsubtxns == 0); + + /* set the reference to top-level transaction */ + subtxn->toptxn = txn; + + /* add to subtransaction list */ + dlist_push_tail(&txn->subtxns, &subtxn->node); + txn->nsubtxns++; + } + + /* remove the change from the top-level list of changes */ + dlist_delete(miter.cur); + + change->txn = subtxn; + + /* add it to the next subxact */ + dlist_push_tail(&subtxn->changes, &change->node); + subtxn->nentries++; + subtxn->nentries_mem++; + + /* + * XXX should this do memory limit checks and partial changes similar to + * ReorderBufferQueueChange? + * + * We have to do at least the accounting, so that the numbers match. + */ + + /* update memory accounting information */ + ReorderBufferChangeMemoryUpdate(rb, change, true, + ReorderBufferChangeSize(change)); + + /* check the memory limits and evict something if needed */ + ReorderBufferCheckMemoryLimit(rb); + } +} + /* * Commit a transaction. * @@ -3388,6 +3393,19 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false); + /* + * Maybe add a "fake" subtransaction with non-transactional sequence + * changes that happened since the last commit. + * + * XXX I wonder if this might have issues with sequence changes preceding + * the first LSN of the transaction. What snapshot would those changes use? + * Although, maybe we should just skip those changes, at it has to be from + * not-committed transaction. + * + * XXX Should this be looking at commit_lsn or end_lsn? + */ + ReorderBufferAddSequenceTXN(rb, txn, commit_lsn); + /* unknown transaction, nothing to replay */ if (txn == NULL) return; diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index e5d8e1c0d1..97d6e56b4c 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -602,6 +602,11 @@ struct ReorderBuffer */ dlist_head txns_by_base_snapshot_lsn; + /* + * Non-transactional sequence changes, applied in the nearest commit. + */ + dlist_head sequence_changes; + /* * Transactions and subtransactions that have modified system catalogs. */ -- 2.41.0