74 elog(
ERROR,
"too much output for sql interface");
78 memset(nulls, 0,
sizeof(nulls));
103 (
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
104 errmsg(
"must be superuser or replication role to use replication slots")));
133 (
errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
134 errmsg(
"slot name must not be null")));
149 (
errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
150 errmsg(
"options array must not be null")));
156 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
157 errmsg(
"set-valued function called in context that cannot accept a set")));
160 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
161 errmsg(
"materialize mode required, but it is not allowed in this context")));
166 p->binary_output = binary;
170 elog(
ERROR,
"return type must be a row type");
180 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
181 errmsg(
"array must be one-dimensional")));
186 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
187 errmsg(
"array must not contain nulls")));
198 &datum_opts, NULL, &nelems);
202 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
203 errmsg(
"array must have even number of elements")));
205 for (i = 0; i < nelems; i += 2)
250 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
251 errmsg(
"can no longer get changes from replication slot \"%s\"",
253 errdetail(
"This slot has never previously reserved WAL, or has been invalidated.")));
264 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
265 errmsg(
"logical decoding output plugin \"%s\" produces binary output, but function \"%s\" expects textual data",
269 ctx->output_writer_private = p;
282 while (ctx->reader->EndRecPtr < end_of_wal)
300 upto_lsn <= ctx->reader->EndRecPtr)
302 if (upto_nchanges != 0 &&
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
Value * makeString(char *str)
#define PG_GETARG_INT32(n)
#define InvalidXLogRecPtr
#define IsA(nodeptr, _type_)
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
void wal_segment_close(XLogReaderState *state)
#define PointerGetDatum(X)
ResourceOwner CurrentResourceOwner
static void check_permissions(void)
#define tuplestore_donestoring(state)
Datum pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
int errcode(int sqlerrcode)
Datum pg_logical_slot_get_binary_changes(PG_FUNCTION_ARGS)
Tuplestorestate * tupstore
XLogRecPtr GetFlushRecPtr(void)
#define PG_GETARG_BOOL(n)
ReplicationSlotPersistentData data
bool RecoveryInProgress(void)
DefElem * makeDefElem(char *name, Node *arg, int location)
void InvalidateSystemCaches(void)
bool pg_verify_mbstr(int encoding, const char *mbstr, int len, bool noError)
#define PG_GETARG_TEXT_PP(n)
#define PG_GETARG_ARRAYTYPE_P(n)
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
struct DecodingOutputState DecodingOutputState
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
text * cstring_to_text_with_len(const char *s, int len)
int errdetail(const char *fmt,...)
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
void resetStringInfo(StringInfo str)
void ReplicationSlotRelease(void)
static void LogicalOutputPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
List * lappend(List *list, void *datum)
XLogRecPtr LogLogicalMessage(const char *prefix, const char *message, size_t size, bool transactional)
#define XLogRecPtrIsInvalid(r)
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
#define TextDatumGetCString(d)
#define TransactionIdGetDatum(X)
void * palloc0(Size size)
int GetDatabaseEncoding(void)
TimeLineID ThisTimeLineID
#define ereport(elevel,...)
char * format_procedure(Oid procedure_oid)
Datum pg_logical_slot_get_changes(PG_FUNCTION_ARGS)
SetFunctionReturnMode returnMode
ReplicationSlot * MyReplicationSlot
#define Assert(condition)
void FreeDecodingContext(LogicalDecodingContext *ctx)
#define PG_GETARG_BYTEA_PP(n)
bool has_rolreplication(Oid roleid)
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
MemoryContext ecxt_per_query_memory
Tuplestorestate * setResult
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
static Datum values[MAXATTR]
char * text_to_cstring(const text *t)
#define VARSIZE_ANY_EXHDR(PTR)
int errmsg(const char *fmt,...)
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Datum pg_logical_slot_peek_changes(PG_FUNCTION_ARGS)
#define CHECK_FOR_INTERRUPTS()
bool array_contains_nulls(ArrayType *array)
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
void CheckLogicalDecodingRequirements(void)
void * output_writer_private
static void LogicalOutputWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
void ReplicationSlotMarkDirty(void)
#define PG_GETARG_NAME(n)
Datum pg_logical_emit_message_text(PG_FUNCTION_ARGS)
Datum pg_logical_slot_peek_binary_changes(PG_FUNCTION_ARGS)