在 PostgreSQL 源码树的 contrib/test_decoding 子目录下包含了一个官方的完整示例输出插件。
输出插件以动态共享库（.so / .dylib / .dll）的形式提供，系统根据插件名称在标准动态库搜索路径中定位。插件必须导出一个名为 _PG_output_plugin_init 的入口函数。该函数接收一个结构体指针，开发者需在其中填入用于处理具体事件的回调函数指针：
typedef struct OutputPluginCallbacks
{
LogicalDecodeStartupCB startup_cb;
LogicalDecodeBeginCB begin_cb;
LogicalDecodeChangeCB change_cb;
LogicalDecodeTruncateCB truncate_cb;
LogicalDecodeCommitCB commit_cb;
LogicalDecodeMessageCB message_cb;
LogicalDecodeFilterByOriginCB filter_by_origin_cb;
LogicalDecodeShutdownCB shutdown_cb;
LogicalDecodeFilterPrepareCB filter_prepare_cb;
LogicalDecodeBeginPrepareCB begin_prepare_cb;
LogicalDecodePrepareCB prepare_cb;
LogicalDecodeCommitPreparedCB commit_prepared_cb;
LogicalDecodeRollbackPreparedCB rollback_prepared_cb;
LogicalDecodeStreamStartCB stream_start_cb;
LogicalDecodeStreamStopCB stream_stop_cb;
LogicalDecodeStreamAbortCB stream_abort_cb;
LogicalDecodeStreamPrepareCB stream_prepare_cb;
LogicalDecodeStreamCommitCB stream_commit_cb;
LogicalDecodeStreamChangeCB stream_change_cb;
LogicalDecodeStreamMessageCB stream_message_cb;
LogicalDecodeStreamTruncateCB stream_truncate_cb;
} OutputPluginCallbacks;
typedef void (*LogicalOutputPluginInit) (struct OutputPluginCallbacks *cb);
其中 begin_cb、change_cb 与 commit_cb 为强制必须实现的回调；startup_cb、truncate_cb、message_cb、filter_by_origin_cb 与 shutdown_cb 为可选回调。若未设置 truncate_cb，解码流将直接忽略 TRUNCATE 动作。
若插件希望支持对执行中的大事务进行流式传输，则必须实现 stream_start_cb、stream_stop_cb、stream_abort_cb、stream_commit_cb 与 stream_change_cb 回调；stream_message_cb 与 stream_truncate_cb 为可选。若同时支持两阶段提交，还必须实现 stream_prepare_cb。
若插件希望支持两阶段提交，必须实现 begin_prepare_cb、prepare_cb、commit_prepared_cb 与 rollback_prepared_cb，而 filter_prepare_cb 为可选。
为了解码、格式化并输出变更，输出插件可以使用服务端后台进程的大部分常规基础设施，包括调用数据类型的输出函数。允许以只读方式访问关系，但仅限于由 initdb 在 pg_catalog 模式下创建的关系，或者通过以下命令显式标记为用户系统目录表的关系：
ALTER TABLE user_catalog_table SET (user_catalog_table = true); CREATE TABLE another_catalog_table(data text) WITH (user_catalog_table = true);
在输出插件中访问系统目录表或用户目录表时，必须严格使用 systable_* 扫描接口，使用 heap_* 扫描接口会报错退出。此外，严禁在插件中执行任何会导致分配事务 ID 的写操作，这包括向数据表写入数据、执行 DDL 变更以及调用 pg_current_xact_id()。
输出插件回调可以向消费者传递几乎任意格式的数据。若插件仅输出当前数据库字符集编码下的纯文本字符串，可以在启动回调中将 OutputPluginOptions.output_type 设置为 OUTPUT_PLUGIN_TEXTUAL_OUTPUT 而非 OUTPUT_PLUGIN_BINARY_OUTPUT。在此模式下所有数据必须与数据库编码完全一致，以便直接作为 text 类型处理。
输出插件通过一组标准回调函数接收发生的数据变更。
并发事务严格按照其真实的提交物理顺序进行解码。在 begin 与 commit 回调之间，仅会传递属于该具体事务的变更数据。显式回滚或隐式异常中止的事务永远不会被解码。保存点（Savepoints）按照执行顺序被自动合并到所属的主事务中。
只有已安全刷写到磁盘的事务才会被逻辑解码。当 synchronous_commit 设为 off 时，紧随事务执行后的 pg_logical_slot_get_changes() 可能不会立即看到刚刚提交的事务。
每当创建复制槽或请求流式传输变更时，可选的 startup_cb 回调均会被触发调用：
typedef void (*LogicalDecodeStartupCB) (struct LogicalDecodingContext *ctx,
OutputPluginOptions *options,
bool is_init);
当复制槽初次创建时 is_init 为 true，其余情况为 false。options 允许插件设置运行参数，包括输出格式模式与是否接收堆重写变更（receive_rewrites）。启动回调还应校验传入的插件选项，并可利用 ctx->output_plugin_private 保存插件的专属私有状态。
当活跃的复制槽断开或停止流式传输时，可选的 shutdown_cb 回调会被调用，用于释放插件分配的私有资源。这并不代表复制槽被删除，仅代表会话流传输终止。
typedef void (*LogicalDecodeShutdownCB) (struct LogicalDecodingContext *ctx);
当解码出一个已提交事务的开始边界时，调用必须的 begin_cb 回调：
typedef void (*LogicalDecodeBeginCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn);
参数 txn 包含了事务的元数据，如提交时间戳与其 XID。
当事务提交被解码完成时调用必须的 commit_cb 回调。在此之前，该事务内所有涉及的行变更均已通过 change_cb 派发完毕：
typedef void (*LogicalDecodeCommitCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr commit_lsn);
对于事务内的每一条单行数据修改（无论是 INSERT、UPDATE 还是 DELETE），均会触发必须的 change_cb 回调：
typedef void (*LogicalDecodeChangeCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
Relation relation,
ReorderBufferChange *change);
逻辑解码仅能提取非不记录日志（Unlogged，参见 UNLOGGED）且非临时表（Temporary，参见 TEMPORARY or TEMP）的用户自定义表的变更。
当执行 TRUNCATE 命令时调用可选的 truncate_cb 回调：
typedef void (*LogicalDecodeTruncateCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
int nrelations,
Relation relations[],
ReorderBufferChange *change);
由于存在外键级联关联的多张表截断必须原子同步生效，该回调接收关系数组指针。详细语法说明请参考 TRUNCATE。
可选的 filter_by_origin_cb 回调用于在解码最前端核验特定复制源 origin_id 回放的数据是否需要被插件接收：
typedef bool (*LogicalDecodeFilterByOriginCB) (struct LogicalDecodingContext *ctx,
RepOriginId origin_id);
若返回 true，代表来自该节点的变更与当前插件无关，系统会将其直接过滤丢弃，从而避免多活与级联架构中的数据循环回环；若返回 false，则正常派发给后续回调处理。通过此回调过滤的效率远高于在输出插件中逐行判断。
当解码出通过逻辑解码消息机制写入的自定义负载时，触发可选的 message_cb 回调：
typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr message_lsn,
bool transactional,
const char *prefix,
Size message_size,
const char *message);
prefix 用于标识消息所属插件或命名空间，message 为长度为 message_size 的二进制或文本负载。
可选的 filter_prepare_cb 回调用于判定两阶段提交事务是应该在此预备阶段提前解码，还是推迟到最终 COMMIT PREPARED 时再作为普通单阶段事务解码：
typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx,
TransactionId xid,
const char *gid);
若返回 true 表示跳过此预备阶段的解码；返回 false 则在此阶段解码。若未定义该回调，系统默认对所有两阶段事务均执行两阶段完整解码。
当解码出预备事务的起始标记时调用必须的 begin_prepare_cb：
typedef void (*LogicalDecodeBeginPrepareCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn);
当两阶段事务已完成预备且其涉及的所有行变更已派发完毕时，调用必须的 prepare_cb 回调：
typedef void (*LogicalDecodePrepareCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn);
当解码出 COMMIT PREPARED 时调用必须的 commit_prepared_cb 回调：
typedef void (*LogicalDecodeCommitPreparedCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr commit_lsn);
当解码出 ROLLBACK PREPARED 时调用必须的 rollback_prepared_cb 回调：
typedef void (*LogicalDecodeRollbackPreparedCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr prepare_end_lsn,
TimestampTz prepare_time);
当为正在执行的大事务开启一个流式变更分块时调用：
typedef void (*LogicalDecodeStreamStartCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn);
当关闭一个流式变更分块时调用：
typedef void (*LogicalDecodeStreamStopCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn);
当中止之前已分批流式传输的事务时调用：
typedef void (*LogicalDecodeStreamAbortCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);
当作为两阶段提交的一部分对流式事务进行预备时调用：
typedef void (*LogicalDecodeStreamPrepareCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn);
当提交之前已分批流式传输的事务时调用：
typedef void (*LogicalDecodeStreamCommitCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr commit_lsn);
在流式变更分块内部发送具体单行修改时调用：
typedef void (*LogicalDecodeStreamChangeCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
Relation relation,
ReorderBufferChange *change);
在流式分块内部发送通用消息时调用：
typedef void (*LogicalDecodeStreamMessageCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr message_lsn,
bool transactional,
const char *prefix,
Size message_size,
const char *message);
在流式分块内部发送 TRUNCATE 命令时调用：
typedef void (*LogicalDecodeStreamTruncateCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
int nrelations,
Relation relations[],
ReorderBufferChange *change);
输出插件在 begin_cb、commit_cb 或 change_cb 等回调函数内部，通过向 ctx->out 中的 StringInfo 缓冲区写入数据来实际生成输出。在写入前需调用 OutputPluginPrepareWrite(ctx, last_write)，写入完毕后调用 OutputPluginWrite(ctx, last_write) 完成发送：
OutputPluginPrepareWrite(ctx, true); appendStringInfo(ctx->out, "BEGIN %u", txn->xid); OutputPluginWrite(ctx, true);