RE: Rework LogicalOutputPluginWriterUpdateProgress

On Thur, Feb 23, 2023 at 18:41 PM Kuroda, Hayato/\田 隼人 <kuroda(dot)hayato(at)fujitsu(dot)com> wrote:
> Dear Wang,
>
> Thank you for making the patch. IIUC your patch basically can achieve that
> output plugin
> does not have to call UpdateProgress.

Thanks for your review and comments.

> I think the basic approach is as follows, is it right?
>
> 1. In *_cb_wrapper, set ctx->did_write to false
> 2. In OutputPluginWrite() set ctx->did_write to true.
> This means that changes are really written, not skipped.
> 3. At the end of the transaction, call update_progress_and_keepalive().
> Even if we are not at the end, check skipped count and call the function if
> needed.
> The counter will be reset if ctx->did_write is true or we exceed the threshold.

Yes, you are right.
For the reset of the counter, please also refer to the reply to #05.

> Followings are my comments. I apologize if I missed some previous discussions.
>
> 01. logical.c
>
> ```
> +static void update_progress_and_keepalive(struct LogicalDecodingContext *ctx,
> + bool finished_xact);
> +
> +static bool is_skip_threshold_change(struct LogicalDecodingContext *ctx);
> ```
>
> "struct" may be not needed.

Removed.

> 02. UpdateDecodingProgressAndKeepalive
>
> I think the name should be UpdateDecodingProgressAndSendKeepalive(),
> keepalive is not verb.
> (But it's ok to ignore if you prefer the shorter name)
> Same thing can be said for the name of datatype and callback.

Yes, I prefer the shorter one. Otherwise, I think some names would be longer.

> 03. UpdateDecodingProgressAndKeepalive
>
> ```
> + /* set output state */
> + ctx->accept_writes = false;
> + ctx->write_xid = xid;
> + ctx->write_location = lsn;
> + ctx->did_write = false;
> ```
>
> Do we have to modify accept_writes, write_xid, and write_location here?
> These value is not used in WalSndUpdateProgressAndKeepalive().

I think it might be better to set these three flags.
Since LogicalOutputPluginWriterUpdateProgressAndKeepalive is an open callback, I
think setting write_xid and write_location is not just for the function
WalSndUpdateProgressAndKeepalive. And I think setting accept_writes could
prevent some wrong usage.

> 04. stream_abort_cb_wrapper
>
> ```
> + update_progress_and_keepalive(ctx, true)
> ```
>
> I'm not sure, but is it correct that call update_progress_and_keepalive() with
> finished_xact = true? Isn't there a possibility that streamed sub-transaciton is
> aborted?

Fixed.

> 05. is_skip_threshold_change
>
> At the end of the transaction, update_progress_and_keepalive() is called directly.
> Don't we have to reset change_count here?

I think this might complicate the function is_skip_threshold_change, so I didn't
reset the counter in this case.
I think the worst case of not resetting the counter is to delay sending the
keepalive message for the next transaction. But since the threshold we're using
is safe enough, it seems fine to me not to reset the counter in this case.
Added these related comments in the function is_skip_threshold_change.

> 06. ReorderBufferAbort
>
> Assuming that the top transaction is aborted. At that time
> update_progress_and_keepalive()
> is called in stream_abort_cb_wrapper(), an then
> WalSndUpdateProgressAndKeepalive()
> is called at the end of ReorderBufferAbort(). Do we have to do in both?
> I think stream_abort_cb_wrapper() may be not needed.

Yes, I think we only need one call for this case.
To make the behavior in *_cb_wrapper look consistent, I avoided the second call
for this case in the function ReorderBufferAbort.

> 07. WalSndUpdateProgress
>
> You renamed ProcessPendingWrites() to WalSndSendPending(), but it may be
> still strange
> because it will be called even if there are no pending writes.
>
> Isn't it sufficient to call ProcessRepliesIfAny(), WalSndCheckTimeOut() and
> (at least) WalSndKeepaliveIfNecessary()in the case? Or better name may be
> needed.

I think after sending the keepalive message (in WalSndKeepaliveIfNecessary), we
need to make sure the pending data is flushed through the loop.

Attach the new patch.

Regards,
Wang wei