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/xlogrecovery.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 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

◆ LogicalOutputPrepareWrite()

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

Definition at line 56 of file logicalfuncs.c.

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

References LogicalDecodingContext::out, and resetStringInfo().

Referenced by pg_logical_slot_get_changes_guts().

◆ LogicalOutputWrite()

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

Definition at line 66 of file logicalfuncs.c.

68 {
69  Datum values[3];
70  bool nulls[3];
72 
73  /* SQL Datums can only be of a limited length... */
74  if (ctx->out->len > MaxAllocSize - VARHDRSZ)
75  elog(ERROR, "too much output for sql interface");
76 
78 
79  memset(nulls, 0, sizeof(nulls));
80  values[0] = LSNGetDatum(lsn);
81  values[1] = TransactionIdGetDatum(xid);
82 
83  /*
84  * Assert ctx->out is in database encoding when we're writing textual
85  * output.
86  */
87  if (!p->binary_output)
89  ctx->out->data, ctx->out->len,
90  false));
91 
92  /* ick, but cstring_to_text_with_len works for bytea perfectly fine */
94 
96  p->returned_rows++;
97 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define VARHDRSZ
Definition: c.h:627
#define ERROR
Definition: elog.h:33
#define elog(elevel,...)
Definition: elog.h:218
Assert(fmt[strlen(fmt) - 1] !='\n')
int GetDatabaseEncoding(void)
Definition: mbutils.c:1210
bool pg_verify_mbstr(int encoding, const char *mbstr, int len, bool noError)
Definition: mbutils.c:1515
#define MaxAllocSize
Definition: memutils.h:40
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
#define TransactionIdGetDatum(X)
Definition: postgres.h:565
uintptr_t Datum
Definition: postgres.h:411
#define PointerGetDatum(X)
Definition: postgres.h:600
Tuplestorestate * tupstore
Definition: logicalfuncs.c:46
void * output_writer_private
Definition: logical.h:81
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
Definition: tuplestore.c:750
text * cstring_to_text_with_len(const char *s, int len)
Definition: varlena.c:200

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().

◆ pg_logical_emit_message_bytea()

Datum pg_logical_emit_message_bytea ( PG_FUNCTION_ARGS  )

Definition at line 374 of file logicalfuncs.c.

375 {
376  bool transactional = PG_GETARG_BOOL(0);
377  char *prefix = text_to_cstring(PG_GETARG_TEXT_PP(1));
379  XLogRecPtr lsn;
380 
382  transactional);
383  PG_RETURN_LSN(lsn);
384 }
#define PG_GETARG_BYTEA_PP(n)
Definition: fmgr.h:308
#define PG_GETARG_TEXT_PP(n)
Definition: fmgr.h:309
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
XLogRecPtr LogLogicalMessage(const char *prefix, const char *message, size_t size, bool transactional)
Definition: message.c:46
const void * data
#define PG_RETURN_LSN(x)
Definition: pg_lsn.h:25
#define VARDATA_ANY(PTR)
Definition: postgres.h:361
#define VARSIZE_ANY_EXHDR(PTR)
Definition: postgres.h:354
Definition: c.h:622
char * text_to_cstring(const text *t)
Definition: varlena.c:221
uint64 XLogRecPtr
Definition: xlogdefs.h:21

References data, 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().

◆ pg_logical_emit_message_text()

Datum pg_logical_emit_message_text ( PG_FUNCTION_ARGS  )

Definition at line 387 of file logicalfuncs.c.

388 {
389  /* bytea and text are compatible */
390  return pg_logical_emit_message_bytea(fcinfo);
391 }
Datum pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:374

References pg_logical_emit_message_bytea().

◆ pg_logical_slot_get_binary_changes()

Datum pg_logical_slot_get_binary_changes ( PG_FUNCTION_ARGS  )

Definition at line 355 of file logicalfuncs.c.

356 {
357  return pg_logical_slot_get_changes_guts(fcinfo, true, true);
358 }
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
Definition: logicalfuncs.c:103

References pg_logical_slot_get_changes_guts().

◆ pg_logical_slot_get_changes()

Datum pg_logical_slot_get_changes ( PG_FUNCTION_ARGS  )

Definition at line 337 of file logicalfuncs.c.

338 {
339  return pg_logical_slot_get_changes_guts(fcinfo, true, false);
340 }

References pg_logical_slot_get_changes_guts().

◆ pg_logical_slot_get_changes_guts()

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

Definition at line 103 of file logicalfuncs.c.

104 {
105  Name name;
106  XLogRecPtr upto_lsn;
107  int32 upto_nchanges;
108  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
109  MemoryContext per_query_ctx;
110  MemoryContext oldcontext;
111  XLogRecPtr end_of_wal;
113  ResourceOwner old_resowner = CurrentResourceOwner;
114  ArrayType *arr;
115  Size ndim;
116  List *options = NIL;
118 
120 
122 
123  if (PG_ARGISNULL(0))
124  ereport(ERROR,
125  (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
126  errmsg("slot name must not be null")));
127  name = PG_GETARG_NAME(0);
128 
129  if (PG_ARGISNULL(1))
130  upto_lsn = InvalidXLogRecPtr;
131  else
132  upto_lsn = PG_GETARG_LSN(1);
133 
134  if (PG_ARGISNULL(2))
135  upto_nchanges = InvalidXLogRecPtr;
136  else
137  upto_nchanges = PG_GETARG_INT32(2);
138 
139  if (PG_ARGISNULL(3))
140  ereport(ERROR,
141  (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
142  errmsg("options array must not be null")));
143  arr = PG_GETARG_ARRAYTYPE_P(3);
144 
145  /* state to write output to */
146  p = palloc0(sizeof(DecodingOutputState));
147 
148  p->binary_output = binary;
149 
150  per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
151  oldcontext = MemoryContextSwitchTo(per_query_ctx);
152 
153  /* Deconstruct options array */
154  ndim = ARR_NDIM(arr);
155  if (ndim > 1)
156  {
157  ereport(ERROR,
158  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
159  errmsg("array must be one-dimensional")));
160  }
161  else if (array_contains_nulls(arr))
162  {
163  ereport(ERROR,
164  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
165  errmsg("array must not contain nulls")));
166  }
167  else if (ndim == 1)
168  {
169  int nelems;
170  Datum *datum_opts;
171  int i;
172 
173  Assert(ARR_ELEMTYPE(arr) == TEXTOID);
174 
175  deconstruct_array(arr, TEXTOID, -1, false, TYPALIGN_INT,
176  &datum_opts, NULL, &nelems);
177 
178  if (nelems % 2 != 0)
179  ereport(ERROR,
180  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
181  errmsg("array must have even number of elements")));
182 
183  for (i = 0; i < nelems; i += 2)
184  {
185  char *name = TextDatumGetCString(datum_opts[i]);
186  char *opt = TextDatumGetCString(datum_opts[i + 1]);
187 
189  }
190  }
191 
192  SetSingleFuncCall(fcinfo, 0);
193  p->tupstore = rsinfo->setResult;
194  p->tupdesc = rsinfo->setDesc;
195 
196  /*
197  * Compute the current end-of-wal.
198  */
199  if (!RecoveryInProgress())
200  end_of_wal = GetFlushRecPtr(NULL);
201  else
202  end_of_wal = GetXLogReplayRecPtr(NULL);
203 
205 
206  PG_TRY();
207  {
208  /* restart at slot's confirmed_flush */
210  options,
211  false,
212  XL_ROUTINE(.page_read = read_local_xlog_page,
213  .segment_open = wal_segment_open,
214  .segment_close = wal_segment_close),
216  LogicalOutputWrite, NULL);
217 
218  /*
219  * After the sanity checks in CreateDecodingContext, make sure the
220  * restart_lsn is valid. Avoid "cannot get changes" wording in this
221  * errmsg because that'd be confusingly ambiguous about no changes
222  * being available.
223  */
225  ereport(ERROR,
226  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
227  errmsg("can no longer get changes from replication slot \"%s\"",
228  NameStr(*name)),
229  errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
230 
231  MemoryContextSwitchTo(oldcontext);
232 
233  /*
234  * Check whether the output plugin writes textual output if that's
235  * what we need.
236  */
237  if (!binary &&
238  ctx->options.output_type !=OUTPUT_PLUGIN_TEXTUAL_OUTPUT)
239  ereport(ERROR,
240  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
241  errmsg("logical decoding output plugin \"%s\" produces binary output, but function \"%s\" expects textual data",
243  format_procedure(fcinfo->flinfo->fn_oid))));
244 
245  ctx->output_writer_private = p;
246 
247  /*
248  * Decoding of WAL must start at restart_lsn so that the entirety of
249  * xacts that committed after the slot's confirmed_flush can be
250  * accumulated into reorder buffers.
251  */
253 
254  /* invalidate non-timetravel entries */
256 
257  /* Decode until we run out of records */
258  while (ctx->reader->EndRecPtr < end_of_wal)
259  {
260  XLogRecord *record;
261  char *errm = NULL;
262 
263  record = XLogReadRecord(ctx->reader, &errm);
264  if (errm)
265  elog(ERROR, "could not find record for logical decoding: %s", errm);
266 
267  /*
268  * The {begin_txn,change,commit_txn}_wrapper callbacks above will
269  * store the description into our tuplestore.
270  */
271  if (record != NULL)
272  LogicalDecodingProcessRecord(ctx, ctx->reader);
273 
274  /* check limits */
275  if (upto_lsn != InvalidXLogRecPtr &&
276  upto_lsn <= ctx->reader->EndRecPtr)
277  break;
278  if (upto_nchanges != 0 &&
279  upto_nchanges <= p->returned_rows)
280  break;
282  }
283 
284  /*
285  * Logical decoding could have clobbered CurrentResourceOwner during
286  * transaction management, so restore the executor's value. (This is
287  * a kluge, but it's not worth cleaning up right now.)
288  */
289  CurrentResourceOwner = old_resowner;
290 
291  /*
292  * Next time, start where we left off. (Hunting things, the family
293  * business..)
294  */
295  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr && confirm)
296  {
297  LogicalConfirmReceivedLocation(ctx->reader->EndRecPtr);
298 
299  /*
300  * If only the confirmed_flush_lsn has changed the slot won't get
301  * marked as dirty by the above. Callers on the walsender
302  * interface are expected to keep track of their own progress and
303  * don't need it written out. But SQL-interface users cannot
304  * specify their own start positions and it's harder for them to
305  * keep track of their progress, so we should make more of an
306  * effort to save it for them.
307  *
308  * Dirty the slot so it's written out at the next checkpoint.
309  * We'll still lose its position on crash, as documented, but it's
310  * better than always losing the position even on clean restart.
311  */
313  }
314 
315  /* free context, call shutdown callback */
316  FreeDecodingContext(ctx);
317 
320  }
321  PG_CATCH();
322  {
323  /* clear all timetravel entries */
325 
326  PG_RE_THROW();
327  }
328  PG_END_TRY();
329 
330  return (Datum) 0;
331 }
#define ARR_NDIM(a)
Definition: array.h:283
#define PG_GETARG_ARRAYTYPE_P(n)
Definition: array.h:256
#define ARR_ELEMTYPE(a)
Definition: array.h:285
bool array_contains_nulls(ArrayType *array)
Definition: arrayfuncs.c:3558
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3491
#define TextDatumGetCString(d)
Definition: builtins.h:86
#define NameStr(name)
Definition: c.h:681
signed int int32
Definition: c.h:429
size_t Size
Definition: c.h:540
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:91
int errdetail(const char *fmt,...)
Definition: elog.c:1037
int errcode(int sqlerrcode)
Definition: elog.c:693
int errmsg(const char *fmt,...)
Definition: elog.c:904
#define PG_RE_THROW()
Definition: elog.h:340
#define PG_END_TRY()
Definition: elog.h:324
#define PG_TRY()
Definition: elog.h:299
#define PG_CATCH()
Definition: elog.h:309
#define ereport(elevel,...)
Definition: elog.h:143
const char * name
Definition: encode.c:561
#define PG_ARGISNULL(n)
Definition: fmgr.h:209
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278
#define PG_GETARG_INT32(n)
Definition: fmgr.h:269
void SetSingleFuncCall(FunctionCallInfo fcinfo, bits32 flags)
Definition: funcapi.c:76
void InvalidateSystemCaches(void)
Definition: inval.c:702
int i
Definition: isn.c:73
List * lappend(List *list, void *datum)
Definition: list.c:336
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1736
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:634
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:103
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:479
static void LogicalOutputPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:56
static void LogicalOutputWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:66
DefElem * makeDefElem(char *name, Node *arg, int location)
Definition: makefuncs.c:547
void * palloc0(Size size)
Definition: mcxt.c:1099
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
@ OUTPUT_PLUGIN_TEXTUAL_OUTPUT
Definition: output_plugin.h:20
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
#define NIL
Definition: pg_list.h:65
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:24
char * format_procedure(Oid procedure_oid)
Definition: regproc.c:323
ResourceOwner CurrentResourceOwner
Definition: resowner.c:146
void ReplicationSlotMarkDirty(void)
Definition: slot.c:798
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:425
ReplicationSlot * MyReplicationSlot
Definition: slot.c:97
void CheckSlotPermissions(void)
Definition: slot.c:1136
void ReplicationSlotRelease(void)
Definition: slot.c:522
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:239
Oid fn_oid
Definition: fmgr.h:59
fmNodePtr resultinfo
Definition: fmgr.h:89
FmgrInfo * flinfo
Definition: fmgr.h:87
Definition: pg_list.h:51
Definition: nodes.h:574
XLogRecPtr restart_lsn
Definition: slot.h:73
ReplicationSlotPersistentData data
Definition: slot.h:147
ExprContext * econtext
Definition: execnodes.h:309
TupleDesc setDesc
Definition: execnodes.h:317
Tuplestorestate * setResult
Definition: execnodes.h:316
Definition: c.h:676
String * makeString(char *str)
Definition: value.c:63
bool RecoveryInProgress(void)
Definition: xlog.c:5753
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
Definition: xlog.c:5918
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:418
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:264
#define XL_ROUTINE(...)
Definition: xlogreader.h:117
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:856
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:831
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:875

References ARR_ELEMTYPE, ARR_NDIM, array_contains_nulls(), Assert(), CHECK_FOR_INTERRUPTS, CheckLogicalDecodingRequirements(), CheckSlotPermissions(), 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(), GetFlushRecPtr(), GetXLogReplayRecPtr(), i, InvalidateSystemCaches(), InvalidXLogRecPtr, 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, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, SetSingleFuncCall(), TextDatumGetCString, wal_segment_close(), wal_segment_open(), 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().

◆ pg_logical_slot_peek_binary_changes()

Datum pg_logical_slot_peek_binary_changes ( PG_FUNCTION_ARGS  )

Definition at line 364 of file logicalfuncs.c.

365 {
366  return pg_logical_slot_get_changes_guts(fcinfo, false, true);
367 }

References pg_logical_slot_get_changes_guts().

◆ pg_logical_slot_peek_changes()

Datum pg_logical_slot_peek_changes ( PG_FUNCTION_ARGS  )

Definition at line 346 of file logicalfuncs.c.

347 {
348  return pg_logical_slot_get_changes_guts(fcinfo, false, false);
349 }

References pg_logical_slot_get_changes_guts().