Re: logical decoding and replication of sequences, take 2

cfbot didn't like the rebased / split patch, and after looking at it I
believe it's a bug in parallel apply of large transactions (216a784829),
which seems to have changed interpretation of in_remote_transaction and
in_streamed_transaction. I've reported the issue on that thread [1], but
here's a version with a temporary workaround so that we can continue
reviewing it.

regards

[1]
https://www.postgresql.org/message-id/984ff689-adde-9977-affe-cd6029e850be%40enterprisedb.com

On 1/15/23 00:39, Tomas Vondra wrote:
> Hi,
>
> here's a slightly updated version - the main change is splitting the
> patch into multiple parts, along the lines of the original patch
> reverted in 2c7ea57e56ca5f668c32d4266e0a3e45b455bef5:
>
> - basic sequence decoding infrastructure
> - support in test_decoding
> - support in built-in logical replication
>
> The revert mentions a couple additional parts, but those were mostly
> fixes / improvements. And those are not merged into the three parts.
>
>
> On 1/11/23 22:46, Tomas Vondra wrote:
>>
>>> ...
>>>
>>>> +/*
>>>> + * Update the sequence state by modifying the existing sequence data row.
>>>> + *
>>>> + * This keeps the same relfilenode, so the behavior is non-transactional.
>>>> + */
>>>> +static void
>>>> +SetSequence_non_transactional(Oid seqrelid, int64 last_value, int64 log_cnt, bool is_called)
>>>> +{
>>>> ...
>>>>
>>>> +void
>>>> +SetSequence(Oid seq_relid, bool transactional, int64 last_value, int64 log_cnt, bool is_called)
>>>> +{
>>>> + if (transactional)
>>>> + SetSequence_transactional(seq_relid, last_value, log_cnt, is_called);
>>>> + else
>>>> + SetSequence_non_transactional(seq_relid, last_value, log_cnt, is_called);
>>>> +}
>>>
>>> That's a lot of duplication with existing code. There's no explanation why
>>> SetSequence() as well as do_setval() exists.
>>>
>>
>> Thanks, I'll look into this.
>>
>
> I haven't done anything about this yet. The functions are doing similar
> things, but there's also a fair amount of differences so I haven't found
> a good way to merge them yet.
>
>>>
>>>> /*
>>>> * Initialize a sequence's relation with the specified tuple as content
>>>> *
>>>> @@ -406,8 +560,13 @@ fill_seq_fork_with_data(Relation rel, HeapTuple tuple, ForkNumber forkNum)
>>>>
>>>> /* check the comment above nextval_internal()'s equivalent call. */
>>>> if (RelationNeedsWAL(rel))
>>>> + {
>>>> GetTopTransactionId();
>>>>
>>>> + if (XLogLogicalInfoActive())
>>>> + GetCurrentTransactionId();
>>>> + }
>>>
>>> Is it actually possible to reach this without an xid already having been
>>> assigned for the current xact?
>>>
>>
>> I believe it is. That's probably how I found this change is needed,
>> actually.
>>
>
> I've added a comment explaining why this needed. I don't think it's
> worth trying to optimize this, because in plausible workloads we'd just
> delay the work a little bit.
>
>>>
>>>
>>>> @@ -806,10 +966,28 @@ nextval_internal(Oid relid, bool check_permissions)
>>>> * It's sufficient to ensure the toplevel transaction has an xid, no need
>>>> * to assign xids subxacts, that'll already trigger an appropriate wait.
>>>> * (Have to do that here, so we're outside the critical section)
>>>> + *
>>>> + * We have to ensure we have a proper XID, which will be included in
>>>> + * the XLOG record by XLogRecordAssemble. Otherwise the first nextval()
>>>> + * in a subxact (without any preceding changes) would get XID 0, and it
>>>> + * would then be impossible to decide which top xact it belongs to.
>>>> + * It'd also trigger assert in DecodeSequence. We only do that with
>>>> + * wal_level=logical, though.
>>>> + *
>>>> + * XXX This might seem unnecessary, because if there's no XID the xact
>>>> + * couldn't have done anything important yet, e.g. it could not have
>>>> + * created a sequence. But that's incorrect, because of subxacts. The
>>>> + * current subtransaction might not have done anything yet (thus no XID),
>>>> + * but an earlier one might have created the sequence.
>>>> */
>>>
>>> What about restricting this to the case you're mentioning,
>>> i.e. subtransactions?
>>>
>>
>> That might work, but I need to think about it a bit.
>>
>> I don't think it'd save us much, though. I mean, vast majority of
>> transactions (and subtransactions) calling nextval() will then do
>> something else which requires a XID. This just moves the XID a bit,
>> that's all.
>>
>
> After thinking about this a bit more, I don't think the optimization is
> worth it, for the reasons explained above.
>
>>>
>>>> +/*
>>>> + * Handle sequence decode
>>>> + *
>>>> + * Decoding sequences is a bit tricky, because while most sequence actions
>>>> + * are non-transactional (not subject to rollback), some need to be handled
>>>> + * as transactional.
>>>> + *
>>>> + * By default, a sequence increment is non-transactional - we must not queue
>>>> + * it in a transaction as other changes, because the transaction might get
>>>> + * rolled back and we'd discard the increment. The downstream would not be
>>>> + * notified about the increment, which is wrong.
>>>> + *
>>>> + * On the other hand, the sequence may be created in a transaction. In this
>>>> + * case we *should* queue the change as other changes in the transaction,
>>>> + * because we don't want to send the increments for unknown sequence to the
>>>> + * plugin - it might get confused about which sequence it's related to etc.
>>>> + */
>>>> +void
>>>> +sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
>>>> +{
>>>
>>>> + /* extract the WAL record, with "created" flag */
>>>> + xlrec = (xl_seq_rec *) XLogRecGetData(r);
>>>> +
>>>> + /* XXX how could we have sequence change without data? */
>>>> + if(!datalen || !tupledata)
>>>> + return;
>>>
>>> Yea, I think we should error out here instead, something has gone quite wrong
>>> if this happens.
>>>
>>
>> OK
>>
>
> Done.
>
>>>
>>>> + tuplebuf = ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
>>>> + DecodeSeqTuple(tupledata, datalen, tuplebuf);
>>>> +
>>>> + /*
>>>> + * Should we handle the sequence increment as transactional or not?
>>>> + *
>>>> + * If the sequence was created in a still-running transaction, treat
>>>> + * it as transactional and queue the increments. Otherwise it needs
>>>> + * to be treated as non-transactional, in which case we send it to
>>>> + * the plugin right away.
>>>> + */
>>>> + transactional = ReorderBufferSequenceIsTransactional(ctx->reorder,
>>>> + target_locator,
>>>> + xlrec->created);
>>>
>>> Why re-create this information during decoding, when we basically already have
>>> it available on the primary? I think we already pay the price for that
>>> tracking, which we e.g. use for doing a non-transactional truncate:
>>>
>>> /*
>>> * Normally, we need a transaction-safe truncation here. However, if
>>> * the table was either created in the current (sub)transaction or has
>>> * a new relfilenumber in the current (sub)transaction, then we can
>>> * just truncate it in-place, because a rollback would cause the whole
>>> * table or the current physical file to be thrown away anyway.
>>> */
>>> if (rel->rd_createSubid == mySubid ||
>>> rel->rd_newRelfilelocatorSubid == mySubid)
>>> {
>>> /* Immediate, non-rollbackable truncation is OK */
>>> heap_truncate_one_rel(rel);
>>> }
>>>
>>> Afaict we could do something similar for sequences, except that I think we
>>> would just check if the sequence was created in the current transaction
>>> (i.e. any of the fields are set).
>>>
>>
>> Hmm, good point.
>>
>
> But rd_createSubid/rd_newRelfilelocatorSubid fields are available only
> in the original transaction, not during decoding. So we'd have to do
> this check there and add the result to the WAL record. Is that what you
> had in mind?
>
>>>
>>>> +/*
>>>> + * A transactional sequence increment is queued to be processed upon commit
>>>> + * and a non-transactional increment gets processed immediately.
>>>> + *
>>>> + * A sequence update may be both transactional and non-transactional. When
>>>> + * created in a running transaction, treat it as transactional and queue
>>>> + * the change in it. Otherwise treat it as non-transactional, so that we
>>>> + * don't forget the increment in case of a rollback.
>>>> + */
>>>> +void
>>>> +ReorderBufferQueueSequence(ReorderBuffer *rb, TransactionId xid,
>>>> + Snapshot snapshot, XLogRecPtr lsn, RepOriginId origin_id,
>>>> + RelFileLocator rlocator, bool transactional, bool created,
>>>> + ReorderBufferTupleBuf *tuplebuf)
>>>
>>>
>>>> + /*
>>>> + * 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();
>>>
>>> This duplicates a lot of the code from ReorderBufferProcessTXN(). But only
>>> does so partially. It's hard to tell whether some of the differences are
>>> intentional. Could we de-duplicate that code with ReorderBufferProcessTXN()?
>>>
>>> Maybe something like
>>>
>>> void
>>> ReorderBufferSetupXactEnv(ReorderBufferXactEnv *, bool process_invals);
>>>
>>> void
>>> ReorderBufferTeardownXactEnv(ReorderBufferXactEnv *, bool is_error);
>>>
>>
>> Thanks for the suggestion, I'll definitely consider that in the next
>> version of the patch.
>
> I did look at the code a bit, but I'm not sure there really is a lot of
> duplicated code - yes, we start/abort the (sub)transaction, setup and
> tear down the snapshot, etc. Or what else would you put into the two new
> functions?
>
>
> regards
>

--
Tomas Vondra
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company