49.9. Передача больших транзакций для логического декодирования
Базовые обработчики модуля вывода (например, begin_cb, change_cb, commit_cb и message_cb) вызываются, только когда транзакция на самом деле фиксируется. Изменения в любом случае декодируются из журнала транзакций, но передаются в модуль вывода только при фиксации (и отбрасываются, если транзакция прерывается).
Это означает, что, хотя декодирование происходит последовательно и декодируемые записи могут вытесняться на диск, чтобы уменьшить нагрузку на оперативную память, все декодированные изменения должны быть переданы, когда транзакция окончательно фиксируется (или, точнее, когда фиксация декодируется из журнала транзакций). В зависимости от размера транзакции и пропускной способности сети время передачи может значительно увеличить задержку применения.
Чтобы уменьшить задержку, связанную с большими транзакциями, модуль вывода может предоставлять дополнительный обработчик для поддержки инкрементальной передачи выполняющихся транзакций. Определены несколько обязательных обработчиков передачи (stream_start_cb, stream_stop_cb, stream_abort_cb, stream_commit_cb и stream_change_cb) и два необязательных обработчика (stream_message_cb) и (stream_truncate_cb).
Когда выполняющаяся транзакция передаётся в потоке, её изменения (и сообщения) передаются блоками, границы которых обозначают вызовы stream_start_cb и stream_stop_cb. После того, как все декодированные изменения переданы, транзакция может быть зафиксирована обработчиком stream_commit_cb (или, возможно, прервана обработчиком stream_abort_cb). Если поддерживаются двухфазные фиксации, транзакция может быть подготовлена обработчиком stream_prepare_cb, зафиксирована (COMMIT PREPARED) обработчиком commit_prepared_cb или прервана обработчиком rollback_prepared_cb.
Например, последовательность вызовов обработчиков потока для одной транзакции может быть такой:
stream_start_cb(...); <-- начало первого блока изменений stream_change_cb(...); stream_change_cb(...); stream_message_cb(...); stream_change_cb(...); ... stream_change_cb(...); stream_stop_cb(...); <-- конец первого блока изменений stream_start_cb(...); <-- начало второго блока изменений stream_change_cb(...); stream_change_cb(...); stream_change_cb(...); ... stream_message_cb(...); stream_change_cb(...); stream_stop_cb(...); <-- конец второго блока изменений stream_commit_cb(...); <-- фиксация переданной транзакции
На практике же последовательность вызовов обработчика может быть более сложной. В ней могут быть блоки нескольких передаваемых в потоке транзакций, какие-то транзакции могут прерываться и т. д.
Подобно поведению вытеснения на диск, передача в потоке запускается, когда общее количество изменений, декодированных из WAL (для всех текущих транзакций), превышает предел, заданный параметром logical_decoding_work_mem. В этот момент выбирается самая большая транзакция (это определяется по объёму памяти, который уже занимают декодированные изменения) верхнего уровня, и она передаётся в потоке. Тем не менее, в некоторых случаях всё равно приходится переносить данные на диск, даже когда включена передача в потоке, поскольку порог памяти уже превышен, но полный кортеж ещё не декодирован, например декодировано добавление только в таблицу TOAST, но не в основную таблицу.
Даже при передаче в потоке больших транзакций изменения по-прежнему применяются в порядке фиксации, сохраняя те же гарантии, что и не в потоковом режиме.
49.9. Streaming of Large Transactions for Logical Decoding
The basic output plugin callbacks (e.g., begin_cb, change_cb, commit_cb and message_cb) are only invoked when the transaction actually commits. The changes are still decoded from the transaction log, but are only passed to the output plugin at commit (and discarded if the transaction aborts).
This means that while the decoding happens incrementally, and may spill to disk to keep memory usage under control, all the decoded changes have to be transmitted when the transaction finally commits (or more precisely, when the commit is decoded from the transaction log). Depending on the size of the transaction and network bandwidth, the transfer time may significantly increase the apply lag.
To reduce the apply lag caused by large transactions, an output plugin may provide additional callback to support incremental streaming of in-progress transactions. There are multiple required streaming callbacks (stream_start_cb, stream_stop_cb, stream_abort_cb, stream_commit_cb and stream_change_cb) and two optional callbacks (stream_message_cb and stream_truncate_cb).
When streaming an in-progress transaction, the changes (and messages) are streamed in blocks demarcated by stream_start_cb and stream_stop_cb callbacks. Once all the decoded changes are transmitted, the transaction can be committed using the the stream_commit_cb callback (or possibly aborted using the stream_abort_cb callback). If two-phase commits are supported, the transaction can be prepared using the stream_prepare_cb callback, COMMIT PREPARED using the commit_prepared_cb callback or aborted using the rollback_prepared_cb.
One example sequence of streaming callback calls for one transaction may look like this:
stream_start_cb(...); <-- start of first block of changes stream_change_cb(...); stream_change_cb(...); stream_message_cb(...); stream_change_cb(...); ... stream_change_cb(...); stream_stop_cb(...); <-- end of first block of changes stream_start_cb(...); <-- start of second block of changes stream_change_cb(...); stream_change_cb(...); stream_change_cb(...); ... stream_message_cb(...); stream_change_cb(...); stream_stop_cb(...); <-- end of second block of changes stream_commit_cb(...); <-- commit of the streamed transaction
The actual sequence of callback calls may be more complicated, of course. There may be blocks for multiple streamed transactions, some of the transactions may get aborted, etc.
Similar to spill-to-disk behavior, streaming is triggered when the total amount of changes decoded from the WAL (for all in-progress transactions) exceeds the limit defined by logical_decoding_work_mem setting. At that point, the largest top-level transaction (measured by the amount of memory currently used for decoded changes) is selected and streamed. However, in some cases we still have to spill to disk even if streaming is enabled because we exceed the memory threshold but still have not decoded the complete tuple e.g., only decoded toast table insert but not the main table insert.
Even when streaming large transactions, the changes are still applied in commit order, preserving the same guarantees as the non-streaming mode.
