逻辑复制支持Alter table的DDL语句

contrib/sql_decoding/sql_decoding.cpp

@@ -44,6 +44,7 @@
 #include "utils/typcache.h"
 #include "replication/output_plugin.h"
 #include "replication/logical.h"
+#include "tcop/ddldeparse.h"
 
 PG_MODULE_MAGIC;
 
@@ -59,6 +60,15 @@ static void pg_decode_abort_txn(LogicalDecodingContext* ctx, ReorderBufferTXN* t
 static void pg_decode_change(
     LogicalDecodingContext* ctx, ReorderBufferTXN* txn, Relation rel, ReorderBufferChange* change);
 static bool pg_decode_filter(LogicalDecodingContext* ctx, RepOriginId origin_id);
+static void pg_decode_truncate(LogicalDecodingContext *ctx,
+                               ReorderBufferTXN *txn,
+                               int nrelations, Relation relations[],
+                               ReorderBufferChange *change);
+static void pg_decode_ddl(LogicalDecodingContext *ctx,
+                          ReorderBufferTXN *txn, XLogRecPtr message_lsn,
+                          const char *prefix, Oid relid,
+                          DeparsedCommandType cmdtype,
+                          Size sz, const char *message);
 
 typedef struct {
     MemoryContext context;
@@ -78,6 +88,8 @@ void _PG_output_plugin_init(OutputPluginCallbacks* cb)
     cb->begin_cb = pg_decode_begin_txn;
     cb->change_cb = pg_decode_change;
     cb->commit_cb = pg_decode_commit_txn;
+    cb->truncate_cb = pg_decode_truncate;
+    cb->ddl_cb = pg_decode_ddl;
     cb->abort_cb = pg_decode_abort_txn;
     cb->filter_by_origin_cb = pg_decode_filter;
     cb->shutdown_cb = pg_decode_shutdown;
@@ -507,3 +519,117 @@ static void pg_decode_change(
 
     OutputPluginWrite(ctx, true);
 }
+
+
+static void pg_decode_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
+                   int nrelations, Relation relations[], ReorderBufferChange *change)
+{
+    TestDecodingData *data;
+    MemoryContext old;
+    int            i;
+
+    data = (TestDecodingData*)ctx->output_plugin_private;
+
+    /* output BEGIN if we haven't yet */
+    if (data->skip_empty_xacts && !data->xact_wrote_changes) {
+        pg_output_begin(ctx, data, txn, false);
+    }
+    data->xact_wrote_changes = true;
+
+    /* Avoid leaking memory by using and resetting our own context */
+    old = MemoryContextSwitchTo(data->context);
+
+    OutputPluginPrepareWrite(ctx, true);
+
+    appendStringInfoString(ctx->out, "table ");
+
+    for (i = 0; i < nrelations; i++) {
+        if (i > 0)
+            appendStringInfoString(ctx->out, ", ");
+
+        appendStringInfoString(ctx->out,
+                               quote_qualified_identifier(get_namespace_name(relations[i]->rd_rel->relnamespace),
+                                                          NameStr(relations[i]->rd_rel->relname)));
+    }
+
+    appendStringInfoString(ctx->out, ": TRUNCATE:");
+
+    if (change->data.truncate.restart_seqs
+        || change->data.truncate.cascade) {
+        if (change->data.truncate.restart_seqs)
+            appendStringInfo(ctx->out, " restart_seqs");
+        if (change->data.truncate.cascade)
+            appendStringInfo(ctx->out, " cascade");
+    } else
+        appendStringInfoString(ctx->out, " (no-flags)");
+
+    MemoryContextSwitchTo(old);
+    MemoryContextReset(data->context);
+
+    OutputPluginWrite(ctx, true);
+}
+
+static char* deparse_command_type(DeparsedCommandType cmdtype)
+{
+    switch (cmdtype) {
+        case DCT_SimpleCmd:
+            return "Simple";
+        case DCT_TableDropStart:
+            return "Drop Table";
+        case DCT_TableDropEnd:
+            return "Drop Table End";
+        case DCT_TableAlter:
+            return "Alter Table";
+        case DCT_ObjectCreate:
+            return "Create Object";
+        case DCT_ObjectDrop:
+            return "Drop Object";
+        case DCT_TypeDropStart:
+            return "Drop Type";
+        case DCT_TypeDropEnd:
+            return "Drop Type End";
+        default:
+            Assert(false);
+    }
+    return NULL;
+}
+
+static void pg_decode_ddl(LogicalDecodingContext *ctx,
+                         ReorderBufferTXN *txn, XLogRecPtr message_lsn,
+                         const char *prefix, Oid relid,
+                         DeparsedCommandType cmdtype,
+                         Size sz, const char *message)
+{
+    TestDecodingData *data;
+    MemoryContext old;
+
+    data = (TestDecodingData*)ctx->output_plugin_private;
+
+    if (data->skip_empty_xacts && !data->xact_wrote_changes) {
+        pg_output_begin(ctx, data, txn, false);
+    }
+    data->xact_wrote_changes = true;
+
+    /* Avoid leaking memory by using and resetting our own context */
+    old = MemoryContextSwitchTo(data->context);
+    OutputPluginPrepareWrite(ctx, true);
+
+    appendStringInfo(ctx->out, "message: prefix %s, relid %u, cmdtype: %s, sz: %lu content: %s",
+        prefix,
+        relid,
+        deparse_command_type(cmdtype),
+        sz,
+        message);
+    if (cmdtype != DCT_TableDropStart) {
+        char *tmp = pstrdup(message);
+        char* owner = NULL;
+        char* decodestring = deparse_ddl_json_to_string(tmp, &owner);
+        appendStringInfo(ctx->out, "\ndecode to : %s, [owner %s]", decodestring, owner ? owner : "none");
+        pfree(tmp);
+    }
+    
+    MemoryContextSwitchTo(old);
+    MemoryContextReset(data->context);
+
+    OutputPluginWrite(ctx, true);
+}