PostgreSQL Source Code git master
Loading...
Searching...
No Matches
Functions
wait.h File Reference
#include "nodes/parsenodes.h"
#include "parser/parse_node.h"
#include "tcop/dest.h"
Include dependency graph for wait.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions

void ExecWaitStmt (ParseState *pstate, WaitStmt *stmt, DestReceiver *dest)
 
TupleDesc WaitStmtResultDesc (WaitStmt *stmt)
 

Function Documentation

◆ ExecWaitStmt()

void ExecWaitStmt ( ParseState *  pstate,
WaitStmt *  stmt,
DestReceiver *  dest 
)
extern

Definition at line 33 of file wait.c.

34{
35 XLogRecPtr lsn;
36 int64 timeout = 0;
37 WaitLSNResult waitLSNResult;
38 WaitLSNType lsnType = WAIT_LSN_TYPE_STANDBY_REPLAY; /* default */
39 bool throw = true;
40 TupleDesc tupdesc;
41 TupOutputState *tstate;
42 const char *result = "<unset>";
43 bool timeout_specified = false;
44 bool no_throw_specified = false;
45 bool mode_specified = false;
46
47 /* Parse and validate the mandatory LSN */
48 lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in,
49 CStringGetDatum(stmt->lsn_literal)));
50
51 foreach_node(DefElem, defel, stmt->options)
52 {
53 if (strcmp(defel->defname, "mode") == 0)
54 {
55 char *mode_str;
56
57 if (mode_specified)
58 errorConflictingDefElem(defel, pstate);
59 mode_specified = true;
60
61 mode_str = defGetString(defel);
62
63 if (pg_strcasecmp(mode_str, "standby_replay") == 0)
64 lsnType = WAIT_LSN_TYPE_STANDBY_REPLAY;
65 else if (pg_strcasecmp(mode_str, "standby_write") == 0)
66 lsnType = WAIT_LSN_TYPE_STANDBY_WRITE;
67 else if (pg_strcasecmp(mode_str, "standby_flush") == 0)
68 lsnType = WAIT_LSN_TYPE_STANDBY_FLUSH;
69 else if (pg_strcasecmp(mode_str, "primary_flush") == 0)
70 lsnType = WAIT_LSN_TYPE_PRIMARY_FLUSH;
71 else
72 ereport(ERROR,
73 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
74 errmsg("unrecognized value for %s option \"%s\": \"%s\"",
75 "WAIT", defel->defname, mode_str),
76 parser_errposition(pstate, defel->location)));
77 }
78 else if (strcmp(defel->defname, "timeout") == 0)
79 {
80 char *timeout_str;
81 const char *hintmsg;
82 double result;
83
84 if (timeout_specified)
85 errorConflictingDefElem(defel, pstate);
86 timeout_specified = true;
87
88 timeout_str = defGetString(defel);
89
90 if (!parse_real(timeout_str, &result, GUC_UNIT_MS, &hintmsg))
91 {
92 ereport(ERROR,
93 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
94 errmsg("invalid timeout value: \"%s\"", timeout_str),
95 hintmsg ? errhint("%s", _(hintmsg)) : 0);
96 }
97
98 /*
99 * Get rid of any fractional part in the input. This is so we
100 * don't fail on just-out-of-range values that would round into
101 * range.
102 */
103 result = rint(result);
104
105 /* Range check */
106 if (unlikely(isnan(result) || !FLOAT8_FITS_IN_INT64(result)))
107 ereport(ERROR,
108 errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
109 errmsg("timeout value is out of range"));
110
111 if (result < 0)
112 ereport(ERROR,
113 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
114 errmsg("timeout cannot be negative"));
115
116 timeout = (int64) result;
117 }
118 else if (strcmp(defel->defname, "no_throw") == 0)
119 {
120 if (no_throw_specified)
121 errorConflictingDefElem(defel, pstate);
122
123 no_throw_specified = true;
124
125 throw = !defGetBoolean(defel);
126 }
127 else
128 {
129 ereport(ERROR,
130 errcode(ERRCODE_SYNTAX_ERROR),
131 errmsg("option \"%s\" not recognized",
132 defel->defname),
133 parser_errposition(pstate, defel->location));
134 }
135 }
136
137 /*
138 * We are going to wait for the LSN. We should first care that we don't
139 * hold a snapshot and correspondingly our MyProc->xmin is invalid.
140 * Otherwise, our snapshot could prevent the replay of WAL records
141 * implying a kind of self-deadlock. This is the reason why WAIT FOR is a
142 * command, not a procedure or function.
143 *
144 * At first, we should check there is no active snapshot. According to
145 * PlannedStmtRequiresSnapshot(), even in an atomic context, CallStmt is
146 * processed with a snapshot. Thankfully, we can pop this snapshot,
147 * because PortalRunUtility() can tolerate this.
148 */
149 if (ActiveSnapshotSet())
150 PopActiveSnapshot();
151
152 /*
153 * At second, invalidate a catalog snapshot if any. And we should be done
154 * with the preparation.
155 */
156 InvalidateCatalogSnapshot();
157
158 /* Give up if there is still an active or registered snapshot. */
159 if (HaveRegisteredOrActiveSnapshot())
160 ereport(ERROR,
161 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
162 errmsg("WAIT FOR must be called without an active or registered snapshot"),
163 errdetail("WAIT FOR cannot be executed from a function or procedure, nor within a transaction with an isolation level higher than READ COMMITTED."));
164
165 /*
166 * As the result we should hold no snapshot, and correspondingly our xmin
167 * should be unset.
168 */
169 Assert(MyProc->xmin == InvalidTransactionId);
170
171 /*
172 * Validate that the requested mode matches the current server state.
173 * Primary modes can only be used on a primary.
174 */
175 if (lsnType == WAIT_LSN_TYPE_PRIMARY_FLUSH)
176 {
177 if (RecoveryInProgress())
178 ereport(ERROR,
179 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
180 errmsg("recovery is in progress"),
181 errhint("Waiting for primary_flush can only be done on a primary server. "
182 "Use standby_flush mode on a standby server.")));
183 }
184
185 /* Now wait for the LSN */
186 waitLSNResult = WaitForLSN(lsnType, lsn, timeout);
187
188 /*
189 * Process the result of WaitForLSN(). Throw appropriate error if needed.
190 */
191 switch (waitLSNResult)
192 {
193 case WAIT_LSN_RESULT_SUCCESS:
194 /* Nothing to do on success */
195 result = "success";
196 break;
197
198 case WAIT_LSN_RESULT_TIMEOUT:
199 if (throw)
200 {
201 XLogRecPtr currentLSN = GetCurrentLSNForWaitType(lsnType);
202
203 switch (lsnType)
204 {
205 case WAIT_LSN_TYPE_STANDBY_REPLAY:
206 ereport(ERROR,
207 errcode(ERRCODE_QUERY_CANCELED),
208 errmsg("timed out while waiting for target LSN %X/%08X to be replayed; current standby_replay LSN %X/%08X",
209 LSN_FORMAT_ARGS(lsn),
210 LSN_FORMAT_ARGS(currentLSN)));
211 break;
212
213 case WAIT_LSN_TYPE_STANDBY_WRITE:
214 ereport(ERROR,
215 errcode(ERRCODE_QUERY_CANCELED),
216 errmsg("timed out while waiting for target LSN %X/%08X to be written; current standby_write LSN %X/%08X",
217 LSN_FORMAT_ARGS(lsn),
218 LSN_FORMAT_ARGS(currentLSN)));
219 break;
220
221 case WAIT_LSN_TYPE_STANDBY_FLUSH:
222 ereport(ERROR,
223 errcode(ERRCODE_QUERY_CANCELED),
224 errmsg("timed out while waiting for target LSN %X/%08X to be flushed; current standby_flush LSN %X/%08X",
225 LSN_FORMAT_ARGS(lsn),
226 LSN_FORMAT_ARGS(currentLSN)));
227 break;
228
229 case WAIT_LSN_TYPE_PRIMARY_FLUSH:
230 ereport(ERROR,
231 errcode(ERRCODE_QUERY_CANCELED),
232 errmsg("timed out while waiting for target LSN %X/%08X to be flushed; current primary_flush LSN %X/%08X",
233 LSN_FORMAT_ARGS(lsn),
234 LSN_FORMAT_ARGS(currentLSN)));
235 break;
236
237 default:
238 elog(ERROR, "unexpected wait LSN type %d", lsnType);
239 }
240 }
241 else
242 result = "timeout";
243 break;
244
245 case WAIT_LSN_RESULT_NOT_IN_RECOVERY:
246 if (throw)
247 {
248 if (PromoteIsTriggered())
249 {
250 XLogRecPtr currentLSN = GetCurrentLSNForWaitType(lsnType);
251
252 switch (lsnType)
253 {
254 case WAIT_LSN_TYPE_STANDBY_REPLAY:
255 ereport(ERROR,
256 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
257 errmsg("recovery is not in progress"),
258 errdetail("Recovery ended before target LSN %X/%08X was replayed; last standby_replay LSN %X/%08X.",
259 LSN_FORMAT_ARGS(lsn),
260 LSN_FORMAT_ARGS(currentLSN)));
261 break;
262
263 case WAIT_LSN_TYPE_STANDBY_WRITE:
264 ereport(ERROR,
265 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
266 errmsg("recovery is not in progress"),
267 errdetail("Recovery ended before target LSN %X/%08X was written; last standby_write LSN %X/%08X.",
268 LSN_FORMAT_ARGS(lsn),
269 LSN_FORMAT_ARGS(currentLSN)));
270 break;
271
272 case WAIT_LSN_TYPE_STANDBY_FLUSH:
273 ereport(ERROR,
274 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
275 errmsg("recovery is not in progress"),
276 errdetail("Recovery ended before target LSN %X/%08X was flushed; last standby_flush LSN %X/%08X.",
277 LSN_FORMAT_ARGS(lsn),
278 LSN_FORMAT_ARGS(currentLSN)));
279 break;
280
281 default:
282 elog(ERROR, "unexpected wait LSN type %d", lsnType);
283 }
284 }
285 else
286 {
287 switch (lsnType)
288 {
289 case WAIT_LSN_TYPE_STANDBY_REPLAY:
290 ereport(ERROR,
291 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
292 errmsg("recovery is not in progress"),
293 errhint("Waiting for the standby_replay LSN can only be executed during recovery."));
294 break;
295
296 case WAIT_LSN_TYPE_STANDBY_WRITE:
297 ereport(ERROR,
298 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
299 errmsg("recovery is not in progress"),
300 errhint("Waiting for the standby_write LSN can only be executed during recovery."));
301 break;
302
303 case WAIT_LSN_TYPE_STANDBY_FLUSH:
304 ereport(ERROR,
305 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
306 errmsg("recovery is not in progress"),
307 errhint("Waiting for the standby_flush LSN can only be executed during recovery."));
308 break;
309
310 default:
311 elog(ERROR, "unexpected wait LSN type %d", lsnType);
312 }
313 }
314 }
315 else
316 result = "not in recovery";
317 break;
318 }
319
320 /* need a tuple descriptor representing a single TEXT column */
321 tupdesc = WaitStmtResultDesc(stmt);
322
323 /* prepare for projection of tuples */
324 tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
325
326 /* Send it */
327 do_text_output_oneline(tstate, result);
328
329 end_tup_output(tstate);
330}
Assert
#define Assert(condition)
Definition c.h:873
int64
int64_t int64
Definition c.h:543
FLOAT8_FITS_IN_INT64
#define FLOAT8_FITS_IN_INT64(num)
Definition c.h:1085
unlikely
#define unlikely(x)
Definition c.h:412
defGetString
char * defGetString(DefElem *def)
Definition define.c:34
defGetBoolean
bool defGetBoolean(DefElem *def)
Definition define.c:93
errorConflictingDefElem
void errorConflictingDefElem(DefElem *defel, ParseState *pstate)
Definition define.c:370
errdetail
int errdetail(const char *fmt,...)
Definition elog.c:1216
errhint
int errhint(const char *fmt,...)
Definition elog.c:1330
errcode
int errcode(int sqlerrcode)
Definition elog.c:863
errmsg
int errmsg(const char *fmt,...)
Definition elog.c:1080
_
#define _(x)
Definition elog.c:91
ERROR
#define ERROR
Definition elog.h:39
elog
#define elog(elevel,...)
Definition elog.h:226
ereport
#define ereport(elevel,...)
Definition elog.h:150
TTSOpsVirtual
const TupleTableSlotOps TTSOpsVirtual
Definition execTuples.c:84
end_tup_output
void end_tup_output(TupOutputState *tstate)
Definition execTuples.c:2522
begin_tup_output_tupdesc
TupOutputState * begin_tup_output_tupdesc(DestReceiver *dest, TupleDesc tupdesc, const TupleTableSlotOps *tts_ops)
Definition execTuples.c:2444
do_text_output_oneline
#define do_text_output_oneline(tstate, str_to_emit)
Definition executor.h:628
DirectFunctionCall1
#define DirectFunctionCall1(func, arg1)
Definition fmgr.h:684
parse_real
bool parse_real(const char *value, double *result, int flags, const char **hintmsg)
Definition guc.c:2833
GUC_UNIT_MS
#define GUC_UNIT_MS
Definition guc.h:239
stmt
#define stmt
Definition indent_codes.h:59
parser_errposition
int parser_errposition(ParseState *pstate, int location)
Definition parse_node.c:106
foreach_node
#define foreach_node(type, var, lst)
Definition pg_list.h:496
pg_lsn_in
Datum pg_lsn_in(PG_FUNCTION_ARGS)
Definition pg_lsn.c:64
DatumGetLSN
static XLogRecPtr DatumGetLSN(Datum X)
Definition pg_lsn.h:25
pg_strcasecmp
int pg_strcasecmp(const char *s1, const char *s2)
Definition pgstrcasecmp.c:32
CStringGetDatum
static Datum CStringGetDatum(const char *X)
Definition postgres.h:380
fb
static int fb(int x)
Definition preproc-init.c:92
ActiveSnapshotSet
bool ActiveSnapshotSet(void)
Definition snapmgr.c:812
HaveRegisteredOrActiveSnapshot
bool HaveRegisteredOrActiveSnapshot(void)
Definition snapmgr.c:1644
PopActiveSnapshot
void PopActiveSnapshot(void)
Definition snapmgr.c:775
InvalidateCatalogSnapshot
void InvalidateCatalogSnapshot(void)
Definition snapmgr.c:455
MyProc
PGPROC * MyProc
Definition proc.c:67
PGPROC::xmin
TransactionId xmin
Definition proc.h:194
InvalidTransactionId
#define InvalidTransactionId
Definition transam.h:31
WaitStmtResultDesc
TupleDesc WaitStmtResultDesc(WaitStmt *stmt)
Definition wait.c:333
RecoveryInProgress
bool RecoveryInProgress(void)
Definition xlog.c:6460
LSN_FORMAT_ARGS
#define LSN_FORMAT_ARGS(lsn)
Definition xlogdefs.h:47
XLogRecPtr
uint64 XLogRecPtr
Definition xlogdefs.h:21
PromoteIsTriggered
bool PromoteIsTriggered(void)
Definition xlogrecovery.c:4459
GetCurrentLSNForWaitType
XLogRecPtr GetCurrentLSNForWaitType(WaitLSNType lsnType)
Definition xlogwait.c:88
WaitForLSN
WaitLSNResult WaitForLSN(WaitLSNType lsnType, XLogRecPtr targetLSN, int64 timeout)
Definition xlogwait.c:375
WaitLSNResult
WaitLSNResult
Definition xlogwait.h:26
WAIT_LSN_RESULT_NOT_IN_RECOVERY
@ WAIT_LSN_RESULT_NOT_IN_RECOVERY
Definition xlogwait.h:28
WAIT_LSN_RESULT_TIMEOUT
@ WAIT_LSN_RESULT_TIMEOUT
Definition xlogwait.h:30
WAIT_LSN_RESULT_SUCCESS
@ WAIT_LSN_RESULT_SUCCESS
Definition xlogwait.h:27
WaitLSNType
WaitLSNType
Definition xlogwait.h:37
WAIT_LSN_TYPE_PRIMARY_FLUSH
@ WAIT_LSN_TYPE_PRIMARY_FLUSH
Definition xlogwait.h:44
WAIT_LSN_TYPE_STANDBY_REPLAY
@ WAIT_LSN_TYPE_STANDBY_REPLAY
Definition xlogwait.h:39
WAIT_LSN_TYPE_STANDBY_FLUSH
@ WAIT_LSN_TYPE_STANDBY_FLUSH
Definition xlogwait.h:41
WAIT_LSN_TYPE_STANDBY_WRITE
@ WAIT_LSN_TYPE_STANDBY_WRITE
Definition xlogwait.h:40

References _, ActiveSnapshotSet(), Assert, begin_tup_output_tupdesc(), CStringGetDatum(), DatumGetLSN(), defGetBoolean(), defGetString(), DirectFunctionCall1, do_text_output_oneline, elog, end_tup_output(), ereport, errcode(), errdetail(), errhint(), errmsg(), ERROR, errorConflictingDefElem(), fb(), FLOAT8_FITS_IN_INT64, foreach_node, GetCurrentLSNForWaitType(), GUC_UNIT_MS, HaveRegisteredOrActiveSnapshot(), InvalidateCatalogSnapshot(), InvalidTransactionId, LSN_FORMAT_ARGS, MyProc, parse_real(), parser_errposition(), pg_lsn_in(), pg_strcasecmp(), PopActiveSnapshot(), PromoteIsTriggered(), RecoveryInProgress(), stmt, TTSOpsVirtual, unlikely, WAIT_LSN_RESULT_NOT_IN_RECOVERY, WAIT_LSN_RESULT_SUCCESS, WAIT_LSN_RESULT_TIMEOUT, WAIT_LSN_TYPE_PRIMARY_FLUSH, WAIT_LSN_TYPE_STANDBY_FLUSH, WAIT_LSN_TYPE_STANDBY_REPLAY, WAIT_LSN_TYPE_STANDBY_WRITE, WaitForLSN(), WaitStmtResultDesc(), and PGPROC::xmin.

Referenced by standard_ProcessUtility().

◆ WaitStmtResultDesc()

TupleDesc WaitStmtResultDesc ( WaitStmt *  stmt)
extern

Definition at line 333 of file wait.c.

334{
335 TupleDesc tupdesc;
336
337 /* Need a tuple descriptor representing a single TEXT column */
338 tupdesc = CreateTemplateTupleDesc(1);
339 TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
340 TEXTOID, -1, 0);
341 return tupdesc;
342}
AttrNumber
int16 AttrNumber
Definition attnum.h:21
CreateTemplateTupleDesc
TupleDesc CreateTemplateTupleDesc(int natts)
Definition tupdesc.c:182
TupleDescInitEntry
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Definition tupdesc.c:842

References CreateTemplateTupleDesc(), fb(), and TupleDescInitEntry().

Referenced by ExecWaitStmt(), and UtilityTupleDescriptor().