67 int nrelations,
Relation relations[],
73 bool transactional,
const char *prefix,
74 Size sz,
const char *message);
113 bool transactional,
const char *prefix,
114 Size sz,
const char *message);
117 int nrelations,
Relation relations[],
161 bool enable_streaming =
false;
165 "text conversion context",
167 data->include_xids =
true;
168 data->include_timestamp =
false;
169 data->skip_empty_xacts =
false;
170 data->only_local =
false;
183 if (strcmp(elem->
defname,
"include-xids") == 0)
186 if (elem->
arg == NULL)
187 data->include_xids =
true;
190 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
191 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
194 else if (strcmp(elem->
defname,
"include-timestamp") == 0)
196 if (elem->
arg == NULL)
197 data->include_timestamp =
true;
200 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
201 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
204 else if (strcmp(elem->
defname,
"force-binary") == 0)
208 if (elem->
arg == NULL)
212 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
213 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
219 else if (strcmp(elem->
defname,
"skip-empty-xacts") == 0)
222 if (elem->
arg == NULL)
223 data->skip_empty_xacts =
true;
226 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
227 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
230 else if (strcmp(elem->
defname,
"only-local") == 0)
233 if (elem->
arg == NULL)
234 data->only_local =
true;
237 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
238 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
241 else if (strcmp(elem->
defname,
"include-rewrites") == 0)
244 if (elem->
arg == NULL)
248 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
249 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
252 else if (strcmp(elem->
defname,
"stream-changes") == 0)
254 if (elem->
arg == NULL)
258 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
259 errmsg(
"could not parse value \"%s\" for parameter \"%s\"",
265 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
266 errmsg(
"option \"%s\" = \"%s\" is unknown",
300 if (
data->skip_empty_xacts)
310 if (
data->include_xids)
329 if (
data->skip_empty_xacts && !xact_wrote_changes)
333 if (
data->include_xids)
338 if (
data->include_timestamp)
360 if (
data->skip_empty_xacts)
386 if (
data->include_xids)
389 if (
data->include_timestamp)
408 if (
data->include_xids)
411 if (
data->include_timestamp)
432 if (
data->include_xids)
435 if (
data->include_timestamp)
453 if (strstr(gid,
"_nodecode") != NULL)
501 if (strcmp(outputstr,
"t") == 0)
509 for (valptr = outputstr; *valptr; valptr++)
529 for (natt = 0; natt < tupdesc->
natts; natt++)
544 if (attr->attisdropped)
551 if (attr->attnum < 0)
554 typid = attr->atttypid;
557 origval =
heap_getattr(tuple, natt + 1, tupdesc, &isnull);
559 if (isnull && skip_nulls)
573 &typoutput, &typisvarlena);
583 else if (!typisvarlena)
630 class_form->relrewrite ?
632 NameStr(class_form->relname)));
639 if (change->
data.
tp.newtuple == NULL)
648 if (change->
data.
tp.oldtuple != NULL)
657 if (change->
data.
tp.newtuple == NULL)
668 if (change->
data.
tp.oldtuple == NULL)
712 for (
i = 0;
i < nrelations;
i++)
719 NameStr(relations[
i]->rd_rel->relname)));
744 const char *prefix,
Size sz,
const char *message)
760 transactional, prefix, sz);
784 if (
data->skip_empty_xacts)
793 if (
data->include_xids)
811 if (
data->include_xids)
841 if (
data->skip_empty_xacts && !xact_wrote_changes)
845 if (
data->include_xids)
865 if (
data->include_xids)
872 if (
data->include_timestamp)
891 if (
data->skip_empty_xacts && !xact_wrote_changes)
896 if (
data->include_xids)
901 if (
data->include_timestamp)
930 if (
data->include_xids)
945 const char *prefix,
Size sz,
const char *message)
965 transactional, prefix, sz);
969 appendStringInfo(ctx->
out,
"streaming message: transactional: %d prefix: %s, sz: %zu content:",
970 transactional, prefix, sz);
983 int nrelations,
Relation relations[],
996 if (
data->include_xids)
const char * timestamptz_to_str(TimestampTz t)
bool parse_bool(const char *value, bool *result)
#define Assert(condition)
#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)
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
FormData_pg_attribute * Form_pg_attribute
FormData_pg_class * Form_pg_class
static Datum PointerGetDatum(const void *X)
char * quote_literal_cstr(const char *rawstr)
MemoryContextSwitchTo(old_ctx)
#define RelationGetForm(relation)
#define RelationGetRelid(relation)
#define RelationGetDescr(relation)
#define rbtxn_is_toptxn(txn)
#define rbtxn_get_toptxn(txn)
@ 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 void *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
union ReorderBufferChange::@108 data
struct ReorderBufferChange::@108::@110 truncate
struct ReorderBufferChange::@108::@109 tp
void * output_plugin_private
union ReorderBufferTXN::@114 xact_time
bool stream_wrote_changes
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 relation, 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_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, const char *message)
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_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)
static void pg_decode_stream_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, const char *message)
#define TupleDescAttr(tupdesc, i)
#define VARATT_IS_EXTERNAL_ONDISK(PTR)