On Sat, Nov 21, 2020 at 12:23 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
>
> On Fri, Nov 20, 2020 at 11:36 AM Dilip Kumar <dilipbalaut@gmail.com> wrote:
> >
> > On Fri, Nov 20, 2020 at 11:22 AM Dilip Kumar <dilipbalaut@gmail.com> wrote:
> > >
> > > On Fri, Nov 20, 2020 at 11:17 AM Amit Kapila <amit.kapila16@gmail.com> wrote:
> > > >
> > > >
> > > > And what about apply_handle_stream_abort? And, I guess we need to
> > > > avoid other related things like update of
> > > > replorigin_session_origin_lsn, replorigin_session_origin_timestamp,
> > > > etc. in apply_handle_stream_commit() as we are apply_handle_commit().
> > >
> > > Yes, we need to change these as well. I have tested using the POC
> > > patch and working fine. I will send the patch after some more
> > > testing.
> > > >
> > > > I think it is difficult to have a reliable test case for this but feel
> > > > free to propose if you have any ideas on the same.
> > >
> > > I am not sure how to write an automated test case for this.
> >
> > Here is the patch.
> >
>
> Few comments:
> ==============
> 1.
> + /* The synchronization worker runs in single transaction. */
> + if (!am_tablesync_worker())
> + {
> + /*
> + * Update origin state so we can restart streaming from correct position
> + * in case of crash.
> + */
> + replorigin_session_origin_lsn = commit_data.end_lsn;
> + replorigin_session_origin_timestamp = commit_data.committime;
> + CommitTransactionCommand();
> + pgstat_report_stat(false);
> + store_flush_position(commit_data.end_lsn);
> + }
> + else
> + {
> + /* Process any invalidation messages that might have accumulated. */
> + AcceptInvalidationMessages();
> + maybe_reread_subscription();
> + }
>
> Here, why the check IsTransactionState() is not there similar to
> apply_handle_commit? Also, this code looks the same as in
> apply_handle_commit(), can we move this into a common function say
> apply_handle_commit_internal or something like that? If we decide to
> do so then we can move a few more things as mentioned below in the
> common function:
Moved to the common function as suggested.
> apply_handle_commit()
> {
> ..
> in_remote_transaction = false;
>
> /* Process any tables that are being synchronized in parallel. */
> process_syncing_tables(commit_data.end_lsn);
> ..
> }
This as well.
> 2.
> @@ -902,7 +906,9 @@ apply_handle_stream_abort(StringInfo s)
> {
> /* Cleanup the subxact info */
> cleanup_subxact_info();
> - CommitTransactionCommand();
> +
> + if (!am_tablesync_worker())
> + CommitTransactionCommand();
>
> Here, also you can add a comment: "/* The synchronization worker runs
> in single transaction. */"
>
Done
--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com