Dear Wang, Thank you for making the patch. IIUC your patch basically can achieve that output plugin does not have to call UpdateProgress.
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. 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. 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. 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(). 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? 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? 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. 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. Best Regards, Hayato Kuroda FUJITSU LIMITED
