Re: PATCH: logical_work_mem and logical streaming of large in-progress transactions
Dilip Kumar <dilipbalaut@gmail.com>
Commits
GET /api/v1/messages/:b64id/commits
the thread's linked commits as JSON, with link sources.
API reference →
-
Tighten the concurrent abort check during decoding.
- 2ce353fc1902 14.0 landed
-
Improve hash_create()'s API for some added robustness.
- b3817f5f7746 14.0 landed
-
Use HASH_BLOBS for xidhash.
- a1b8aa1e4eec 14.0 landed
-
Fix initialization of RelationSyncEntry for streaming transactions.
- 69bd60672af6 14.0 landed
-
Remove unused function declaration in logicalproto.h.
- ddd5f6d2609b 14.0 landed
-
Add additional tests to test streaming of in-progress transactions.
- 58b5ae9d62bd 14.0 landed
-
Fix inline marking introduced in commit 464824323e.
- ac15b499f7f9 14.0 landed
-
Add support for streaming to built-in logical replication.
- 464824323e57 14.0 landed
-
Fix the SharedFileSetUnregister API.
- 4ab77697f67a 14.0 landed
-
Fix comment in procarray.c
- 77c7267c37f7 14.0 cited
-
Suppress compiler warning in non-cassert builds.
- e942af7b8261 14.0 cited
-
Extend the BufFile interface.
- 808e13b282ef 14.0 landed
-
Mark a few logical decoding related variables with PGDLLIMPORT.
- b48cac3b10a0 14.0 landed
-
Implement streaming mode in ReorderBuffer.
- 7259736a6e5b 14.0 landed
-
Extend the logical decoding output plugin API with stream methods.
- 45fdc9738b36 14.0 landed
-
WAL Log invalidations at command end with wal_level=logical.
- c55040ccd017 14.0 landed
-
Immediately WAL-log subtransaction and top-level XID association.
- 0bead9af484c 14.0 landed
-
Allow logical replication to transfer data in binary format.
- 9de77b545313 14.0 cited
-
Only superuser can set sslcert/sslkey in postgres_fdw user mappings
- cebf9d6e6ee1 13.0 cited
-
Track statistics for spilling of changes from ReorderBuffer.
- 9290ad198b15 13.0 landed
-
Add logical_decoding_work_mem to limit ReorderBuffer memory usage.
- cec2edfa7859 13.0 landed
-
logical decoding: process ASSIGNMENT during snapshot build
- bac2fae05c77 13.0 cited
-
Emit invalidations to standby for transactions without xid.
- c6ff84b06a68 9.6.0 cited
On Mon, Dec 2, 2019 at 2:01 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:
>
> On Sun, Dec 1, 2019 at 7:58 AM Michael Paquier <michael@paquier.xyz> wrote:
> >
> > On Fri, Nov 22, 2019 at 01:18:11PM +0530, Dilip Kumar wrote:
> > > I have rebased the patch on the latest head and also fix the issue of
> > > "concurrent abort handling of the (sub)transaction." and attached as
> > > (v1-0013-Extend-handling-of-concurrent-aborts-for-streamin) along with
> > > the complete patch set. I have added the version number so that we
> > > can track the changes.
> >
> > The patch has rotten a bit and does not apply anymore. Could you
> > please send a rebased version? I have moved it to next CF, waiting on
> > author.
>
> I have rebased the patch set on the latest head.
>
I have review the patch set and here are few comments/questions
1.
+static void
+pg_decode_stream_change(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ Relation relation,
+ ReorderBufferChange *change)
+{
+ OutputPluginPrepareWrite(ctx, true);
+ appendStringInfo(ctx->out, "streaming change for TXN %u", txn->xid);
+ OutputPluginWrite(ctx, true);
+}
Should we show the tuple in the streamed change like we do for the
pg_decode_change?
2. pg_logical_slot_get_changes_guts
It recreate the decoding slot [ctx =
CreateDecodingContext(InvalidXLogRecPtr] but doesn't set the streaming
to false, should we pass a parameter to
pg_logical_slot_get_changes_guts saying whether we want streamed results or not
3.
+ XLogRecPtr prev_lsn = InvalidXLogRecPtr;
ReorderBufferChange *change;
ReorderBufferChange *specinsert = NULL;
@@ -1565,6 +1965,16 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
Relation relation = NULL;
Oid reloid;
+ /*
+ * Enforce correct ordering of changes, merged from multiple
+ * subtransactions. The changes may have the same LSN due to
+ * MULTI_INSERT xlog records.
+ */
+ if (prev_lsn != InvalidXLogRecPtr)
+ Assert(prev_lsn <= change->lsn);
+
+ prev_lsn = change->lsn;
I did not understand, how this change is relavent to this patch
4.
+ /*
+ * TOCHECK: We have to rebuild historic snapshot to be sure it includes all
+ * information about subtransactions, which could arrive after streaming start.
+ */
+ if (!txn->is_schema_sent)
+ snapshot_now = ReorderBufferCopySnap(rb, txn->base_snapshot,
+ txn, command_id);
In which case, txn->is_schema_sent will be true, because at the end of
the stream in ReorderBufferExecuteInvalidations we are always setting
it false,
so while sending next stream it will always be false. That means we
never required snapshot_now variable in ReorderBufferTXN.
5.
@@ -2299,6 +2746,23 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer
*rb, TransactionId xid,
txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
txn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
+
+ /*
+ * We read catalog changes from WAL, which are not yet sent, so
+ * invalidate current schema in order output plugin can resend
+ * schema again.
+ */
+ txn->is_schema_sent = false;
Same as point 4, during decode time it will never be true.
6.
+ /* send fields */
+ pq_sendint64(out, commit_lsn);
+ pq_sendint64(out, txn->end_lsn);
+ pq_sendint64(out, txn->commit_time);
Commit_time and end_lsn is used in standby_feedback
7.
+ /* FIXME optimize the search by bsearch on sorted data */
+ for (i = nsubxacts; i > 0; i--)
+ {
+ if (subxacts[i - 1].xid == subxid)
+ {
+ subidx = (i - 1);
+ found = true;
+ break;
+ }
+ }
We can not rollback intermediate subtransaction without rollbacking
latest sub-transaction, so why do we need
to search in the array? It will always be the the last subxact no?
8.
+ /*
+ * send feedback to upstream
+ *
+ * XXX Probably should send a valid LSN. But which one?
+ */
+ send_feedback(InvalidXLogRecPtr, false, false);
Why feedback is sent for every change?
--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com