PostgreSQL Source Code  git master
logicalfuncs.c File Reference
#include "postgres.h"
#include <unistd.h>
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "access/xlogutils.h"
#include "catalog/pg_type.h"
#include "fmgr.h"
#include "funcapi.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/message.h"
#include "storage/fd.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/regproc.h"
#include "utils/resowner.h"
Include dependency graph for logicalfuncs.c:

Go to the source code of this file.

Data Structures

struct  DecodingOutputState
 

Typedefs

typedef struct DecodingOutputState DecodingOutputState
 

Functions

static void LogicalOutputPrepareWrite (LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
 
static void LogicalOutputWrite (LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
 
static void check_permissions (void)
 
static Datum pg_logical_slot_get_changes_guts (FunctionCallInfo fcinfo, bool confirm, bool binary)
 
Datum pg_logical_slot_get_changes (PG_FUNCTION_ARGS)
 
Datum pg_logical_slot_peek_changes (PG_FUNCTION_ARGS)
 
Datum pg_logical_slot_get_binary_changes (PG_FUNCTION_ARGS)
 
Datum pg_logical_slot_peek_binary_changes (PG_FUNCTION_ARGS)
 
Datum pg_logical_emit_message_bytea (PG_FUNCTION_ARGS)
 
Datum pg_logical_emit_message_text (PG_FUNCTION_ARGS)
 

Typedef Documentation

◆ DecodingOutputState

Function Documentation

◆ check_permissions()

static void check_permissions ( void  )
static

Definition at line 99 of file logicalfuncs.c.

References ereport, errcode(), errmsg(), ERROR, GetUserId(), has_rolreplication(), and superuser().

Referenced by pg_logical_slot_get_changes_guts().

100 {
101  if (!superuser() && !has_rolreplication(GetUserId()))
102  ereport(ERROR,
103  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
104  errmsg("must be superuser or replication role to use replication slots")));
105 }
Oid GetUserId(void)
Definition: miscinit.c:476
int errcode(int sqlerrcode)
Definition: elog.c:610
bool superuser(void)
Definition: superuser.c:46
#define ERROR
Definition: elog.h:43
#define ereport(elevel,...)
Definition: elog.h:144
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:657
int errmsg(const char *fmt,...)
Definition: elog.c:821

◆ LogicalOutputPrepareWrite()

static void LogicalOutputPrepareWrite ( LogicalDecodingContext ctx,
XLogRecPtr  lsn,
TransactionId  xid,
bool  last_write 
)
static

Definition at line 55 of file logicalfuncs.c.

References LogicalDecodingContext::out, and resetStringInfo().

Referenced by pg_logical_slot_get_changes_guts().

57 {
58  resetStringInfo(ctx->out);
59 }
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:75
StringInfo out
Definition: logical.h:70

◆ LogicalOutputWrite()

static void LogicalOutputWrite ( LogicalDecodingContext ctx,
XLogRecPtr  lsn,
TransactionId  xid,
bool  last_write 
)
static

Definition at line 65 of file logicalfuncs.c.

References Assert, DecodingOutputState::binary_output, cstring_to_text_with_len(), StringInfoData::data, elog, ERROR, GetDatabaseEncoding(), StringInfoData::len, LSNGetDatum, MaxAllocSize, LogicalDecodingContext::out, LogicalDecodingContext::output_writer_private, pg_verify_mbstr(), PointerGetDatum, DecodingOutputState::returned_rows, TransactionIdGetDatum, DecodingOutputState::tupdesc, tuplestore_putvalues(), DecodingOutputState::tupstore, values, and VARHDRSZ.

Referenced by pg_logical_slot_get_changes_guts().

67 {
68  Datum values[3];
69  bool nulls[3];
71 
72  /* SQL Datums can only be of a limited length... */
73  if (ctx->out->len > MaxAllocSize - VARHDRSZ)
74  elog(ERROR, "too much output for sql interface");
75 
77 
78  memset(nulls, 0, sizeof(nulls));
79  values[0] = LSNGetDatum(lsn);
80  values[1] = TransactionIdGetDatum(xid);
81 
82  /*
83  * Assert ctx->out is in database encoding when we're writing textual
84  * output.
85  */
86  if (!p->binary_output)
88  ctx->out->data, ctx->out->len,
89  false));
90 
91  /* ick, but cstring_to_text_with_len works for bytea perfectly fine */
92  values[2] = PointerGetDatum(cstring_to_text_with_len(ctx->out->data, ctx->out->len));
93 
94  tuplestore_putvalues(p->tupstore, p->tupdesc, values, nulls);
95  p->returned_rows++;
96 }
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
Definition: tuplestore.c:750
#define PointerGetDatum(X)
Definition: postgres.h:556
#define VARHDRSZ
Definition: c.h:569
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
Tuplestorestate * tupstore
Definition: logicalfuncs.c:45
bool pg_verify_mbstr(int encoding, const char *mbstr, int len, bool noError)
Definition: mbutils.c:1457
#define ERROR
Definition: elog.h:43
text * cstring_to_text_with_len(const char *s, int len)
Definition: varlena.c:200
#define MaxAllocSize
Definition: memutils.h:40
#define TransactionIdGetDatum(X)
Definition: postgres.h:521
uintptr_t Datum
Definition: postgres.h:367
int GetDatabaseEncoding(void)
Definition: mbutils.c:1151
#define Assert(condition)
Definition: c.h:746
static Datum values[MAXATTR]
Definition: bootstrap.c:165
#define elog(elevel,...)
Definition: elog.h:214
StringInfo out
Definition: logical.h:70
void * output_writer_private
Definition: logical.h:80

◆ pg_logical_emit_message_bytea()

Datum pg_logical_emit_message_bytea ( PG_FUNCTION_ARGS  )

Definition at line 400 of file logicalfuncs.c.

References LogLogicalMessage(), PG_GETARG_BOOL, PG_GETARG_BYTEA_PP, PG_GETARG_TEXT_PP, PG_RETURN_LSN, text_to_cstring(), VARDATA_ANY, and VARSIZE_ANY_EXHDR.

Referenced by pg_logical_emit_message_text().

401 {
402  bool transactional = PG_GETARG_BOOL(0);
403  char *prefix = text_to_cstring(PG_GETARG_TEXT_PP(1));
404  bytea *data = PG_GETARG_BYTEA_PP(2);
405  XLogRecPtr lsn;
406 
407  lsn = LogLogicalMessage(prefix, VARDATA_ANY(data), VARSIZE_ANY_EXHDR(data),
408  transactional);
409  PG_RETURN_LSN(lsn);
410 }
#define VARDATA_ANY(PTR)
Definition: postgres.h:348
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
#define PG_GETARG_TEXT_PP(n)
Definition: fmgr.h:308
#define PG_RETURN_LSN(x)
Definition: pg_lsn.h:25
XLogRecPtr LogLogicalMessage(const char *prefix, const char *message, size_t size, bool transactional)
Definition: message.c:46
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define PG_GETARG_BYTEA_PP(n)
Definition: fmgr.h:307
char * text_to_cstring(const text *t)
Definition: varlena.c:221
#define VARSIZE_ANY_EXHDR(PTR)
Definition: postgres.h:341
Definition: c.h:563

◆ pg_logical_emit_message_text()

Datum pg_logical_emit_message_text ( PG_FUNCTION_ARGS  )

Definition at line 413 of file logicalfuncs.c.

References pg_logical_emit_message_bytea().

414 {
415  /* bytea and text are compatible */
416  return pg_logical_emit_message_bytea(fcinfo);
417 }
Datum pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:400

◆ pg_logical_slot_get_binary_changes()

Datum pg_logical_slot_get_binary_changes ( PG_FUNCTION_ARGS  )

Definition at line 381 of file logicalfuncs.c.

References pg_logical_slot_get_changes_guts().

382 {
383  return pg_logical_slot_get_changes_guts(fcinfo, true, true);
384 }
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
Definition: logicalfuncs.c:111

◆ pg_logical_slot_get_changes()

Datum pg_logical_slot_get_changes ( PG_FUNCTION_ARGS  )

Definition at line 363 of file logicalfuncs.c.

References pg_logical_slot_get_changes_guts().

364 {
365  return pg_logical_slot_get_changes_guts(fcinfo, true, false);
366 }
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
Definition: logicalfuncs.c:111

◆ pg_logical_slot_get_changes_guts()

static Datum pg_logical_slot_get_changes_guts ( FunctionCallInfo  fcinfo,
bool  confirm,
bool  binary 
)
static

Definition at line 111 of file logicalfuncs.c.

References ReturnSetInfo::allowedModes, ARR_ELEMTYPE, ARR_NDIM, array_contains_nulls(), Assert, CHECK_FOR_INTERRUPTS, check_permissions(), CheckLogicalDecodingRequirements(), CreateDecodingContext(), CurrentResourceOwner, ReplicationSlot::data, deconstruct_array(), ReturnSetInfo::econtext, ExprContext::ecxt_per_query_memory, elog, ereport, errcode(), errdetail(), errmsg(), ERROR, FunctionCallInfoBaseData::flinfo, FmgrInfo::fn_oid, format_procedure(), FreeDecodingContext(), get_call_result_type(), GetFlushRecPtr(), GetXLogReplayRecPtr(), i, InvalidateSystemCaches(), InvalidXLogRecPtr, IsA, lappend(), LogicalConfirmReceivedLocation(), LogicalDecodingProcessRecord(), LogicalOutputPrepareWrite(), LogicalOutputWrite(), makeDefElem(), makeString(), MemoryContextSwitchTo(), MyReplicationSlot, name, NameStr, NIL, OUTPUT_PLUGIN_TEXTUAL_OUTPUT, palloc0(), PG_ARGISNULL, PG_CATCH, PG_END_TRY, PG_GETARG_ARRAYTYPE_P, PG_GETARG_INT32, PG_GETARG_LSN, PG_GETARG_NAME, PG_RE_THROW, PG_TRY, ReplicationSlotPersistentData::plugin, read_local_xlog_page(), RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotPersistentData::restart_lsn, FunctionCallInfoBaseData::resultinfo, DecodingOutputState::returned_rows, ReturnSetInfo::returnMode, SAB_Error, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, SFRM_Materialize, TextDatumGetCString, ThisTimeLineID, tuplestore_begin_heap(), tuplestore_donestoring, DecodingOutputState::tupstore, TYPEFUNC_COMPOSITE, wal_segment_close(), wal_segment_open(), work_mem, XL_ROUTINE, XLogBeginRead(), XLogReadRecord(), and XLogRecPtrIsInvalid.

Referenced by pg_logical_slot_get_binary_changes(), pg_logical_slot_get_changes(), pg_logical_slot_peek_binary_changes(), and pg_logical_slot_peek_changes().

112 {
113  Name name;
114  XLogRecPtr upto_lsn;
115  int32 upto_nchanges;
116  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
117  MemoryContext per_query_ctx;
118  MemoryContext oldcontext;
119  XLogRecPtr end_of_wal;
121  ResourceOwner old_resowner = CurrentResourceOwner;
122  ArrayType *arr;
123  Size ndim;
124  List *options = NIL;
126 
128 
130 
131  if (PG_ARGISNULL(0))
132  ereport(ERROR,
133  (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
134  errmsg("slot name must not be null")));
135  name = PG_GETARG_NAME(0);
136 
137  if (PG_ARGISNULL(1))
138  upto_lsn = InvalidXLogRecPtr;
139  else
140  upto_lsn = PG_GETARG_LSN(1);
141 
142  if (PG_ARGISNULL(2))
143  upto_nchanges = InvalidXLogRecPtr;
144  else
145  upto_nchanges = PG_GETARG_INT32(2);
146 
147  if (PG_ARGISNULL(3))
148  ereport(ERROR,
149  (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
150  errmsg("options array must not be null")));
151  arr = PG_GETARG_ARRAYTYPE_P(3);
152 
153  /* check to see if caller supports us returning a tuplestore */
154  if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
155  ereport(ERROR,
156  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
157  errmsg("set-valued function called in context that cannot accept a set")));
158  if (!(rsinfo->allowedModes & SFRM_Materialize))
159  ereport(ERROR,
160  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
161  errmsg("materialize mode required, but it is not allowed in this context")));
162 
163  /* state to write output to */
164  p = palloc0(sizeof(DecodingOutputState));
165 
166  p->binary_output = binary;
167 
168  /* Build a tuple descriptor for our result type */
169  if (get_call_result_type(fcinfo, NULL, &p->tupdesc) != TYPEFUNC_COMPOSITE)
170  elog(ERROR, "return type must be a row type");
171 
172  per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
173  oldcontext = MemoryContextSwitchTo(per_query_ctx);
174 
175  /* Deconstruct options array */
176  ndim = ARR_NDIM(arr);
177  if (ndim > 1)
178  {
179  ereport(ERROR,
180  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
181  errmsg("array must be one-dimensional")));
182  }
183  else if (array_contains_nulls(arr))
184  {
185  ereport(ERROR,
186  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
187  errmsg("array must not contain nulls")));
188  }
189  else if (ndim == 1)
190  {
191  int nelems;
192  Datum *datum_opts;
193  int i;
194 
195  Assert(ARR_ELEMTYPE(arr) == TEXTOID);
196 
197  deconstruct_array(arr, TEXTOID, -1, false, TYPALIGN_INT,
198  &datum_opts, NULL, &nelems);
199 
200  if (nelems % 2 != 0)
201  ereport(ERROR,
202  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
203  errmsg("array must have even number of elements")));
204 
205  for (i = 0; i < nelems; i += 2)
206  {
207  char *name = TextDatumGetCString(datum_opts[i]);
208  char *opt = TextDatumGetCString(datum_opts[i + 1]);
209 
210  options = lappend(options, makeDefElem(name, (Node *) makeString(opt), -1));
211  }
212  }
213 
214  p->tupstore = tuplestore_begin_heap(true, false, work_mem);
215  rsinfo->returnMode = SFRM_Materialize;
216  rsinfo->setResult = p->tupstore;
217  rsinfo->setDesc = p->tupdesc;
218 
219  /*
220  * Compute the current end-of-wal and maintain ThisTimeLineID.
221  * RecoveryInProgress() will update ThisTimeLineID on promotion.
222  */
223  if (!RecoveryInProgress())
224  end_of_wal = GetFlushRecPtr();
225  else
226  end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID);
227 
228  (void) ReplicationSlotAcquire(NameStr(*name), SAB_Error);
229 
230  PG_TRY();
231  {
232  /* restart at slot's confirmed_flush */
234  options,
235  false,
236  XL_ROUTINE(.page_read = read_local_xlog_page,
237  .segment_open = wal_segment_open,
238  .segment_close = wal_segment_close),
240  LogicalOutputWrite, NULL);
241 
242  /*
243  * After the sanity checks in CreateDecodingContext, make sure the
244  * restart_lsn is valid. Avoid "cannot get changes" wording in this
245  * errmsg because that'd be confusingly ambiguous about no changes
246  * being available.
247  */
249  ereport(ERROR,
250  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
251  errmsg("can no longer get changes from replication slot \"%s\"",
252  NameStr(*name)),
253  errdetail("This slot has never previously reserved WAL, or has been invalidated.")));
254 
255  MemoryContextSwitchTo(oldcontext);
256 
257  /*
258  * Check whether the output plugin writes textual output if that's
259  * what we need.
260  */
261  if (!binary &&
262  ctx->options.output_type !=OUTPUT_PLUGIN_TEXTUAL_OUTPUT)
263  ereport(ERROR,
264  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
265  errmsg("logical decoding output plugin \"%s\" produces binary output, but function \"%s\" expects textual data",
267  format_procedure(fcinfo->flinfo->fn_oid))));
268 
269  ctx->output_writer_private = p;
270 
271  /*
272  * Decoding of WAL must start at restart_lsn so that the entirety of
273  * xacts that committed after the slot's confirmed_flush can be
274  * accumulated into reorder buffers.
275  */
277 
278  /* invalidate non-timetravel entries */
280 
281  /* Decode until we run out of records */
282  while (ctx->reader->EndRecPtr < end_of_wal)
283  {
284  XLogRecord *record;
285  char *errm = NULL;
286 
287  record = XLogReadRecord(ctx->reader, &errm);
288  if (errm)
289  elog(ERROR, "%s", errm);
290 
291  /*
292  * The {begin_txn,change,commit_txn}_wrapper callbacks above will
293  * store the description into our tuplestore.
294  */
295  if (record != NULL)
296  LogicalDecodingProcessRecord(ctx, ctx->reader);
297 
298  /* check limits */
299  if (upto_lsn != InvalidXLogRecPtr &&
300  upto_lsn <= ctx->reader->EndRecPtr)
301  break;
302  if (upto_nchanges != 0 &&
303  upto_nchanges <= p->returned_rows)
304  break;
306  }
307 
308  tuplestore_donestoring(tupstore);
309 
310  /*
311  * Logical decoding could have clobbered CurrentResourceOwner during
312  * transaction management, so restore the executor's value. (This is
313  * a kluge, but it's not worth cleaning up right now.)
314  */
315  CurrentResourceOwner = old_resowner;
316 
317  /*
318  * Next time, start where we left off. (Hunting things, the family
319  * business..)
320  */
321  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr && confirm)
322  {
323  LogicalConfirmReceivedLocation(ctx->reader->EndRecPtr);
324 
325  /*
326  * If only the confirmed_flush_lsn has changed the slot won't get
327  * marked as dirty by the above. Callers on the walsender
328  * interface are expected to keep track of their own progress and
329  * don't need it written out. But SQL-interface users cannot
330  * specify their own start positions and it's harder for them to
331  * keep track of their progress, so we should make more of an
332  * effort to save it for them.
333  *
334  * Dirty the slot so it's written out at the next checkpoint.
335  * We'll still lose its position on crash, as documented, but it's
336  * better than always losing the position even on clean restart.
337  */
339  }
340 
341  /* free context, call shutdown callback */
342  FreeDecodingContext(ctx);
343 
346  }
347  PG_CATCH();
348  {
349  /* clear all timetravel entries */
351 
352  PG_RE_THROW();
353  }
354  PG_END_TRY();
355 
356  return (Datum) 0;
357 }
Value * makeString(char *str)
Definition: value.c:53
#define NIL
Definition: pg_list.h:65
#define PG_GETARG_INT32(n)
Definition: fmgr.h:269
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define IsA(nodeptr, _type_)
Definition: nodes.h:579
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:205
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:812
ResourceOwner CurrentResourceOwner
Definition: resowner.c:142
static void check_permissions(void)
Definition: logicalfuncs.c:99
#define tuplestore_donestoring(state)
Definition: tuplestore.h:60
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
Definition: nodes.h:528
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:787
int errcode(int sqlerrcode)
Definition: elog.c:610
XLogRecPtr GetFlushRecPtr(void)
Definition: xlog.c:8428
ReplicationSlotPersistentData data
Definition: slot.h:143
bool RecoveryInProgress(void)
Definition: xlog.c:8076
DefElem * makeDefElem(char *name, Node *arg, int location)
Definition: makefuncs.c:546
void InvalidateSystemCaches(void)
Definition: inval.c:646
signed int int32
Definition: c.h:363
#define PG_GETARG_ARRAYTYPE_P(n)
Definition: array.h:251
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:268
#define ERROR
Definition: elog.h:43
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:94
fmNodePtr resultinfo
Definition: fmgr.h:89
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11500
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:435
Definition: c.h:617
int errdetail(const char *fmt,...)
Definition: elog.c:954
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:240
void ReplicationSlotRelease(void)
Definition: slot.c:484
static void LogicalOutputPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:55
List * lappend(List *list, void *datum)
Definition: list.c:321
#define XL_ROUTINE(...)
Definition: xlogreader.h:116
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:318
#define TextDatumGetCString(d)
Definition: builtins.h:87
void * palloc0(Size size)
Definition: mcxt.c:981
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:24
uintptr_t Datum
Definition: postgres.h:367
FmgrInfo * flinfo
Definition: fmgr.h:87
int work_mem
Definition: globals.c:121
TimeLineID ThisTimeLineID
Definition: xlog.c:192
Oid fn_oid
Definition: fmgr.h:59
#define ereport(elevel,...)
Definition: elog.h:144
int allowedModes
Definition: execnodes.h:304
char * format_procedure(Oid procedure_oid)
Definition: regproc.c:322
SetFunctionReturnMode returnMode
Definition: execnodes.h:306
#define PG_CATCH()
Definition: elog.h:305
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define PG_ARGISNULL(n)
Definition: fmgr.h:209
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:746
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:570
XLogRecPtr restart_lsn
Definition: slot.h:80
size_t Size
Definition: c.h:474
#define PG_RE_THROW()
Definition: elog.h:336
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1372
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:232
#define ARR_NDIM(a)
Definition: array.h:278
const char * name
Definition: encode.c:561
Tuplestorestate * setResult
Definition: execnodes.h:309
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3483
ExprContext * econtext
Definition: execnodes.h:302
TupleDesc setDesc
Definition: execnodes.h:310
int errmsg(const char *fmt,...)
Definition: elog.c:821
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:375
#define elog(elevel,...)
Definition: elog.h:214
int i
Definition: slot.h:42
#define NameStr(name)
Definition: c.h:623
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
#define PG_TRY()
Definition: elog.h:295
bool array_contains_nulls(ArrayType *array)
Definition: arrayfuncs.c:3550
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:831
Definition: pg_list.h:50
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:94
#define ARR_ELEMTYPE(a)
Definition: array.h:280
#define PG_END_TRY()
Definition: elog.h:320
static void LogicalOutputWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:65
void ReplicationSlotMarkDirty(void)
Definition: slot.c:737
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278

◆ pg_logical_slot_peek_binary_changes()

Datum pg_logical_slot_peek_binary_changes ( PG_FUNCTION_ARGS  )

Definition at line 390 of file logicalfuncs.c.

References pg_logical_slot_get_changes_guts().

391 {
392  return pg_logical_slot_get_changes_guts(fcinfo, false, true);
393 }
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
Definition: logicalfuncs.c:111

◆ pg_logical_slot_peek_changes()

Datum pg_logical_slot_peek_changes ( PG_FUNCTION_ARGS  )

Definition at line 372 of file logicalfuncs.c.

References pg_logical_slot_get_changes_guts().

373 {
374  return pg_logical_slot_get_changes_guts(fcinfo, false, false);
375 }
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
Definition: logicalfuncs.c:111