71 int nrelations,
Relation relations[],
77 bool transactional,
const char *prefix,
78 Size sz,
const char *message);
117 bool transactional,
const char *prefix,
118 Size sz,
const char *message);
121 int nrelations,
Relation relations[],
167 bool enable_streaming =
false;
171 "text conversion context",
173 data->include_xids =
true;
174 data->include_timestamp =
false;
175 data->skip_empty_xacts =
false;
176 data->only_local =
false;
189 if (strcmp(elem->
defname,
"include-xids") == 0)
192 if (elem->
arg == NULL)
193 data->include_xids =
true;
196 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
197 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
200 else if (strcmp(elem->
defname,
"include-timestamp") == 0)
202 if (elem->
arg == NULL)
203 data->include_timestamp =
true;
206 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
207 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
210 else if (strcmp(elem->
defname,
"force-binary") == 0)
214 if (elem->
arg == NULL)
218 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
219 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
225 else if (strcmp(elem->
defname,
"skip-empty-xacts") == 0)
228 if (elem->
arg == NULL)
229 data->skip_empty_xacts =
true;
232 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
233 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
236 else if (strcmp(elem->
defname,
"only-local") == 0)
239 if (elem->
arg == NULL)
240 data->only_local =
true;
243 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
244 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
247 else if (strcmp(elem->
defname,
"include-rewrites") == 0)
250 if (elem->
arg == NULL)
254 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
255 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
258 else if (strcmp(elem->
defname,
"stream-changes") == 0)
260 if (elem->
arg == NULL)
264 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
265 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
271 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
272 errmsg(
"option \"%s\" = \"%s\" is unknown",
306 if (
data->skip_empty_xacts)
316 if (
data->include_xids)
335 if (
data->skip_empty_xacts && !xact_wrote_changes)
339 if (
data->include_xids)
344 if (
data->include_timestamp)
366 if (
data->skip_empty_xacts)
392 if (
data->include_xids)
395 if (
data->include_timestamp)
414 if (
data->include_xids)
417 if (
data->include_timestamp)
438 if (
data->include_xids)
441 if (
data->include_timestamp)
459 if (strstr(gid,
"_nodecode") != NULL)
507 if (strcmp(outputstr,
"t") == 0)
515 for (valptr = outputstr; *valptr; valptr++)
535 for (natt = 0; natt < tupdesc->
natts; natt++)
550 if (attr->attisdropped)
557 if (attr->attnum < 0)
560 typid = attr->atttypid;
563 origval =
heap_getattr(tuple, natt + 1, tupdesc, &isnull);
565 if (isnull && skip_nulls)
579 &typoutput, &typisvarlena);
589 else if (!typisvarlena)
636 class_form->relrewrite ?
638 NameStr(class_form->relname)));
645 if (change->
data.
tp.newtuple == NULL)
649 &change->
data.
tp.newtuple->tuple,
654 if (change->
data.
tp.oldtuple != NULL)
658 &change->
data.
tp.oldtuple->tuple,
663 if (change->
data.
tp.newtuple == NULL)
667 &change->
data.
tp.newtuple->tuple,
674 if (change->
data.
tp.oldtuple == NULL)
679 &change->
data.
tp.oldtuple->tuple,
718 for (
i = 0;
i < nrelations;
i++)
725 NameStr(relations[
i]->rd_rel->relname)));
750 const char *prefix,
Size sz,
const char *message)
754 transactional, prefix, sz);
778 if (
data->skip_empty_xacts)
787 if (
data->include_xids)
805 if (
data->include_xids)
835 if (
data->skip_empty_xacts && !xact_wrote_changes)
839 if (
data->include_xids)
859 if (
data->include_xids)
866 if (
data->include_timestamp)
885 if (
data->skip_empty_xacts && !xact_wrote_changes)
890 if (
data->include_xids)
895 if (
data->include_timestamp)
924 if (
data->include_xids)
939 const char *prefix,
Size sz,
const char *message)
946 transactional, prefix, sz);
950 appendStringInfo(ctx->
out,
"streaming message: transactional: %d prefix: %s, sz: %zu content:",
951 transactional, prefix, sz);
964 int nrelations,
Relation relations[],
977 if (
data->include_xids)
const char * timestamptz_to_str(TimestampTz t)
bool parse_bool(const char *value, bool *result)
#define AssertVariableIsOfType(varname, typename)
#define SQL_STR_DOUBLE(ch, escape_backslash)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
char * OidOutputFunctionCall(Oid functionId, Datum val)
#define PG_DETOAST_DATUM(datum)
static Datum heap_getattr(HeapTuple tup, int attnum, TupleDesc tupleDesc, bool *isnull)
Assert(fmt[strlen(fmt) - 1] !='\n')
void OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
void OutputPluginPrepareWrite(struct LogicalDecodingContext *ctx, bool last_write)
char * get_namespace_name(Oid nspid)
void getTypeOutputInfo(Oid type, Oid *typOutput, bool *typIsVarlena)
Oid get_rel_namespace(Oid relid)
char * get_rel_name(Oid relid)
void MemoryContextReset(MemoryContext context)
void pfree(void *pointer)
void * palloc0(Size size)
void * MemoryContextAllocZero(MemoryContext context, Size size)
void MemoryContextDelete(MemoryContext context)
#define AllocSetContextCreate
#define ALLOCSET_DEFAULT_SIZES
#define IsA(nodeptr, _type_)
#define InvalidRepOriginId
@ OUTPUT_PLUGIN_BINARY_OUTPUT
@ OUTPUT_PLUGIN_TEXTUAL_OUTPUT
void(* LogicalOutputPluginInit)(struct OutputPluginCallbacks *cb)
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
FormData_pg_attribute * Form_pg_attribute
FormData_pg_class * Form_pg_class
#define VARATT_IS_EXTERNAL_ONDISK(PTR)
#define PointerGetDatum(X)
char * quote_literal_cstr(const char *rawstr)
#define RelationGetForm(relation)
#define RelationGetRelid(relation)
#define RelationGetDescr(relation)
@ REORDER_BUFFER_CHANGE_INSERT
@ REORDER_BUFFER_CHANGE_DELETE
@ REORDER_BUFFER_CHANGE_UPDATE
const char * quote_identifier(const char *ident)
char * quote_qualified_identifier(const char *qualifier, const char *ident)
void appendStringInfo(StringInfo str, const char *fmt,...)
void appendBinaryStringInfo(StringInfo str, const char *data, int datalen)
void appendStringInfoString(StringInfo str, const char *s)
void appendStringInfoChar(StringInfo str, char ch)
void * output_plugin_private
List * output_plugin_options
LogicalDecodeStreamChangeCB stream_change_cb
LogicalDecodeMessageCB message_cb
LogicalDecodeStreamTruncateCB stream_truncate_cb
LogicalDecodeStreamMessageCB stream_message_cb
LogicalDecodeFilterPrepareCB filter_prepare_cb
LogicalDecodeFilterByOriginCB filter_by_origin_cb
LogicalDecodeTruncateCB truncate_cb
LogicalDecodeStreamStopCB stream_stop_cb
LogicalDecodeStreamCommitCB stream_commit_cb
LogicalDecodeRollbackPreparedCB rollback_prepared_cb
LogicalDecodeStreamPrepareCB stream_prepare_cb
LogicalDecodeCommitPreparedCB commit_prepared_cb
LogicalDecodeStreamStartCB stream_start_cb
LogicalDecodePrepareCB prepare_cb
LogicalDecodeStartupCB startup_cb
LogicalDecodeCommitCB commit_cb
LogicalDecodeBeginCB begin_cb
LogicalDecodeStreamAbortCB stream_abort_cb
LogicalDecodeBeginPrepareCB begin_prepare_cb
LogicalDecodeChangeCB change_cb
LogicalDecodeShutdownCB shutdown_cb
OutputPluginOutputType output_type
ReorderBufferChangeType action
struct ReorderBufferChange::@105::@107 truncate
struct ReorderBufferChange::@105::@106 tp
union ReorderBufferChange::@105 data
struct ReorderBufferTXN * toptxn
void * output_plugin_private
union ReorderBufferTXN::@111 xact_time
bool stream_wrote_changes
static void pg_decode_stream_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message)
static void pg_decode_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
static void pg_decode_stream_start(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void pg_decode_rollback_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)
static void pg_decode_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
static void tuple_to_stringinfo(StringInfo s, TupleDesc tupdesc, HeapTuple tuple, bool skip_nulls)
static void pg_decode_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Relation rel, ReorderBufferChange *change)
static void pg_decode_begin_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void pg_output_begin(LogicalDecodingContext *ctx, TestDecodingData *data, ReorderBufferTXN *txn, bool last_write)
static void pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
static void print_literal(StringInfo s, Oid typid, char *outputstr)
static void pg_decode_stream_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
static void pg_decode_shutdown(LogicalDecodingContext *ctx)
static void pg_decode_stream_abort(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr abort_lsn)
static void pg_decode_stream_prepare(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
static void pg_decode_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message)
static void pg_decode_stream_commit(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
static bool pg_decode_filter(LogicalDecodingContext *ctx, RepOriginId origin_id)
static void pg_output_stream_start(LogicalDecodingContext *ctx, TestDecodingData *data, ReorderBufferTXN *txn, bool last_write)
static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid)
static void pg_decode_stream_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
void _PG_output_plugin_init(OutputPluginCallbacks *cb)
static void pg_decode_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change)
static void pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
static void pg_decode_stream_stop(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
#define TupleDescAttr(tupdesc, i)