PolarDB-for-PostgreSQL

1
/*
2
 * contrib/pgrowlocks/pgrowlocks.c
3
 *
4
 * Copyright (c) 2005-2006	Tatsuo Ishii
5
 *
6
 * Permission to use, copy, modify, and distribute this software and
7
 * its documentation for any purpose, without fee, and without a
8
 * written agreement is hereby granted, provided that the above
9
 * copyright notice and this paragraph and the following two
10
 * paragraphs appear in all copies.
11
 *
12
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE TO ANY PARTY FOR DIRECT,
13
 * INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING
14
 * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
15
 * DOCUMENTATION, EVEN IF THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED
16
 * OF THE POSSIBILITY OF SUCH DAMAGE.
17
 *
18
 * THE AUTHOR SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT
19
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
20
 * A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS ON AN "AS
21
 * IS" BASIS, AND THE AUTHOR HAS NO OBLIGATIONS TO PROVIDE MAINTENANCE,
22
 * SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
23
 */
24

25
#include "postgres.h"
26

27
#include "access/multixact.h"
28
#include "access/relscan.h"
29
#include "access/xact.h"
30
#include "catalog/namespace.h"
31
#include "catalog/pg_authid.h"
32
#include "funcapi.h"
33
#include "miscadmin.h"
34
#include "pgstat.h"
35
#include "storage/bufmgr.h"
36
#include "storage/procarray.h"
37
#include "utils/acl.h"
38
#include "utils/builtins.h"
39
#include "utils/rel.h"
40
#include "utils/snapmgr.h"
41
#include "utils/tqual.h"
42
#include "utils/varlena.h"
43

44
PG_MODULE_MAGIC;
45

46
PG_FUNCTION_INFO_V1(pgrowlocks);
47

48
/* ----------
49
 * pgrowlocks:
50
 * returns tids of rows being locked
51
 * ----------
52
 */
53

54
#define NCHARS 32
55

56
#define		Atnum_tid		0
57
#define		Atnum_xmax		1
58
#define		Atnum_ismulti	2
59
#define		Atnum_xids		3
60
#define		Atnum_modes		4
61
#define		Atnum_pids		5
62

63
Datum
64
pgrowlocks(PG_FUNCTION_ARGS)
65
{
66
	text	   *relname = PG_GETARG_TEXT_PP(0);
67
	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
68
	bool		randomAccess;
69
	TupleDesc	tupdesc;
70
	Tuplestorestate *tupstore;
71
	AttInMetadata *attinmeta;
72
	Relation	rel;
73
	RangeVar   *relrv;
74
	HeapScanDesc scan;
75
	HeapTuple	tuple;
76
	MemoryContext oldcontext;
77
	AclResult	aclresult;
78
	char	  **values;
79

80
	/* check to see if caller supports us returning a tuplestore */
81
	if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
82
		ereport(ERROR,
83
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
84
				 errmsg("set-valued function called in context that cannot accept a set")));
85
	if (!(rsinfo->allowedModes & SFRM_Materialize))
86
		ereport(ERROR,
87
				(errcode(ERRCODE_SYNTAX_ERROR),
88
				 errmsg("materialize mode required, but it is not allowed in this context")));
89

90
	/* The tupdesc and tuplestore must be created in ecxt_per_query_memory */
91
	oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
92

93
	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
94
		elog(ERROR, "return type must be a row type");
95

96
	randomAccess = (rsinfo->allowedModes & SFRM_Materialize_Random) != 0;
97
	tupstore = tuplestore_begin_heap(randomAccess, false, work_mem);
98
	rsinfo->returnMode = SFRM_Materialize;
99
	rsinfo->setResult = tupstore;
100
	rsinfo->setDesc = tupdesc;
101

102
	MemoryContextSwitchTo(oldcontext);
103

104
	/* Access the table */
105
	relrv = makeRangeVarFromNameList(textToQualifiedNameList(relname));
106
	rel = relation_openrv(relrv, AccessShareLock);
107

108
	if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
109
		ereport(ERROR,
110
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
111
				 errmsg("\"%s\" is a partitioned table",
112
						RelationGetRelationName(rel)),
113
				 errdetail("Partitioned tables do not contain rows.")));
114
	else if (rel->rd_rel->relkind != RELKIND_RELATION)
115
		ereport(ERROR,
116
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
117
				 errmsg("\"%s\" is not a table",
118
						RelationGetRelationName(rel))));
119

120
	/*
121
	 * check permissions: must have SELECT on table or be in
122
	 * pg_stat_scan_tables
123
	 */
124
	aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
125
								  ACL_SELECT);
126
	if (aclresult != ACLCHECK_OK)
127
		aclresult = is_member_of_role(GetUserId(), DEFAULT_ROLE_STAT_SCAN_TABLES) ? ACLCHECK_OK : ACLCHECK_NO_PRIV;
128

129
	if (aclresult != ACLCHECK_OK)
130
		aclcheck_error(aclresult, get_relkind_objtype(rel->rd_rel->relkind),
131
					   RelationGetRelationName(rel));
132

133
	/* Scan the relation */
134
	scan = heap_beginscan(rel, GetActiveSnapshot(), 0, NULL);
135

136
	attinmeta = TupleDescGetAttInMetadata(tupdesc);
137

138
	values = (char **) palloc(tupdesc->natts * sizeof(char *));
139

140
	while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
141
	{
142
		HTSU_Result htsu;
143
		TransactionId xmax;
144
		uint16		infomask;
145

146
		/* must hold a buffer lock to call HeapTupleSatisfiesUpdate */
147
		LockBuffer(scan->rs_cbuf, BUFFER_LOCK_SHARE);
148

149
		htsu = HeapTupleSatisfiesUpdate(tuple,
150
										GetCurrentCommandId(false),
151
										scan->rs_cbuf);
152
		xmax = HeapTupleHeaderGetRawXmax(tuple->t_data);
153
		infomask = tuple->t_data->t_infomask;
154

155
		/*
156
		 * A tuple is locked if HTSU returns BeingUpdated.
157
		 */
158
		if (htsu == HeapTupleBeingUpdated)
159
		{
160
			values[Atnum_tid] = (char *) DirectFunctionCall1(tidout,
161
															 PointerGetDatum(&tuple->t_self));
162

163
			values[Atnum_xmax] = palloc(NCHARS * sizeof(char));
164
			snprintf(values[Atnum_xmax], NCHARS, "%d", xmax);
165
			if (infomask & HEAP_XMAX_IS_MULTI)
166
			{
167
				MultiXactMember *members;
168
				int			nmembers;
169
				bool		first = true;
170
				bool		allow_old;
171

172
				values[Atnum_ismulti] = pstrdup("true");
173

174
				allow_old = HEAP_LOCKED_UPGRADED(infomask);
175
				nmembers = GetMultiXactIdMembers(xmax, &members, allow_old,
176
												 false);
177
				if (nmembers == -1)
178
				{
179
					values[Atnum_xids] = "{0}";
180
					values[Atnum_modes] = "{transient upgrade status}";
181
					values[Atnum_pids] = "{0}";
182
				}
183
				else
184
				{
185
					int			j;
186

187
					values[Atnum_xids] = palloc(NCHARS * nmembers);
188
					values[Atnum_modes] = palloc(NCHARS * nmembers);
189
					values[Atnum_pids] = palloc(NCHARS * nmembers);
190

191
					strcpy(values[Atnum_xids], "{");
192
					strcpy(values[Atnum_modes], "{");
193
					strcpy(values[Atnum_pids], "{");
194

195
					for (j = 0; j < nmembers; j++)
196
					{
197
						char		buf[NCHARS];
198

199
						if (!first)
200
						{
201
							strcat(values[Atnum_xids], ",");
202
							strcat(values[Atnum_modes], ",");
203
							strcat(values[Atnum_pids], ",");
204
						}
205
						snprintf(buf, NCHARS, "%d", members[j].xid);
206
						strcat(values[Atnum_xids], buf);
207
						switch (members[j].status)
208
						{
209
							case MultiXactStatusUpdate:
210
								snprintf(buf, NCHARS, "Update");
211
								break;
212
							case MultiXactStatusNoKeyUpdate:
213
								snprintf(buf, NCHARS, "No Key Update");
214
								break;
215
							case MultiXactStatusForUpdate:
216
								snprintf(buf, NCHARS, "For Update");
217
								break;
218
							case MultiXactStatusForNoKeyUpdate:
219
								snprintf(buf, NCHARS, "For No Key Update");
220
								break;
221
							case MultiXactStatusForShare:
222
								snprintf(buf, NCHARS, "Share");
223
								break;
224
							case MultiXactStatusForKeyShare:
225
								snprintf(buf, NCHARS, "Key Share");
226
								break;
227
						}
228
						strcat(values[Atnum_modes], buf);
229
						snprintf(buf, NCHARS, "%d",
230
								 polar_pgstat_get_virtual_pid(BackendXidGetPid(members[j].xid), false));
231
						strcat(values[Atnum_pids], buf);
232

233
						first = false;
234
					}
235

236
					strcat(values[Atnum_xids], "}");
237
					strcat(values[Atnum_modes], "}");
238
					strcat(values[Atnum_pids], "}");
239
				}
240
			}
241
			else
242
			{
243
				values[Atnum_ismulti] = pstrdup("false");
244

245
				values[Atnum_xids] = palloc(NCHARS * sizeof(char));
246
				snprintf(values[Atnum_xids], NCHARS, "{%d}", xmax);
247

248
				values[Atnum_modes] = palloc(NCHARS);
249
				if (infomask & HEAP_XMAX_LOCK_ONLY)
250
				{
251
					if (HEAP_XMAX_IS_SHR_LOCKED(infomask))
252
						snprintf(values[Atnum_modes], NCHARS, "{For Share}");
253
					else if (HEAP_XMAX_IS_KEYSHR_LOCKED(infomask))
254
						snprintf(values[Atnum_modes], NCHARS, "{For Key Share}");
255
					else if (HEAP_XMAX_IS_EXCL_LOCKED(infomask))
256
					{
257
						if (tuple->t_data->t_infomask2 & HEAP_KEYS_UPDATED)
258
							snprintf(values[Atnum_modes], NCHARS, "{For Update}");
259
						else
260
							snprintf(values[Atnum_modes], NCHARS, "{For No Key Update}");
261
					}
262
					else
263
						/* neither keyshare nor exclusive bit it set */
264
						snprintf(values[Atnum_modes], NCHARS,
265
								 "{transient upgrade status}");
266
				}
267
				else
268
				{
269
					if (tuple->t_data->t_infomask2 & HEAP_KEYS_UPDATED)
270
						snprintf(values[Atnum_modes], NCHARS, "{Update}");
271
					else
272
						snprintf(values[Atnum_modes], NCHARS, "{No Key Update}");
273
				}
274

275
				values[Atnum_pids] = palloc(NCHARS * sizeof(char));
276
				snprintf(values[Atnum_pids], NCHARS, "{%d}",
277
						 polar_pgstat_get_virtual_pid(BackendXidGetPid(xmax), false));
278
			}
279

280
			LockBuffer(scan->rs_cbuf, BUFFER_LOCK_UNLOCK);
281

282
			/* build a tuple */
283
			tuple = BuildTupleFromCStrings(attinmeta, values);
284
			tuplestore_puttuple(tupstore, tuple);
285
		}
286
		else
287
		{
288
			LockBuffer(scan->rs_cbuf, BUFFER_LOCK_UNLOCK);
289
		}
290
	}
291

292
	heap_endscan(scan);
293
	heap_close(rel, AccessShareLock);
294
	return (Datum) 0;
295
}
296