Re: Perform streaming logical transactions by background workers and parallel apply - Mailing list pgsql-hackers

On Tue, Dec 27, 2022 at 11:28 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
>
> On Mon, Dec 26, 2022 at 10:29 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
> >
> > On Mon, Dec 26, 2022 at 6:33 PM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
> > >
> > > ---
> > > +        if (!pa_can_start(xid))
> > > +                return;
> > > +
> > > +        /* First time through, initialize parallel apply worker state
> > > hashtable. */
> > > +        if (!ParallelApplyTxnHash)
> > > +        {
> > > +                HASHCTL                ctl;
> > > +
> > > +                MemSet(&ctl, 0, sizeof(ctl));
> > > +                ctl.keysize = sizeof(TransactionId);
> > > +                ctl.entrysize = sizeof(ParallelApplyWorkerEntry);
> > > +                ctl.hcxt = ApplyContext;
> > > +
> > > +                ParallelApplyTxnHash = hash_create("logical
> > > replication parallel apply workershash",
> > > +
> > >              16, &ctl,
> > > +
> > >              HASH_ELEM |HASH_BLOBS | HASH_CONTEXT);
> > > +        }
> > > +
> > > +        /*
> > > +         * It's necessary to reread the subscription information
> > > before assigning
> > > +         * the transaction to a parallel apply worker. Otherwise, the
> > > leader may
> > > +         * not be able to reread the subscription information if streaming
> > > +         * transactions keep coming and are handled by parallel apply workers.
> > > +         */
> > > +        maybe_reread_subscription();
> > >
> > > pa_can_start() checks if the skiplsn is an invalid xid or not, and
> > > then maybe_reread_subscription() could update the skiplsn to a valid
> > > value. As the comments in pa_can_start() says, it won't work. I think
> > > we should call maybe_reread_subscription() in
> > > apply_handle_stream_start() before calling pa_allocate_worker().
> > >
> >
> > But I think a similar thing can happen when we start the worker and
> > then before the transaction ends, we do maybe_reread_subscription().
>
> Where do we do maybe_reread_subscription() in this case? IIUC if the
> leader sends all changes to the worker, there is no chance for the
> leader to do maybe_reread_subscription except for when waiting for the
> input.

Yes, this is the point where it can happen. IT can happen when there
is some delay between different streaming chunks.

> On reflection, adding maybe_reread_subscription() to
> apply_handle_stream_start() adds one extra call of it so it's not
> good. Alternatively, we can do that in pa_can_start() before checking
> the skiplsn. I think we do a similar thing in AllTablesyncsRead() --
> update the information before the check if necessary.
>
> > I think we should try to call maybe_reread_subscription() when we are
> > reasonably sure that we are going to enter parallel mode, otherwise,
> > anyway, it will be later called by the leader worker.
>
> It isn't a big problem even if we update the skiplsn after launching a
> worker since we will skip the transaction the next time. But it would
> be more consistent with the current behavior. As I mentioned above,
> doing it in pa_can_start() seems to be reasonable to me. What do you
> think?
>

Okay, we can do it in pa_can_start but then let's do it before we
check the parallel_apply flag as that can also be changed if the
streaming mode is changed. Please see the changes in the attached
patch which is atop the 0001 and 0002 patches. I have made a few
comment improvements as well.

-- 
With Regards,
Amit Kapila.