public inbox for [email protected]
help / color / mirror / Atom feedFrom: Amit Kapila <[email protected]>
To: [email protected] <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: Perform streaming logical transactions by background workers and parallel apply
Date: Tue, 20 Dec 2022 14:41:52 +0530
Message-ID: <CAA4eK1JU4J3-Z=fcMcyQw-iG-DNr=u6w_g-iM_4xwg1pKupkFQ@mail.gmail.com> (raw)
In-Reply-To: <CAA4eK1KdVRffYEYiv9=4tVWhP=Jkfh1h6FM0A1R0NEe8cLT8+g@mail.gmail.com>
References: <CAA4eK1JQTDXTfvJ5d+L0ggG4+doyd0Xji=e0OJsb=qcn_jWALA@mail.gmail.com>
<OS0PR01MB5716D6D2765E54DC739F288E940A9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB5716997A7115715F9E4EE520940D9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1JEFVZcymteMYXhYghiX7Lb=MKuEsfxgD1VfaRrLCHyzg@mail.gmail.com>
<OS0PR01MB571663F65904D00895AD159994109@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1JuRPUY2Dx=vUToRksTmW0ptqMVT3K32g0368f_ZCk-zg@mail.gmail.com>
<OS0PR01MB5716B802A1733548A99761AE94129@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1LGKYUDFZ_jFPrU497wQf2HNvt5a+tCTpqSeWSG6kfpSA@mail.gmail.com>
<CAA4eK1Jwh7j86Egk1cye=x2R_yrTjzXGj7Fx12wVybBAEq91kA@mail.gmail.com>
<OS0PR01MB571690D6FF24E9D000ECBDBD94199@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB571639E4141EBF8A2C501E6994189@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1+ELh2vPre3JHyoeKV0A9_V7aXQD0QBPv86WEn7P_rK-g@mail.gmail.com>
<OS0PR01MB57166A377638E57823219E32941A9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KpN=dZPcCj=TLwh3wk9uvPB-M1Jrjw7wt+-L62YrDHFQ@mail.gmail.com>
<OS0PR01MB571604D401DE7A0659BD6609941D9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1LJMpBYbgDVz=g4qig8C2bM10PK=DgW7o1zn8X426vJ+Q@mail.gmail.com>
<OS0PR01MB571663B4EC5D8BE749AACD98941E9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAD21AoD_3ta_29sopuf_KJCExverP5808HjwDeqaHDrbqbxYww@mail.gmail.com>
<OS0PR01MB5716D38323B07EB0A0D097F194E09@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1Kumnc+AR86AGJ666+4Ebfny+nMbs_JZSW3O__UEiGp6w@mail.gmail.com>
<OS0PR01MB57169DAA9A2A6E68EE5E05F094E19@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAD21AoA-i_Gr1Quo2JN56ORafL=phqD=5hUvbA40EHMRZjArQA@mail.gmail.com>
<OS0PR01MB5716E94C5E503CC53177A22994E69@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KkrkAzO-=_k_6s5V1+uNUFkvxMK-33MaFw=JCw5Hqs3w@mail.gmail.com>
<CAA4eK1LRqEXAmh2ZEG-8XK_xDLu2Zzdoj86v4iNXDy++B2SBnQ@mail.gmail.com>
<OS0PR01MB5716C3561CF53027F22B47AF94E79@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KdVRffYEYiv9=4tVWhP=Jkfh1h6FM0A1R0NEe8cLT8+g@mail.gmail.com>
On Mon, Dec 19, 2022 at 6:17 PM Amit Kapila <[email protected]> wrote:
>
> On Sat, Dec 17, 2022 at 7:34 PM [email protected]
> <[email protected]> wrote:
> >
> > Agreed. I have addressed all the comments and did some cosmetic changes.
> > Attach the new version patch set.
> >
>
> Few comments:
> ============
>
Few more minor points:
1.
-static inline void
+void
changes_filename(char *path, Oid subid, TransactionId xid)
{
This function seems to be used only in worker.c. So, what is the need
to make it extern?
2. I have made a few changes in the comments. See attached. This is
atop my yesterday's top-up patch.
I think we should merge the 0001 and 0002 patches as they need to be
committed together.
--
With Regards,
Amit Kapila.
Attachments:
[application/octet-stream] changes_amit_2.patch (4.0K, ../CAA4eK1JU4J3-Z=fcMcyQw-iG-DNr=u6w_g-iM_4xwg1pKupkFQ@mail.gmail.com/2-changes_amit_2.patch)
download | inline diff:
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index df67851ac0..75119183da 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1882,7 +1882,6 @@ apply_handle_stream_abort(StringInfo s)
pa_unlock_stream(xid, AccessExclusiveLock);
(void) pa_free_worker(winfo);
}
-
break;
}
@@ -1927,8 +1926,8 @@ apply_handle_stream_abort(StringInfo s)
* We need to wait after processing rollback to savepoint for the
* next set of changes.
*
- * We have a race condition here when we can start waiting here
- * when there are more chunk of streams in the queue. See
+ * We have a race condition here due to which we can start waiting
+ * here when there are more chunk of streams in the queue. See
* apply_handle_stream_stop.
*/
if (!toplevel_xact)
@@ -4242,8 +4241,8 @@ stream_write_change(char action, StringInfo s)
* Serialize a message to a file for the given transaction.
*
* This function is similar to stream_write_change except that it will open the
- * target file if not already before writing the message and close file at the
- * end.
+ * target file if not already before writing the message and close the file at
+ * the end.
*/
static void
stream_open_and_write_change(TransactionId xid, char action, StringInfo s)
@@ -4926,10 +4925,9 @@ set_apply_error_context_origin(char *originname)
}
/*
- * Return the action to take for the given transaction. *winfo is assigned to
- * the destination parallel worker info (if the action is
- * TRANS_LEADER_SEND_TO_PARALLEL or TRANS_LEADER_PARTIAL_SERIALIZE), otherwise
- * *winfo is assigned NULL.
+ * Return the action to be taken for the given transaction. *winfo is
+ * assigned to the destination parallel worker info when the given
+ * transaction is being processed using a parallel apply worker.
*/
static TransApplyAction
get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo)
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index a362688fde..1a75686b79 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -58,10 +58,12 @@ typedef struct LogicalRepWorker
/*
* Used to create the changes and subxact files for the streaming
- * transactions. Upon the arrival of the first streaming transaction, the
- * fileset will be initialized, and it will be deleted when the worker
- * exits. Under this, separate buffiles would be created for each
- * transaction which will be deleted after the transaction is finished.
+ * transactions. Upon the arrival of the first streaming transaction or
+ * when the first-time leader apply worker times out while sending changes
+ * to the parallel apply worker, the fileset will be initialized, and it
+ * will be deleted when the worker exits. Under this, separate buffiles
+ * would be created for each transaction which will be deleted after the
+ * transaction is finished.
*/
FileSet *stream_fileset;
@@ -96,12 +98,20 @@ typedef enum ParallelTransState
} ParallelTransState;
/*
- * State of fileset in leader apply worker.
+ * State of fileset used to communicate changes from leader to parallel
+ * apply worker.
*
- * FS_SERIALIZE_IN_PROGRESS means that the leader is serializing changes to the
- * file. FS_SERIALIZE_DONE means that the leader has serialized all changes to
- * the file. FS_READY means that it is now ok for a parallel apply worker read
+ * FS_EMPTY indicates an initial state where the leader doesn't need to use
+ * the file to communicate with the parallel apply worker.
+ *
+ * FS_SERIALIZE_IN_PROGRESS indicates that the leader is serializing changes
+ * to the file.
+ *
+ * FS_SERIALIZE_DONE indicates that the leader has serialized all changes to
* the file.
+ *
+ * FS_READY indicates that it is now ok for a parallel apply worker to
+ * read the file.
*/
typedef enum PartialFileSetState
{
view thread (572+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <CAA4eK1JU4J3-Z=fcMcyQw-iG-DNr=u6w_g-iM_4xwg1pKupkFQ@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox