Re: PATCH: logical_work_mem and logical streaming of large in-progress transactions

On Mon, Apr 13, 2020 at 5:20 PM Dilip Kumar <dilipbalaut(at)gmail(dot)com> wrote:
> On Mon, Apr 13, 2020 at 4:14 PM Kuntal Ghosh <kuntalghosh(dot)2007(at)gmail(dot)com> wrote:
> >
> > +#define SizeOfTransactionId (sizeof(TransactionId) + sizeof(char))
> > This looks wrong. We should change the name of this Macro or we can
> > add the 1 byte directly in HEADER_SCRATCH_SIZE and some comments.
>
> I think this is in sync with below code (SizeOfXlogOrigin), SO doen't
> make much sense to add different terminology no?
> #define SizeOfXlogOrigin (sizeof(RepOriginId) + sizeof(char))
> +#define SizeOfTransactionId (sizeof(TransactionId) + sizeof(char))
>
In that case, we can rename this, for example, SizeOfXLogTransactionId.

Some review comments from 0002-Issue-individual-*.path,

+void
+ReorderBufferAddInvalidation(ReorderBuffer *rb, TransactionId xid,
+ XLogRecPtr lsn, int nmsgs,
+ SharedInvalidationMessage *msgs)
+{
+ MemoryContext oldcontext;
+ ReorderBufferChange *change;
+
+ /* XXX Should we even write invalidations without valid XID? */
+ if (xid == InvalidTransactionId)
+ return;
+
+ Assert(xid != InvalidTransactionId);

It seems we don't call the function if xid is not valid. In fact,

@@ -281,6 +281,24 @@ DecodeXactOp(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf)
}
case XLOG_XACT_ASSIGNMENT:
break;
+ case XLOG_XACT_INVALIDATIONS:
+ {
+ TransactionId xid;
+ xl_xact_invalidations *invals;
+
+ xid = XLogRecGetXid(r);
+ invals = (xl_xact_invalidations *) XLogRecGetData(r);
+
+ if (!TransactionIdIsValid(xid))
+ break;
+
+ ReorderBufferAddInvalidation(reorder, xid, buf->origptr,
+ invals->nmsgs, invals->msgs);

Why should we insert an WAL record for such cases?

+ * When wal_level=logical, write invalidations into WAL at each command end to
+ * support the decoding of the in-progress transaction. As of now it was
+ * enough to log invalidation only at commit because we are only decoding the
+ * transaction at the commit time. We only need to log the catalog cache and
+ * relcache invalidation. There can not be any active MVCC scan in logical
+ * decoding so we don't need to log the snapshot invalidation.
The alignment is not right.

/*
* CommandEndInvalidationMessages
- * Process queued-up invalidation messages at end of one command
- * in a transaction.
+ * Process queued-up invalidation messages at end of one command
+ * in a transaction.
Looks unnecessary changes.

* Note:
- * This should be called during CommandCounterIncrement(),
- * after we have advanced the command ID.
+ * This should be called during CommandCounterIncrement(),
+ * after we have advanced the command ID.
*/
Looks unnecessary changes.

if (transInvalInfo == NULL)
- return;
+ return;
Looks unnecessary changes.

+ /* prepare record */
+ memset(&xlrec, 0, sizeof(xlrec));
We should use MinSizeOfXactInvalidations, no?

--
Thanks & Regards,
Kuntal Ghosh
EnterpriseDB: http://www.enterprisedb.com