external
/
postgres-decoderbufs
mirror of https://github.com/debezium/postgres-decoderbufs.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

First working version for debug text output at least

pull/1/head
Xavier Stevens 2014-09-22 16:21:47 -07:00
parent 0241c144fc
commit 6a212e9e4c
5 changed files with 425 additions and 283 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
.clang-format Normal file
View File

@ -0,0 +1,47 @@
---
# BasedOnStyle: Mozilla
AccessModifierOffset: -2
ConstructorInitializerIndentWidth: 4
AlignEscapedNewlinesLeft: false
AlignTrailingComments: true
AllowAllParametersOfDeclarationOnNextLine: false
AllowShortIfStatementsOnASingleLine: false
AllowShortLoopsOnASingleLine: false
AlwaysBreakTemplateDeclarations: false
AlwaysBreakBeforeMultilineStrings: false
BreakBeforeBinaryOperators: false
BreakBeforeTernaryOperators: true
BreakConstructorInitializersBeforeComma: false
BinPackParameters: true
ColumnLimit: 80
ConstructorInitializerAllOnOneLineOrOnePerLine: true
DerivePointerBinding: true
ExperimentalAutoDetectBinPacking: false
IndentCaseLabels: true
MaxEmptyLinesToKeep: 1
NamespaceIndentation: None
ObjCSpaceBeforeProtocolList: false
PenaltyBreakBeforeFirstCallParameter: 19
PenaltyBreakComment: 60
PenaltyBreakString: 1000
PenaltyBreakFirstLessLess: 120
PenaltyExcessCharacter: 1000000
PenaltyReturnTypeOnItsOwnLine: 200
PointerBindsToType: true
SpacesBeforeTrailingComments: 1
Cpp11BracedListStyle: false
Standard: Cpp03
IndentWidth: 2
TabWidth: 8
UseTab: Never
BreakBeforeBraces: Attach
IndentFunctionDeclarationAfterType: false
SpacesInParentheses: false
SpacesInAngles: false
SpaceInEmptyParentheses: false
SpacesInCStyleCastParentheses: false
SpaceAfterControlStatementKeyword: true
SpaceBeforeAssignmentOperators: true
ContinuationIndentWidth: 4
...

13
proto/pg_logicaldec.proto
View File

@ -22,11 +22,10 @@ message DatumMessage {
optional bytes datum_bytes = 9; optional bytes datum_bytes = 9;
} }
message TxnMessage { message RowMessage {
optional sint64 timestamp = 1; optional sint64 commit_time = 1;
optional int64 xid = 2; optional string table = 2;
optional string table = 3; optional Op op = 3;
optional Op op = 4; repeated DatumMessage new_tuple = 4;
optional DatumMessage new_datum = 5; repeated DatumMessage old_tuple = 5;
optional DatumMessage old_datum = 6;
} }

281
src/decoderbufs.c
View File

@ -24,6 +24,9 @@
* SOFTWARE. * SOFTWARE.
*/ */
#include <inttypes.h>
#include <stdarg.h>
#include "postgres.h" #include "postgres.h"
#include "funcapi.h" #include "funcapi.h"
#include "catalog/pg_class.h" #include "catalog/pg_class.h"
@ -43,9 +46,19 @@
PG_MODULE_MAGIC; PG_MODULE_MAGIC;
/* define a time macro to convert TimestampTz into something more sane,
* which in this case is microseconds since epoch
*/
#ifdef HAVE_INT64_TIMESTAMP
#define TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(t) \
t + ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY * USECS_PER_SEC);
#else
#define TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(t) \
(t + ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY)) * 1000.0;
#endif
typedef struct { typedef struct {
MemoryContext context; MemoryContext context;
Decoderbufs__TxnMessage *txn_msg;
bool debug_mode; bool debug_mode;
} DecoderData; } DecoderData;
@ -82,12 +95,11 @@ static void pg_decode_startup(LogicalDecodingContext *ctx,
DecoderData *data; DecoderData *data;
data = palloc(sizeof(DecoderData)); data = palloc(sizeof(DecoderData));
data->context = AllocSetContextCreate(ctx->context, "decoderbufs context", data->context = AllocSetContextCreate(
ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ctx->context, "decoderbufs context", ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_MAXSIZE); ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE);
ctx->output_plugin_private = data; ctx->output_plugin_private = data;
opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT; opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
foreach(option, ctx->output_plugin_options) { foreach(option, ctx->output_plugin_options) {
@ -95,16 +107,15 @@ static void pg_decode_startup(LogicalDecodingContext *ctx,
Assert(elem->arg == NULL || IsA(elem->arg, String)); Assert(elem->arg == NULL || IsA(elem->arg, String));
if (strcmp(elem->defname, "debug-mode") == 0) { if (strcmp(elem->defname, "debug-mode") == 0) {
bool debug_mode;
if (elem->arg == NULL) if (elem->arg == NULL)
debug_mode = false; data->debug_mode = false;
else if (!parse_bool(strVal(elem->arg), &debug_mode)) else if (!parse_bool(strVal(elem->arg), &data->debug_mode))
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE), (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("could not parse value \"%s\" for parameter \"%s\"", errmsg("could not parse value \"%s\" for parameter \"%s\"",
strVal(elem->arg), elem->defname))); strVal(elem->arg), elem->defname)));
if (debug_mode) if (data->debug_mode)
opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT; opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
} else { } else {
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
@ -114,25 +125,6 @@ static void pg_decode_startup(LogicalDecodingContext *ctx,
} }
} }
static void free_txn_msg_datums(Decoderbufs__TxnMessage *msg) {
if (msg->new_datum) {
if (msg->new_datum->has_datum_bytes) {
pfree(msg->new_datum->datum_bytes.data);
msg->new_datum->datum_bytes.data = NULL;
msg->new_datum->datum_bytes.len = 0;
}
pfree(msg->new_datum);
}
if (msg->old_datum) {
if (msg->old_datum->has_datum_bytes) {
pfree(msg->old_datum->datum_bytes.data);
msg->old_datum->datum_bytes.data = NULL;
msg->old_datum->datum_bytes.len = 0;
}
pfree(msg->old_datum);
}
}
/* cleanup this plugin's resources */ /* cleanup this plugin's resources */
static void pg_decode_shutdown(LogicalDecodingContext *ctx) { static void pg_decode_shutdown(LogicalDecodingContext *ctx) {
DecoderData *data = ctx->output_plugin_private; DecoderData *data = ctx->output_plugin_private;
@ -144,28 +136,85 @@ static void pg_decode_shutdown(LogicalDecodingContext *ctx) {
/* BEGIN callback */ /* BEGIN callback */
static void pg_decode_begin_txn(LogicalDecodingContext *ctx, static void pg_decode_begin_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn) { ReorderBufferTXN *txn) {
DecoderData *data = ctx->output_plugin_private;
Decoderbufs__TxnMessage msg = DECODERBUFS__TXN_MESSAGE__INIT;
data->txn_msg = &msg;
data->txn_msg->xid = txn->xid;
} }
/* COMMIT callback */ /* COMMIT callback */
static void pg_decode_commit_txn(LogicalDecodingContext *ctx, static void pg_decode_commit_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn, XLogRecPtr commit_lsn) { ReorderBufferTXN *txn, XLogRecPtr commit_lsn) {
DecoderData *data = ctx->output_plugin_private; }
Decoderbufs__TxnMessage *msg = data->txn_msg;
msg->timestamp = txn->commit_time;
OutputPluginPrepareWrite(ctx, true); /* convenience method to free up sub-messages */
size_t psize = decoderbufs__txn_message__get_packed_size(msg); static void free_row_msg_subs(Decoderbufs__RowMessage *msg) {
void *packed = palloc(psize); if (!msg) {
size_t ssize = decoderbufs__txn_message__pack(msg, packed); return;
appendBinaryStringInfo(ctx->out, packed, ssize); }
OutputPluginWrite(ctx, true);
pfree(packed); pfree(msg->table);
free_txn_msg_datums(msg); if (msg->n_new_tuple > 0) {
for (int i=0; i < msg->n_new_tuple; i++) {
if (msg->new_tuple[i]) {
if (msg->new_tuple[i]->datum_string) {
pfree(msg->new_tuple[i]->datum_string);
} else if (msg->new_tuple[i]->has_datum_bytes) {
pfree(msg->new_tuple[i]->datum_bytes.data);
msg->new_tuple[i]->datum_bytes.data = NULL;
msg->new_tuple[i]->datum_bytes.len = 0;
}
pfree(msg->new_tuple[i]);
}
}
pfree(msg->new_tuple);
}
if (msg->n_old_tuple > 0) {
for (int i=0; i < msg->n_old_tuple; i++) {
if (msg->old_tuple[i]) {
if (msg->old_tuple[i]->datum_string) {
pfree(msg->old_tuple[i]->datum_string);
} else if (msg->old_tuple[i]->has_datum_bytes) {
pfree(msg->old_tuple[i]->datum_bytes.data);
msg->old_tuple[i]->datum_bytes.data = NULL;
msg->old_tuple[i]->datum_bytes.len = 0;
}
pfree(msg->old_tuple[i]);
}
}
pfree(msg->old_tuple);
}
}
static void print_tuple_msg(StringInfo out, Decoderbufs__DatumMessage **tup, size_t n) {
if (tup) {
for (int i=0; i < n; i++) {
Decoderbufs__DatumMessage *dmsg = tup[i];
if (dmsg->column_name)
appendStringInfo(out, "column_name[%s]", dmsg->column_name);
if (dmsg->has_column_type) {
appendStringInfo(out, ", column_type[%" PRId64 "]", dmsg->column_type);
switch (dmsg->column_type) {
case INT2OID:
case INT4OID:
appendStringInfo(out, ", datum[%d]", dmsg->datum_int32);
break;
case INT8OID:
appendStringInfo(out, ", datum[%" PRId64 "]", dmsg->datum_int64);
break;
case FLOAT4OID:
appendStringInfo(out, ", datum[%f]", dmsg->datum_float);
break;
case FLOAT8OID:
case NUMERICOID:
appendStringInfo(out, ", datum[%f]", dmsg->datum_double);
break;
case TEXTOID:
appendStringInfo(out, ", datum[%s]", dmsg->datum_string);
break;
default:
break;
}
appendStringInfo(out, "\n");
}
}
}
} }
/* this doesn't seem to be available in the public api (unfortunate) */ /* this doesn't seem to be available in the public api (unfortunate) */
@ -174,13 +223,11 @@ static double numeric_to_double_no_overflow(Numeric num) {
double val; double val;
char *endptr; char *endptr;
tmp = DatumGetCString(DirectFunctionCall1(numeric_out, tmp = DatumGetCString(DirectFunctionCall1(numeric_out, NumericGetDatum(num)));
NumericGetDatum(num)));
/* unlike float8in, we ignore ERANGE from strtod */ /* unlike float8in, we ignore ERANGE from strtod */
val = strtod(tmp, &endptr); val = strtod(tmp, &endptr);
if (*endptr != '\0') if (*endptr != '\0') {
{
/* shouldn't happen ... */ /* shouldn't happen ... */
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
@ -193,7 +240,8 @@ static double numeric_to_double_no_overflow(Numeric num) {
return val; return val;
} }
static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid, Oid typoutput, Datum datum) { static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid,
Oid typoutput, Datum datum) {
Numeric num; Numeric num;
char c; char c;
bytea *valptr; bytea *valptr;
@ -201,28 +249,34 @@ static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid, Oid
switch (typid) { switch (typid) {
case BOOLOID: case BOOLOID:
datum_msg->datum_bool = DatumGetBool(datum); datum_msg->datum_bool = DatumGetBool(datum);
datum_msg->has_datum_bool = true;
break; break;
case INT2OID: case INT2OID:
datum_msg->datum_int32 = DatumGetInt16(datum); datum_msg->datum_int32 = DatumGetInt16(datum);
datum_msg->has_datum_int32 = true;
break; break;
case INT4OID: case INT4OID:
datum_msg->datum_int32 = DatumGetInt32(datum); datum_msg->datum_int32 = DatumGetInt32(datum);
datum_msg->has_datum_int32 = true;
break; break;
case INT8OID: case INT8OID:
datum_msg->datum_int64 = DatumGetInt64(datum);
break;
case OIDOID: case OIDOID:
datum_msg->datum_int64 = DatumGetInt64(datum);
datum_msg->has_datum_int64 = true;
break; break;
case FLOAT4OID: case FLOAT4OID:
datum_msg->datum_float = DatumGetFloat4(datum); datum_msg->datum_float = DatumGetFloat4(datum);
datum_msg->has_datum_float = true;
break; break;
case FLOAT8OID: case FLOAT8OID:
datum_msg->datum_double = DatumGetFloat8(datum); datum_msg->datum_double = DatumGetFloat8(datum);
datum_msg->has_datum_double = true;
break; break;
case NUMERICOID: case NUMERICOID:
num = DatumGetNumeric(datum); num = DatumGetNumeric(datum);
if (!numeric_is_nan(num)) { if (!numeric_is_nan(num)) {
datum_msg->datum_double = numeric_to_double_no_overflow(num); datum_msg->datum_double = numeric_to_double_no_overflow(num);
datum_msg->has_datum_double = true;
} }
break; break;
case CHAROID: case CHAROID:
@ -231,30 +285,34 @@ static void set_datum_value(Decoderbufs__DatumMessage *datum_msg, Oid typid, Oid
break; break;
case VARCHAROID: case VARCHAROID:
case TEXTOID: case TEXTOID:
datum_msg->datum_string = DatumGetCString(datum); output = OidOutputFunctionCall(typoutput, datum);
datum_msg->datum_string = pnstrdup(output, strlen(output));
break; break;
case BYTEAOID: case BYTEAOID:
valptr = DatumGetByteaPCopy(datum); valptr = DatumGetByteaPCopy(datum);
int size = VARSIZE(valptr); int size = VARSIZE(valptr);
datum_msg->datum_bytes = *((ProtobufCBinaryData *)palloc(sizeof(ProtobufCBinaryData))); datum_msg->datum_bytes =
*((ProtobufCBinaryData *)palloc(sizeof(ProtobufCBinaryData)));
datum_msg->datum_bytes.data = (uint8_t *)VARDATA(valptr); datum_msg->datum_bytes.data = (uint8_t *)VARDATA(valptr);
datum_msg->datum_bytes.len = size - VARHDRSZ; datum_msg->datum_bytes.len = size - VARHDRSZ;
datum_msg->has_datum_bytes = true;
break; break;
default: default:
output = OidOutputFunctionCall(typoutput, datum); output = OidOutputFunctionCall(typoutput, datum);
datum_msg->datum_bytes = *((ProtobufCBinaryData *)palloc(sizeof(ProtobufCBinaryData))); datum_msg->datum_bytes =
*((ProtobufCBinaryData *)palloc(sizeof(ProtobufCBinaryData)));
datum_msg->datum_bytes.data = (uint8_t *)output; datum_msg->datum_bytes.data = (uint8_t *)output;
datum_msg->datum_bytes.len = sizeof(output); datum_msg->datum_bytes.len = sizeof(output);
datum_msg->has_datum_bytes = true;
break; break;
} }
} }
static Decoderbufs__DatumMessage tuple_to_datum_msg(Relation relation, HeapTuple tuple) { static void tuple_to_tuple_msg(Decoderbufs__DatumMessage **tmsg, Relation relation,
TupleDesc tupdesc = RelationGetDescr(relation); HeapTuple tuple, TupleDesc tupdesc) {
int natt; int natt;
Decoderbufs__DatumMessage datum_msg = DECODERBUFS__DATUM_MESSAGE__INIT;
/* Build column names and values */ /* build column names and values */
for (natt = 0; natt < tupdesc->natts; natt++) { for (natt = 0; natt < tupdesc->natts; natt++) {
Form_pg_attribute attr; Form_pg_attribute attr;
Datum origval; Datum origval;
@ -262,18 +320,23 @@ static Decoderbufs__DatumMessage tuple_to_datum_msg(Relation relation, HeapTuple
attr = tupdesc->attrs[natt]; attr = tupdesc->attrs[natt];
/* Skip dropped columns and system columns */ /* skip dropped columns and system columns */
if (attr->attisdropped || attr->attnum < 0) if (attr->attisdropped || attr->attnum < 0) {
continue; continue;
}
/* Set the column name */ Decoderbufs__DatumMessage datum_msg = DECODERBUFS__DATUM_MESSAGE__INIT;
datum_msg.column_name = quote_identifier(NameStr(attr->attname));
/* Get Datum from tuple */ /* set the column name */
const char *col_name = quote_identifier(NameStr(attr->attname));
datum_msg.column_name = col_name;
/* set datum from tuple */
origval = fastgetattr(tuple, natt + 1, tupdesc, &isnull); origval = fastgetattr(tuple, natt + 1, tupdesc, &isnull);
/* Get output function */ /* get output function */
datum_msg.column_type = attr->atttypid; datum_msg.column_type = attr->atttypid;
datum_msg.has_column_type = true;
Oid typoutput; Oid typoutput;
bool typisvarlena; bool typisvarlena;
@ -281,7 +344,8 @@ static Decoderbufs__DatumMessage tuple_to_datum_msg(Relation relation, HeapTuple
getTypeOutputInfo(attr->atttypid, &typoutput, &typisvarlena); getTypeOutputInfo(attr->atttypid, &typoutput, &typisvarlena);
if (!isnull) { if (!isnull) {
if (typisvarlena && VARATT_IS_EXTERNAL_ONDISK(origval)) { if (typisvarlena && VARATT_IS_EXTERNAL_ONDISK(origval)) {
// what to do if anything? // TODO: Is there a way we can handle this?
elog(WARNING, "Not handling external on disk varlena at the moment.");
} else if (!typisvarlena) { } else if (!typisvarlena) {
set_datum_value(&datum_msg, attr->atttypid, typoutput, origval); set_datum_value(&datum_msg, attr->atttypid, typoutput, origval);
} else { } else {
@ -289,9 +353,11 @@ static Decoderbufs__DatumMessage tuple_to_datum_msg(Relation relation, HeapTuple
set_datum_value(&datum_msg, attr->atttypid, typoutput, val); set_datum_value(&datum_msg, attr->atttypid, typoutput, val);
} }
} }
tmsg[natt] = palloc(sizeof(datum_msg));
memcpy(tmsg[natt], &datum_msg, sizeof(datum_msg));
} }
return datum_msg;
} }
/* /*
@ -305,6 +371,7 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Form_pg_class class_form; Form_pg_class class_form;
char replident = relation->rd_rel->relreplident; char replident = relation->rd_rel->relreplident;
bool is_rel_non_selective; bool is_rel_non_selective;
Decoderbufs__RowMessage rmsg = DECODERBUFS__ROW_MESSAGE__INIT;
class_form = RelationGetForm(relation); class_form = RelationGetForm(relation);
data = ctx->output_plugin_private; data = ctx->output_plugin_private;
@ -318,44 +385,51 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
!OidIsValid(relation->rd_replidindex))); !OidIsValid(relation->rd_replidindex)));
/* set common fields */ /* set common fields */
data->txn_msg->table = quote_qualified_identifier( rmsg.commit_time = TIMESTAMPTZ_TO_USEC_SINCE_EPOCH(txn->commit_time);
get_namespace_name( rmsg.has_commit_time = true;
get_rel_namespace(RelationGetRelid(relation))), rmsg.table = quote_qualified_identifier(
get_namespace_name(get_rel_namespace(RelationGetRelid(relation))),
NameStr(class_form->relname)); NameStr(class_form->relname));
/* decode different operation types */ /* decode different operation types */
switch (change->action) { switch (change->action) {
case REORDER_BUFFER_CHANGE_INSERT: case REORDER_BUFFER_CHANGE_INSERT:
data->txn_msg->op = DECODERBUFS__OP__INSERT; rmsg.op = DECODERBUFS__OP__INSERT;
rmsg.has_op = true;
if (change->data.tp.newtuple != NULL) { if (change->data.tp.newtuple != NULL) {
HeapTupleGetDatum(&change->data.tp.newtuple->tuple); TupleDesc tupdesc = RelationGetDescr(relation);
Decoderbufs__DatumMessage new_datum = tuple_to_datum_msg(relation, rmsg.n_new_tuple = tupdesc->natts;
&change->data.tp.newtuple->tuple); rmsg.new_tuple = palloc(sizeof(Decoderbufs__DatumMessage) * tupdesc->natts);
data->txn_msg->new_datum = &new_datum; tuple_to_tuple_msg(rmsg.new_tuple, relation, &change->data.tp.newtuple->tuple, tupdesc);
} }
break; break;
case REORDER_BUFFER_CHANGE_UPDATE: case REORDER_BUFFER_CHANGE_UPDATE:
data->txn_msg->op = DECODERBUFS__OP__UPDATE; rmsg.op = DECODERBUFS__OP__UPDATE;
if (is_rel_non_selective) { rmsg.has_op = true;
if (!is_rel_non_selective) {
if (change->data.tp.oldtuple != NULL) { if (change->data.tp.oldtuple != NULL) {
Decoderbufs__DatumMessage old_datum = tuple_to_datum_msg(relation, TupleDesc tupdesc = RelationGetDescr(relation);
&change->data.tp.oldtuple->tuple); rmsg.n_old_tuple = tupdesc->natts;
data->txn_msg->old_datum = &old_datum; rmsg.old_tuple = palloc(sizeof(Decoderbufs__DatumMessage) * tupdesc->natts);
tuple_to_tuple_msg(rmsg.old_tuple, relation, &change->data.tp.oldtuple->tuple, tupdesc);
} }
if (change->data.tp.newtuple != NULL) { if (change->data.tp.newtuple != NULL) {
Decoderbufs__DatumMessage new_datum = tuple_to_datum_msg(relation, TupleDesc tupdesc = RelationGetDescr(relation);
&change->data.tp.newtuple->tuple); rmsg.n_new_tuple = tupdesc->natts;
data->txn_msg->new_datum = &new_datum; rmsg.new_tuple = palloc(sizeof(Decoderbufs__DatumMessage) * tupdesc->natts);
tuple_to_tuple_msg(rmsg.new_tuple, relation, &change->data.tp.newtuple->tuple, tupdesc);
} }
} }
break; break;
case REORDER_BUFFER_CHANGE_DELETE: case REORDER_BUFFER_CHANGE_DELETE:
data->txn_msg->op = DECODERBUFS__OP__DELETE; rmsg.op = DECODERBUFS__OP__DELETE;
rmsg.has_op = true;
/* if there was no PK, we only know that a delete happened */ /* if there was no PK, we only know that a delete happened */
if (is_rel_non_selective && change->data.tp.oldtuple != NULL) { if (!is_rel_non_selective && change->data.tp.oldtuple != NULL) {
Decoderbufs__DatumMessage old_datum = tuple_to_datum_msg(relation, TupleDesc tupdesc = RelationGetDescr(relation);
&change->data.tp.oldtuple->tuple); rmsg.n_old_tuple = tupdesc->natts;
data->txn_msg->old_datum = &old_datum; rmsg.old_tuple = palloc(sizeof(Decoderbufs__DatumMessage) * tupdesc->natts);
tuple_to_tuple_msg(rmsg.old_tuple, relation, &change->data.tp.oldtuple->tuple, tupdesc);
} }
break; break;
default: default:
@ -363,6 +437,41 @@ static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
break; break;
} }
if (data->debug_mode) {
OutputPluginPrepareWrite(ctx, true);
if (rmsg.has_commit_time)
appendStringInfo(ctx->out, "commit_time[%" PRId64 "]", rmsg.commit_time);
if (rmsg.table)
appendStringInfo(ctx->out, ", table[%s]", rmsg.table);
if (rmsg.has_op)
appendStringInfo(ctx->out, ", op[%d]", rmsg.op);
if (rmsg.old_tuple) {
appendStringInfo(ctx->out, "\nOLD TUPLE: \n");
print_tuple_msg(ctx->out, rmsg.old_tuple, rmsg.n_old_tuple);
appendStringInfo(ctx->out, "\n");
}
if (rmsg.new_tuple) {
appendStringInfo(ctx->out, "\nNEW TUPLE: \n");
print_tuple_msg(ctx->out, rmsg.new_tuple, rmsg.n_new_tuple);
appendStringInfo(ctx->out, "\n");
}
OutputPluginWrite(ctx, true);
} else {
OutputPluginPrepareWrite(ctx, true);
size_t psize = decoderbufs__row_message__get_packed_size(&rmsg);
void *packed = palloc(psize);
size_t ssize = decoderbufs__row_message__pack(&rmsg, packed);
// appendBinaryStringInfo(ctx->out, (void *)psize, sizeof(psize));
appendBinaryStringInfo(ctx->out, packed, ssize);
OutputPluginWrite(ctx, true);
/* free packed buffer */
pfree(packed);
}
/* cleanup msg */
free_row_msg_subs(&rmsg);
MemoryContextSwitchTo(old); MemoryContextSwitchTo(old);
MemoryContextReset(data->context); MemoryContextReset(data->context);
} }

127
src/proto/pg_logicaldec.pb-c.c
View File

@ -6,7 +6,7 @@
#define PROTOBUF_C__NO_DEPRECATED #define PROTOBUF_C__NO_DEPRECATED
#endif #endif
#include "proto/pg_logicaldec.pb-c.h" #include "pg_logicaldec.pb-c.h"
void decoderbufs__datum_message__init void decoderbufs__datum_message__init
(Decoderbufs__DatumMessage *message) (Decoderbufs__DatumMessage *message)
{ {
@ -50,47 +50,47 @@ void decoderbufs__datum_message__free_unpacked
assert(message->base.descriptor == &decoderbufs__datum_message__descriptor); assert(message->base.descriptor == &decoderbufs__datum_message__descriptor);
protobuf_c_message_free_unpacked ((ProtobufCMessage*)message, allocator); protobuf_c_message_free_unpacked ((ProtobufCMessage*)message, allocator);
} }
void decoderbufs__txn_message__init void decoderbufs__row_message__init
(Decoderbufs__TxnMessage *message) (Decoderbufs__RowMessage *message)
{ {
static Decoderbufs__TxnMessage init_value = DECODERBUFS__TXN_MESSAGE__INIT; static Decoderbufs__RowMessage init_value = DECODERBUFS__ROW_MESSAGE__INIT;
*message = init_value; *message = init_value;
} }
size_t decoderbufs__txn_message__get_packed_size size_t decoderbufs__row_message__get_packed_size
(const Decoderbufs__TxnMessage *message) (const Decoderbufs__RowMessage *message)
{ {
assert(message->base.descriptor == &decoderbufs__txn_message__descriptor); assert(message->base.descriptor == &decoderbufs__row_message__descriptor);
return protobuf_c_message_get_packed_size ((const ProtobufCMessage*)(message)); return protobuf_c_message_get_packed_size ((const ProtobufCMessage*)(message));
} }
size_t decoderbufs__txn_message__pack size_t decoderbufs__row_message__pack
(const Decoderbufs__TxnMessage *message, (const Decoderbufs__RowMessage *message,
uint8_t *out) uint8_t *out)
{ {
assert(message->base.descriptor == &decoderbufs__txn_message__descriptor); assert(message->base.descriptor == &decoderbufs__row_message__descriptor);
return protobuf_c_message_pack ((const ProtobufCMessage*)message, out); return protobuf_c_message_pack ((const ProtobufCMessage*)message, out);
} }
size_t decoderbufs__txn_message__pack_to_buffer size_t decoderbufs__row_message__pack_to_buffer
(const Decoderbufs__TxnMessage *message, (const Decoderbufs__RowMessage *message,
ProtobufCBuffer *buffer) ProtobufCBuffer *buffer)
{ {
assert(message->base.descriptor == &decoderbufs__txn_message__descriptor); assert(message->base.descriptor == &decoderbufs__row_message__descriptor);
return protobuf_c_message_pack_to_buffer ((const ProtobufCMessage*)message, buffer); return protobuf_c_message_pack_to_buffer ((const ProtobufCMessage*)message, buffer);
} }
Decoderbufs__TxnMessage * Decoderbufs__RowMessage *
decoderbufs__txn_message__unpack decoderbufs__row_message__unpack
(ProtobufCAllocator *allocator, (ProtobufCAllocator *allocator,
size_t len, size_t len,
const uint8_t *data) const uint8_t *data)
{ {
return (Decoderbufs__TxnMessage *) return (Decoderbufs__RowMessage *)
protobuf_c_message_unpack (&decoderbufs__txn_message__descriptor, protobuf_c_message_unpack (&decoderbufs__row_message__descriptor,
allocator, len, data); allocator, len, data);
} }
void decoderbufs__txn_message__free_unpacked void decoderbufs__row_message__free_unpacked
(Decoderbufs__TxnMessage *message, (Decoderbufs__RowMessage *message,
ProtobufCAllocator *allocator) ProtobufCAllocator *allocator)
{ {
assert(message->base.descriptor == &decoderbufs__txn_message__descriptor); assert(message->base.descriptor == &decoderbufs__row_message__descriptor);
protobuf_c_message_free_unpacked ((ProtobufCMessage*)message, allocator); protobuf_c_message_free_unpacked ((ProtobufCMessage*)message, allocator);
} }
static const ProtobufCFieldDescriptor decoderbufs__datum_message__field_descriptors[9] = static const ProtobufCFieldDescriptor decoderbufs__datum_message__field_descriptors[9] =
@ -235,27 +235,15 @@ const ProtobufCMessageDescriptor decoderbufs__datum_message__descriptor =
(ProtobufCMessageInit) decoderbufs__datum_message__init, (ProtobufCMessageInit) decoderbufs__datum_message__init,
NULL,NULL,NULL /* reserved[123] */ NULL,NULL,NULL /* reserved[123] */
}; };
static const ProtobufCFieldDescriptor decoderbufs__txn_message__field_descriptors[6] = static const ProtobufCFieldDescriptor decoderbufs__row_message__field_descriptors[5] =
{ {
{ {
"timestamp", "commit_time",
1, 1,
PROTOBUF_C_LABEL_OPTIONAL, PROTOBUF_C_LABEL_OPTIONAL,
PROTOBUF_C_TYPE_SINT64, PROTOBUF_C_TYPE_SINT64,
offsetof(Decoderbufs__TxnMessage, has_timestamp), offsetof(Decoderbufs__RowMessage, has_commit_time),
offsetof(Decoderbufs__TxnMessage, timestamp), offsetof(Decoderbufs__RowMessage, commit_time),
NULL,
NULL,
0, /* flags */
0,NULL,NULL /* reserved1,reserved2, etc */
},
{
"xid",
2,
PROTOBUF_C_LABEL_OPTIONAL,
PROTOBUF_C_TYPE_INT64,
offsetof(Decoderbufs__TxnMessage, has_xid),
offsetof(Decoderbufs__TxnMessage, xid),
NULL, NULL,
NULL, NULL,
0, /* flags */ 0, /* flags */
@ -263,11 +251,11 @@ static const ProtobufCFieldDescriptor decoderbufs__txn_message__field_descriptor
}, },
{ {
"table", "table",
3, 2,
PROTOBUF_C_LABEL_OPTIONAL, PROTOBUF_C_LABEL_OPTIONAL,
PROTOBUF_C_TYPE_STRING, PROTOBUF_C_TYPE_STRING,
0, /* quantifier_offset */ 0, /* quantifier_offset */
offsetof(Decoderbufs__TxnMessage, table), offsetof(Decoderbufs__RowMessage, table),
NULL, NULL,
NULL, NULL,
0, /* flags */ 0, /* flags */
@ -275,67 +263,66 @@ static const ProtobufCFieldDescriptor decoderbufs__txn_message__field_descriptor
}, },
{ {
"op", "op",
4, 3,
PROTOBUF_C_LABEL_OPTIONAL, PROTOBUF_C_LABEL_OPTIONAL,
PROTOBUF_C_TYPE_ENUM, PROTOBUF_C_TYPE_ENUM,
offsetof(Decoderbufs__TxnMessage, has_op), offsetof(Decoderbufs__RowMessage, has_op),
offsetof(Decoderbufs__TxnMessage, op), offsetof(Decoderbufs__RowMessage, op),
&decoderbufs__op__descriptor, &decoderbufs__op__descriptor,
NULL, NULL,
0, /* flags */ 0, /* flags */
0,NULL,NULL /* reserved1,reserved2, etc */ 0,NULL,NULL /* reserved1,reserved2, etc */
}, },
{ {
"new_datum", "new_tuple",
5, 4,
PROTOBUF_C_LABEL_OPTIONAL, PROTOBUF_C_LABEL_REPEATED,
PROTOBUF_C_TYPE_MESSAGE, PROTOBUF_C_TYPE_MESSAGE,
0, /* quantifier_offset */ offsetof(Decoderbufs__RowMessage, n_new_tuple),
offsetof(Decoderbufs__TxnMessage, new_datum), offsetof(Decoderbufs__RowMessage, new_tuple),
&decoderbufs__datum_message__descriptor, &decoderbufs__datum_message__descriptor,
NULL, NULL,
0, /* flags */ 0, /* flags */
0,NULL,NULL /* reserved1,reserved2, etc */ 0,NULL,NULL /* reserved1,reserved2, etc */
}, },
{ {
"old_datum", "old_tuple",
6, 5,
PROTOBUF_C_LABEL_OPTIONAL, PROTOBUF_C_LABEL_REPEATED,
PROTOBUF_C_TYPE_MESSAGE, PROTOBUF_C_TYPE_MESSAGE,
0, /* quantifier_offset */ offsetof(Decoderbufs__RowMessage, n_old_tuple),
offsetof(Decoderbufs__TxnMessage, old_datum), offsetof(Decoderbufs__RowMessage, old_tuple),
&decoderbufs__datum_message__descriptor, &decoderbufs__datum_message__descriptor,
NULL, NULL,
0, /* flags */ 0, /* flags */
0,NULL,NULL /* reserved1,reserved2, etc */ 0,NULL,NULL /* reserved1,reserved2, etc */
}, },
}; };
static const unsigned decoderbufs__txn_message__field_indices_by_name[] = { static const unsigned decoderbufs__row_message__field_indices_by_name[] = {
4, /* field[4] = new_datum */ 0, /* field[0] = commit_time */
5, /* field[5] = old_datum */ 3, /* field[3] = new_tuple */
3, /* field[3] = op */ 4, /* field[4] = old_tuple */
2, /* field[2] = table */ 2, /* field[2] = op */
0, /* field[0] = timestamp */ 1, /* field[1] = table */
1, /* field[1] = xid */
}; };
static const ProtobufCIntRange decoderbufs__txn_message__number_ranges[1 + 1] = static const ProtobufCIntRange decoderbufs__row_message__number_ranges[1 + 1] =
{ {
{ 1, 0 }, { 1, 0 },
{ 0, 6 } { 0, 5 }
}; };
const ProtobufCMessageDescriptor decoderbufs__txn_message__descriptor = const ProtobufCMessageDescriptor decoderbufs__row_message__descriptor =
{ {
PROTOBUF_C__MESSAGE_DESCRIPTOR_MAGIC, PROTOBUF_C__MESSAGE_DESCRIPTOR_MAGIC,
"decoderbufs.TxnMessage", "decoderbufs.RowMessage",
"TxnMessage", "RowMessage",
"Decoderbufs__TxnMessage", "Decoderbufs__RowMessage",
"decoderbufs", "decoderbufs",
sizeof(Decoderbufs__TxnMessage), sizeof(Decoderbufs__RowMessage),
6, 5,
decoderbufs__txn_message__field_descriptors, decoderbufs__row_message__field_descriptors,
decoderbufs__txn_message__field_indices_by_name, decoderbufs__row_message__field_indices_by_name,
1, decoderbufs__txn_message__number_ranges, 1, decoderbufs__row_message__number_ranges,
(ProtobufCMessageInit) decoderbufs__txn_message__init, (ProtobufCMessageInit) decoderbufs__row_message__init,
NULL,NULL,NULL /* reserved[123] */ NULL,NULL,NULL /* reserved[123] */
}; };
const ProtobufCEnumValue decoderbufs__op__enum_values_by_number[3] = const ProtobufCEnumValue decoderbufs__op__enum_values_by_number[3] =

54
src/proto/pg_logicaldec.pb-c.h
View File

@ -16,7 +16,7 @@ PROTOBUF_C__BEGIN_DECLS
typedef struct _Decoderbufs__DatumMessage Decoderbufs__DatumMessage; typedef struct _Decoderbufs__DatumMessage Decoderbufs__DatumMessage;
typedef struct _Decoderbufs__TxnMessage Decoderbufs__TxnMessage; typedef struct _Decoderbufs__RowMessage Decoderbufs__RowMessage;
/* --- enums --- */ /* --- enums --- */
@ -55,22 +55,22 @@ struct _Decoderbufs__DatumMessage
, NULL, 0,0, 0,0, 0,0, 0,0, 0,0, 0,0, NULL, 0,{0,NULL} } , NULL, 0,0, 0,0, 0,0, 0,0, 0,0, 0,0, NULL, 0,{0,NULL} }
struct _Decoderbufs__TxnMessage struct _Decoderbufs__RowMessage
{ {
ProtobufCMessage base; ProtobufCMessage base;
protobuf_c_boolean has_timestamp; protobuf_c_boolean has_commit_time;
int64_t timestamp; int64_t commit_time;
protobuf_c_boolean has_xid;
int64_t xid;
char *table; char *table;
protobuf_c_boolean has_op; protobuf_c_boolean has_op;
Decoderbufs__Op op; Decoderbufs__Op op;
Decoderbufs__DatumMessage *new_datum; size_t n_new_tuple;
Decoderbufs__DatumMessage *old_datum; Decoderbufs__DatumMessage **new_tuple;
size_t n_old_tuple;
Decoderbufs__DatumMessage **old_tuple;
}; };
#define DECODERBUFS__TXN_MESSAGE__INIT \ #define DECODERBUFS__ROW_MESSAGE__INIT \
{ PROTOBUF_C_MESSAGE_INIT (&decoderbufs__txn_message__descriptor) \ { PROTOBUF_C_MESSAGE_INIT (&decoderbufs__row_message__descriptor) \
, 0,0, 0,0, NULL, 0,0, NULL, NULL } , 0,0, NULL, 0,0, 0,NULL, 0,NULL }
/* Decoderbufs__DatumMessage methods */ /* Decoderbufs__DatumMessage methods */
@ -92,32 +92,32 @@ Decoderbufs__DatumMessage *
void decoderbufs__datum_message__free_unpacked void decoderbufs__datum_message__free_unpacked
(Decoderbufs__DatumMessage *message, (Decoderbufs__DatumMessage *message,
ProtobufCAllocator *allocator); ProtobufCAllocator *allocator);
/* Decoderbufs__TxnMessage methods */ /* Decoderbufs__RowMessage methods */
void decoderbufs__txn_message__init void decoderbufs__row_message__init
(Decoderbufs__TxnMessage *message); (Decoderbufs__RowMessage *message);
size_t decoderbufs__txn_message__get_packed_size size_t decoderbufs__row_message__get_packed_size
(const Decoderbufs__TxnMessage *message); (const Decoderbufs__RowMessage *message);
size_t decoderbufs__txn_message__pack size_t decoderbufs__row_message__pack
(const Decoderbufs__TxnMessage *message, (const Decoderbufs__RowMessage *message,
uint8_t *out); uint8_t *out);
size_t decoderbufs__txn_message__pack_to_buffer size_t decoderbufs__row_message__pack_to_buffer
(const Decoderbufs__TxnMessage *message, (const Decoderbufs__RowMessage *message,
ProtobufCBuffer *buffer); ProtobufCBuffer *buffer);
Decoderbufs__TxnMessage * Decoderbufs__RowMessage *
decoderbufs__txn_message__unpack decoderbufs__row_message__unpack
(ProtobufCAllocator *allocator, (ProtobufCAllocator *allocator,
size_t len, size_t len,
const uint8_t *data); const uint8_t *data);
void decoderbufs__txn_message__free_unpacked void decoderbufs__row_message__free_unpacked
(Decoderbufs__TxnMessage *message, (Decoderbufs__RowMessage *message,
ProtobufCAllocator *allocator); ProtobufCAllocator *allocator);
/* --- per-message closures --- */ /* --- per-message closures --- */
typedef void (*Decoderbufs__DatumMessage_Closure) typedef void (*Decoderbufs__DatumMessage_Closure)
(const Decoderbufs__DatumMessage *message, (const Decoderbufs__DatumMessage *message,
void *closure_data); void *closure_data);
typedef void (*Decoderbufs__TxnMessage_Closure) typedef void (*Decoderbufs__RowMessage_Closure)
(const Decoderbufs__TxnMessage *message, (const Decoderbufs__RowMessage *message,
void *closure_data); void *closure_data);
/* --- services --- */ /* --- services --- */
@ -127,7 +127,7 @@ typedef void (*Decoderbufs__TxnMessage_Closure)
extern const ProtobufCEnumDescriptor decoderbufs__op__descriptor; extern const ProtobufCEnumDescriptor decoderbufs__op__descriptor;
extern const ProtobufCMessageDescriptor decoderbufs__datum_message__descriptor; extern const ProtobufCMessageDescriptor decoderbufs__datum_message__descriptor;
extern const ProtobufCMessageDescriptor decoderbufs__txn_message__descriptor; extern const ProtobufCMessageDescriptor decoderbufs__row_message__descriptor;
PROTOBUF_C__END_DECLS PROTOBUF_C__END_DECLS