PostgreSQL Source Code: src/backend/replication/logical/logicalfuncs.c Source File

PostgreSQL Source Code git master
logicalfuncs.c
Go to the documentation of this file.
1/*-------------------------------------------------------------------------
2 *
3 * logicalfuncs.c
4 *
5 * Support functions for using logical decoding and management of
6 * logical replication slots via SQL.
7 *
8 *
9 * Copyright (c) 2012-2025, PostgreSQL Global Development Group
10 *
11 * IDENTIFICATION
12 * src/backend/replication/logical/logicalfuncs.c
13 *-------------------------------------------------------------------------
14 */
15
16#include "postgres.h"
17
18#include <unistd.h>
19
20#include "access/xlogrecovery.h"
21#include "access/xlogutils.h"
22#include "catalog/pg_type.h"
23#include "fmgr.h"
24#include "funcapi.h"
25#include "mb/pg_wchar.h"
26#include "miscadmin.h"
27#include "nodes/makefuncs.h"
28#include "replication/decode.h"
29#include "replication/logical.h"
30#include "replication/message.h"
31#include "utils/array.h"
32#include "utils/builtins.h"
33#include "utils/inval.h"
34#include "utils/memutils.h"
35#include "utils/pg_lsn.h"
36#include "utils/regproc.h"
37#include "utils/resowner.h"
38
39/* Private data for writing out data */
40 typedef struct DecodingOutputState
41{
42 Tuplestorestate *tupstore;
43 TupleDesc tupdesc;
44 bool binary_output;
45 int64 returned_rows;
46 } DecodingOutputState;
47
48/*
49 * Prepare for an output plugin write.
50 */
51static void
52 LogicalOutputPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
53 bool last_write)
54{
55 resetStringInfo(ctx->out);
56}
57
58/*
59 * Perform output plugin write into tuplestore.
60 */
61static void
62 LogicalOutputWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
63 bool last_write)
64{
65 Datum values[3];
66 bool nulls[3];
67 DecodingOutputState *p;
68
69 /* SQL Datums can only be of a limited length... */
70 if (ctx->out->len > MaxAllocSize - VARHDRSZ)
71 elog(ERROR, "too much output for sql interface");
72
73 p = (DecodingOutputState *) ctx->output_writer_private;
74
75 memset(nulls, 0, sizeof(nulls));
76 values[0] = LSNGetDatum(lsn);
77 values[1] = TransactionIdGetDatum(xid);
78
79 /*
80 * Assert ctx->out is in database encoding when we're writing textual
81 * output.
82 */
83 if (!p->binary_output)
84 Assert(pg_verify_mbstr(GetDatabaseEncoding(),
85 ctx->out->data, ctx->out->len,
86 false));
87
88 /* ick, but cstring_to_text_with_len works for bytea perfectly fine */
89 values[2] = PointerGetDatum(cstring_to_text_with_len(ctx->out->data, ctx->out->len));
90
91 tuplestore_putvalues(p->tupstore, p->tupdesc, values, nulls);
92 p->returned_rows++;
93}
94
95/*
96 * Helper function for the various SQL callable logical decoding functions.
97 */
98static Datum
99 pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
100{
101 Name name;
102 XLogRecPtr upto_lsn;
103 int32 upto_nchanges;
104 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
105 MemoryContext per_query_ctx;
106 MemoryContext oldcontext;
107 XLogRecPtr end_of_wal;
108 XLogRecPtr wait_for_wal_lsn;
109 LogicalDecodingContext *ctx;
110 ResourceOwner old_resowner PG_USED_FOR_ASSERTS_ONLY = CurrentResourceOwner;
111 ArrayType *arr;
112 Size ndim;
113 List *options = NIL;
114 DecodingOutputState *p;
115
116 CheckSlotPermissions();
117
118 CheckLogicalDecodingRequirements();
119
120 if (PG_ARGISNULL(0))
121 ereport(ERROR,
122 (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
123 errmsg("slot name must not be null")));
124 name = PG_GETARG_NAME(0);
125
126 if (PG_ARGISNULL(1))
127 upto_lsn = InvalidXLogRecPtr;
128 else
129 upto_lsn = PG_GETARG_LSN(1);
130
131 if (PG_ARGISNULL(2))
132 upto_nchanges = InvalidXLogRecPtr;
133 else
134 upto_nchanges = PG_GETARG_INT32(2);
135
136 if (PG_ARGISNULL(3))
137 ereport(ERROR,
138 (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
139 errmsg("options array must not be null")));
140 arr = PG_GETARG_ARRAYTYPE_P(3);
141
142 /* state to write output to */
143 p = palloc0(sizeof(DecodingOutputState));
144
145 p->binary_output = binary;
146
147 per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
148 oldcontext = MemoryContextSwitchTo(per_query_ctx);
149
150 /* Deconstruct options array */
151 ndim = ARR_NDIM(arr);
152 if (ndim > 1)
153 {
154 ereport(ERROR,
155 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
156 errmsg("array must be one-dimensional")));
157 }
158 else if (array_contains_nulls(arr))
159 {
160 ereport(ERROR,
161 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
162 errmsg("array must not contain nulls")));
163 }
164 else if (ndim == 1)
165 {
166 int nelems;
167 Datum *datum_opts;
168 int i;
169
170 Assert(ARR_ELEMTYPE(arr) == TEXTOID);
171
172 deconstruct_array_builtin(arr, TEXTOID, &datum_opts, NULL, &nelems);
173
174 if (nelems % 2 != 0)
175 ereport(ERROR,
176 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
177 errmsg("array must have even number of elements")));
178
179 for (i = 0; i < nelems; i += 2)
180 {
181 char *optname = TextDatumGetCString(datum_opts[i]);
182 char *opt = TextDatumGetCString(datum_opts[i + 1]);
183
184 options = lappend(options, makeDefElem(optname, (Node *) makeString(opt), -1));
185 }
186 }
187
188 InitMaterializedSRF(fcinfo, 0);
189 p->tupstore = rsinfo->setResult;
190 p->tupdesc = rsinfo->setDesc;
191
192 /*
193 * Compute the current end-of-wal.
194 */
195 if (!RecoveryInProgress())
196 end_of_wal = GetFlushRecPtr(NULL);
197 else
198 end_of_wal = GetXLogReplayRecPtr(NULL);
199
200 ReplicationSlotAcquire(NameStr(*name), true, true);
201
202 PG_TRY();
203 {
204 /* restart at slot's confirmed_flush */
205 ctx = CreateDecodingContext(InvalidXLogRecPtr,
206 options,
207 false,
208 XL_ROUTINE(.page_read = read_local_xlog_page,
209 .segment_open = wal_segment_open,
210 .segment_close = wal_segment_close),
211 LogicalOutputPrepareWrite,
212 LogicalOutputWrite, NULL);
213
214 MemoryContextSwitchTo(oldcontext);
215
216 /*
217 * Check whether the output plugin writes textual output if that's
218 * what we need.
219 */
220 if (!binary &&
221 ctx->options.output_type !=OUTPUT_PLUGIN_TEXTUAL_OUTPUT)
222 ereport(ERROR,
223 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
224 errmsg("logical decoding output plugin \"%s\" produces binary output, but function \"%s\" expects textual data",
225 NameStr(MyReplicationSlot->data.plugin),
226 format_procedure(fcinfo->flinfo->fn_oid))));
227
228 /*
229 * Wait for specified streaming replication standby servers (if any)
230 * to confirm receipt of WAL up to wait_for_wal_lsn.
231 */
232 if (XLogRecPtrIsInvalid(upto_lsn))
233 wait_for_wal_lsn = end_of_wal;
234 else
235 wait_for_wal_lsn = Min(upto_lsn, end_of_wal);
236
237 WaitForStandbyConfirmation(wait_for_wal_lsn);
238
239 ctx->output_writer_private = p;
240
241 /*
242 * Decoding of WAL must start at restart_lsn so that the entirety of
243 * xacts that committed after the slot's confirmed_flush can be
244 * accumulated into reorder buffers.
245 */
246 XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
247
248 /* invalidate non-timetravel entries */
249 InvalidateSystemCaches();
250
251 /* Decode until we run out of records */
252 while (ctx->reader->EndRecPtr < end_of_wal)
253 {
254 XLogRecord *record;
255 char *errm = NULL;
256
257 record = XLogReadRecord(ctx->reader, &errm);
258 if (errm)
259 elog(ERROR, "could not find record for logical decoding: %s", errm);
260
261 /*
262 * The {begin_txn,change,commit_txn}_wrapper callbacks above will
263 * store the description into our tuplestore.
264 */
265 if (record != NULL)
266 {
267 LogicalDecodingProcessRecord(ctx, ctx->reader);
268
269 /*
270 * We used to have bugs where logical decoding would fail to
271 * preserve the resource owner. Verify that that doesn't
272 * happen anymore. XXX this could be removed once it's been
273 * battle-tested.
274 */
275 Assert(CurrentResourceOwner == old_resowner);
276 }
277
278 /* check limits */
279 if (upto_lsn != InvalidXLogRecPtr &&
280 upto_lsn <= ctx->reader->EndRecPtr)
281 break;
282 if (upto_nchanges != 0 &&
283 upto_nchanges <= p->returned_rows)
284 break;
285 CHECK_FOR_INTERRUPTS();
286 }
287
288 /*
289 * Next time, start where we left off. (Hunting things, the family
290 * business..)
291 */
292 if (ctx->reader->EndRecPtr != InvalidXLogRecPtr && confirm)
293 {
294 LogicalConfirmReceivedLocation(ctx->reader->EndRecPtr);
295
296 /*
297 * If only the confirmed_flush_lsn has changed the slot won't get
298 * marked as dirty by the above. Callers on the walsender
299 * interface are expected to keep track of their own progress and
300 * don't need it written out. But SQL-interface users cannot
301 * specify their own start positions and it's harder for them to
302 * keep track of their progress, so we should make more of an
303 * effort to save it for them.
304 *
305 * Dirty the slot so it's written out at the next checkpoint.
306 * We'll still lose its position on crash, as documented, but it's
307 * better than always losing the position even on clean restart.
308 */
309 ReplicationSlotMarkDirty();
310 }
311
312 /* free context, call shutdown callback */
313 FreeDecodingContext(ctx);
314
315 ReplicationSlotRelease();
316 InvalidateSystemCaches();
317 }
318 PG_CATCH();
319 {
320 /* clear all timetravel entries */
321 InvalidateSystemCaches();
322
323 PG_RE_THROW();
324 }
325 PG_END_TRY();
326
327 return (Datum) 0;
328}
329
330/*
331 * SQL function returning the changestream as text, consuming the data.
332 */
333Datum
334 pg_logical_slot_get_changes(PG_FUNCTION_ARGS)
335{
336 return pg_logical_slot_get_changes_guts(fcinfo, true, false);
337}
338
339/*
340 * SQL function returning the changestream as text, only peeking ahead.
341 */
342Datum
343 pg_logical_slot_peek_changes(PG_FUNCTION_ARGS)
344{
345 return pg_logical_slot_get_changes_guts(fcinfo, false, false);
346}
347
348/*
349 * SQL function returning the changestream in binary, consuming the data.
350 */
351Datum
352 pg_logical_slot_get_binary_changes(PG_FUNCTION_ARGS)
353{
354 return pg_logical_slot_get_changes_guts(fcinfo, true, true);
355}
356
357/*
358 * SQL function returning the changestream in binary, only peeking ahead.
359 */
360Datum
361 pg_logical_slot_peek_binary_changes(PG_FUNCTION_ARGS)
362{
363 return pg_logical_slot_get_changes_guts(fcinfo, false, true);
364}
365
366
367/*
368 * SQL function for writing logical decoding message into WAL.
369 */
370Datum
371 pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
372{
373 bool transactional = PG_GETARG_BOOL(0);
374 char *prefix = text_to_cstring(PG_GETARG_TEXT_PP(1));
375 bytea *data = PG_GETARG_BYTEA_PP(2);
376 bool flush = PG_GETARG_BOOL(3);
377 XLogRecPtr lsn;
378
379 lsn = LogLogicalMessage(prefix, VARDATA_ANY(data), VARSIZE_ANY_EXHDR(data),
380 transactional, flush);
381 PG_RETURN_LSN(lsn);
382}
383
384Datum
385 pg_logical_emit_message_text(PG_FUNCTION_ARGS)
386{
387 /* bytea and text are compatible */
388 return pg_logical_emit_message_bytea(fcinfo);
389}
#define ARR_NDIM(a)
Definition: array.h:290
#define PG_GETARG_ARRAYTYPE_P(n)
Definition: array.h:263
#define ARR_ELEMTYPE(a)
Definition: array.h:292
bool array_contains_nulls(ArrayType *array)
Definition: arrayfuncs.c:3768
void deconstruct_array_builtin(ArrayType *array, Oid elmtype, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3698
static Datum values[MAXATTR]
Definition: bootstrap.c:153
#define TextDatumGetCString(d)
Definition: builtins.h:98
#define NameStr(name)
Definition: c.h:751
#define Min(x, y)
Definition: c.h:1003
#define PG_USED_FOR_ASSERTS_ONLY
Definition: c.h:223
#define VARHDRSZ
Definition: c.h:697
int64_t int64
Definition: c.h:535
int32_t int32
Definition: c.h:534
uint32 TransactionId
Definition: c.h:657
size_t Size
Definition: c.h:610
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:88
int errcode(int sqlerrcode)
Definition: elog.c:854
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define PG_RE_THROW()
Definition: elog.h:405
#define PG_TRY(...)
Definition: elog.h:372
#define PG_END_TRY(...)
Definition: elog.h:397
#define ERROR
Definition: elog.h:39
#define PG_CATCH(...)
Definition: elog.h:382
#define elog(elevel,...)
Definition: elog.h:226
#define ereport(elevel,...)
Definition: elog.h:150
#define MaxAllocSize
Definition: fe_memutils.h:22
#define PG_GETARG_BYTEA_PP(n)
Definition: fmgr.h:308
#define PG_GETARG_TEXT_PP(n)
Definition: fmgr.h:309
#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
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
#define PG_FUNCTION_ARGS
Definition: fmgr.h:193
void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
Definition: funcapi.c:76
Assert(PointerIsAligned(start, uint64))
void InvalidateSystemCaches(void)
Definition: inval.c:916
i
int i
Definition: isn.c:77
List * lappend(List *list, void *datum)
Definition: list.c:339
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1820
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:677
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:498
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:111
static void LogicalOutputPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:52
Datum pg_logical_slot_get_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:334
static void LogicalOutputWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:62
Datum pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:371
Datum pg_logical_slot_get_binary_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:352
Datum pg_logical_emit_message_text(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:385
Datum pg_logical_slot_peek_binary_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:361
Datum pg_logical_slot_peek_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:343
struct DecodingOutputState DecodingOutputState
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
Definition: logicalfuncs.c:99
DefElem * makeDefElem(char *name, Node *arg, int location)
Definition: makefuncs.c:637
int GetDatabaseEncoding(void)
Definition: mbutils.c:1262
bool pg_verify_mbstr(int encoding, const char *mbstr, int len, bool noError)
Definition: mbutils.c:1567
void * palloc0(Size size)
Definition: mcxt.c:1395
XLogRecPtr LogLogicalMessage(const char *prefix, const char *message, size_t size, bool transactional, bool flush)
Definition: message.c:43
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
@ OUTPUT_PLUGIN_TEXTUAL_OUTPUT
Definition: output_plugin.h:20
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
const void * data
#define NIL
Definition: pg_list.h:68
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:36
static Datum LSNGetDatum(XLogRecPtr X)
Definition: pg_lsn.h:31
#define PG_RETURN_LSN(x)
Definition: pg_lsn.h:37
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:332
static Datum TransactionIdGetDatum(TransactionId X)
Definition: postgres.h:282
uint64_t Datum
Definition: postgres.h:70
char * format_procedure(Oid procedure_oid)
Definition: regproc.c:305
ResourceOwner CurrentResourceOwner
Definition: resowner.c:173
void ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid)
Definition: slot.c:593
void ReplicationSlotMarkDirty(void)
Definition: slot.c:1106
ReplicationSlot * MyReplicationSlot
Definition: slot.c:148
void CheckSlotPermissions(void)
Definition: slot.c:1522
void ReplicationSlotRelease(void)
Definition: slot.c:731
void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
Definition: slot.c:3059
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:126
Definition: array.h:93
TupleDesc tupdesc
Definition: logicalfuncs.c:43
Tuplestorestate * tupstore
Definition: logicalfuncs.c:42
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:280
Oid fn_oid
Definition: fmgr.h:59
FmgrInfo * flinfo
Definition: fmgr.h:87
Node * resultinfo
Definition: fmgr.h:89
Definition: pg_list.h:54
void * output_writer_private
Definition: logical.h:81
StringInfo out
Definition: logical.h:71
Definition: nodes.h:135
XLogRecPtr restart_lsn
Definition: slot.h:107
ReplicationSlotPersistentData data
Definition: slot.h:192
ExprContext * econtext
Definition: execnodes.h:356
TupleDesc setDesc
Definition: execnodes.h:364
Tuplestorestate * setResult
Definition: execnodes.h:363
char * data
Definition: stringinfo.h:48
Definition: c.h:746
Definition: oid2name.c:30
Definition: c.h:692
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, const Datum *values, const bool *isnull)
Definition: tuplestore.c:784
String * makeString(char *str)
Definition: value.c:63
static Size VARSIZE_ANY_EXHDR(const void *PTR)
Definition: varatt.h:472
static char * VARDATA_ANY(const void *PTR)
Definition: varatt.h:486
text * cstring_to_text_with_len(const char *s, int len)
Definition: varlena.c:193
char * text_to_cstring(const text *t)
Definition: varlena.c:214
const char * name
bool RecoveryInProgress(void)
Definition: xlog.c:6386
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
Definition: xlog.c:6551
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:390
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:232
#define XL_ROUTINE(...)
Definition: xlogreader.h:117
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlogrecovery.c:4561
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:831
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:806
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:845

AltStyle によって変換されたページ (->オリジナル) /