Repository navigation
Expand file tree
/
Copy pathpg_query_stack.c
More file actions
784 lines (691 loc) · 41.6 KB
/
Copy pathpg_query_stack.c
File metadata and controls
784 lines (691 loc) · 41.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
/*
* pg_query_stack.c - расширение PostgreSQL для получения стека SQL-запросов
* текущего backend-процесса.
*
* Расширение позволяет в любой момент - из триггера, PL/pgSQL-функции,
* SPI-вызова - узнать полный список SQL-запросов, выполняющихся в данный
* момент в этом же сеансе. Это удобно для отладки, аудита и мониторинга.
*
* Горячий путь (выполняется при каждом запросе):
* - ExecutorRun: кладет указатель на QueryDesc в кольцевой буфер перед
* вызовом downstream и снимает его после возврата
* - ExecutorFinish: то же самое на время ExecutorFinish, внутри которого
* срабатывают AFTER-триггеры
* - SubXactCallback: сохраняет/восстанавливает позицию буфера при
* открытии/откате вложенных транзакций (SAVEPOINT)
* - XactCallback: сбрасывает буфер в ноль при завершении транзакции
*
* Фрейм живет ровно столько, сколько исполняется запрос: пока downstream
* ExecutorRun или ExecutorFinish не вернул управление. Порядок push/pop
* поэтому совпадает с порядком вызовов на C-стеке и не зависит от жизненного
* цикла порталов: открытый, но простаивающий курсор (DECLARE или OPEN без
* FETCH) фреймом не является, CLOSE в любом порядке не оставляет следа, а
* указатель в кольце всегда ведет на живой QueryDesc, потому что его
* исполнитель в этот момент находится в середине собственного Run/Finish.
*
* Путь чтения (вызывается пользователем через SQL):
* - Снимок `ring_head` фиксируется один раз в начале вызова функции
* - Перебираются слоты [0, max_emit) в контексте multi_call_memory_ctx
* - Три защитные проверки на слот (qd / qd->estate / qd->sourceText),
* при срабатывании строка-заполнитель (не NULL, чтобы не ломать счетчик)
* - Текст запроса копируется через pstrdup только при чтении, не при записи
*
* Объем памяти в BSS: 100 x 8 байт кольцо + 256 x 8 байт LIFO субтранзакций
* + два int и флаг ~ 2857 байт.
*
* Важное поведение: pg_query_stack(), вызванный из чужого ExecutorStart- или
* ExecutorEnd-хука, текущий запрос не видит. Его фрейм существует только
* внутри ExecutorRun и ExecutorFinish.
*/
#include "postgres.h"
#include "fmgr.h"
#include "funcapi.h"
#include "executor/executor.h"
#include "nodes/execnodes.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/guc.h"
#include "access/parallel.h"
#include "access/xact.h"
PG_MODULE_MAGIC;
/* ============================================================================
* Константы
* ============================================================================ */
/*
* Максимальная глубина кольцевого буфера запросов.
* 100 уровней хватает для любого практического сценария вложенности.
* Если реальная глубина превысит это значение, более глубокие фреймы в кольцо
* не попадают: хук видит заполненное кольцо, передает управление downstream
* без записи, и снимать ему после возврата нечего. В стеке остаются 100
* самых внешних фреймов.
*/
#define MAX_QUERY_STACK_DEPTH 100
/*
* Максимальная глубина LIFO-стека снимков субтранзакций.
* Каждый SAVEPOINT (или PL/pgSQL EXCEPTION-блок) добавляет одну запись.
* 256 уровней вложенности - практически недостижимый предел для реального кода.
* Глубже снимок сохранить некуда: с этого момента и до конца транзакции стек
* считается недостоверным (pgs_ring_unreliable), pg_query_stack() возвращает
* пустой результат, а хуки продолжают работать в обычном режиме.
*/
#define MAX_SUBXACT_STACK_DEPTH 256
/* ============================================================================
* Глобальные переменные
* ============================================================================ */
/*
* GUC-параметр pg_query_stack.enabled.
* Позволяет отключить трекинг на уровне отдельного сеанса без перезапуска
* сервера: SET pg_query_stack.enabled = off;
* Когда расширение отключено, накладные расходы на горячем пути близки к нулю.
*/
static bool pg_query_stack_enabled = true;
/*
* Объединенный флаг горячего пути: enabled && !IsParallelWorker().
*
* Почему не проверять IsParallelWorker() прямо в хуках?
* IsParallelWorker() - константа на все время жизни процесса, поэтому
* достаточно вычислить ее один раз при инициализации (и при смене GUC).
* Один флаг pg_query_stack_active - это одна ветка вместо двух на горячем
* пути, что ощутимо при тысячах коротких запросов в секунду.
*
* Почему параллельные воркеры исключаются?
* У воркера свое кольцо (BSS на процесс), контекста лидера в нем нет:
* фрагмент плана выглядел бы как запрос верхнего уровня без вызвавшей его
* цепочки. Такой стек только вводит в заблуждение, поэтому воркеры не пишут
* в кольцо, а pg_query_stack() в них возвращает пустой результат.
*/
static bool pg_query_stack_active = true;
/* ============================================================================
* Основные структуры данных: кольцевой буфер + LIFO-стек субтранзакций
* ============================================================================ */
/*
* PgsRingSlot - одна ячейка кольцевого буфера.
*
* Хранит только указатель на QueryDesc текущего запроса.
* Текст запроса (qd->sourceText) намеренно НЕ копируется здесь:
* копирование - дорогая операция, а pg_query_stack() вызывается редко.
* Поэтому текст копируется через pstrdup только в момент чтения (SRF).
*/
typedef struct PgsRingSlot
{
QueryDesc *qd;
} PgsRingSlot;
/*
* Кольцевой буфер активных запросов бэкенда.
* Размещен в BSS (статика, инициализируется нулями - heap-аллокация не нужна).
* Объем: 100 x 8 байт = 800 байт.
*
* pgs_ring_head - текущая глубина заполнения (= глубина стека запросов).
* Хук ExecutorRun или ExecutorFinish делает push (ring[head].qd = qd; head++)
* перед вызовом downstream и возвращает head к прежнему значению после.
*/
static PgsRingSlot pgs_ring[MAX_QUERY_STACK_DEPTH];
static int pgs_ring_head = 0;
/*
* PgsRingSubxactSnap - снимок состояния кольца на момент начала субтранзакции.
*
* Зачем нужны снимки?
* Когда PL/pgSQL-блок BEGIN ... EXCEPTION WHEN OTHERS ... END перехватывает
* ошибку, PostgreSQL откатывает внутреннюю субтранзакцию через longjmp, и
* хук не доходит до строки, снимающей фрейм после возврата downstream.
* Фреймы в кольце "зависают". SubXactCallback при событии ABORT_SUB
* восстанавливает ring_head из снимка, сделанного при старте субтранзакции, -
* этим одним присваиванием отсекаются все "зависшие" фреймы внутри отмененного
* блока. Внутри субтранзакции head не опускается ниже снимка: фреймы под ним
* принадлежат вызовам, которые еще не вернулись на C-стеке, так что
* восстановление никогда не поднимает head и не воскрешает снятые слоты.
*
* Почему LIFO-стек, а не одна переменная-снимок?
* При нескольких уровнях вложенных SAVEPOINT каждый вложенный уровень должен
* иметь свой снимок. Одна переменная перезаписывалась бы на каждом START_SUB
* и при откате внутреннего уровня теряла бы снимок внешнего.
*
* Объем: 256 x (4 + 4) байт = 2048 байт в BSS.
*/
typedef struct PgsRingSubxactSnap
{
SubTransactionId subxid; /* идентификатор субтранзакции */
int ring_head; /* значение ring_head на момент START_SUB */
} PgsRingSubxactSnap;
static PgsRingSubxactSnap pgs_ring_subxact_snap[MAX_SUBXACT_STACK_DEPTH];
static int pgs_ring_subxact_top = 0;
/*
* Флаг "стек недостоверен". Выставляется в pgs_subxact_cb, когда LIFO снимков
* переполнен и снимок очередной субтранзакции сохранить некуда: без снимка
* ABORT_SUB не сможет снять фреймы запросов, упавших внутри нее, и в кольце
* может остаться указатель на освобожденный QueryDesc. Пока флаг стоит,
* pg_query_stack() возвращает пустой результат, хуки работают как обычно.
* Сбрасывается вместе с остальным состоянием в pgs_xact_cb.
*/
static bool pgs_ring_unreliable = false;
/*
* Переменные для цепочки хуков.
*
* prev_* - хук, который стоял до нас (может быть NULL, если мы первые).
* Используется только в _PG_fini для безопасного восстановления.
* chained_* - "готовый" указатель для вызова downstream: prev_* если он есть,
* иначе standard_ExecutorRun или standard_ExecutorFinish.
*
* Почему pre-bake chained_*, а не проверять prev_* на каждом вызове?
* Проверка `if (prev != NULL) prev(...) else standard(...)` - это лишняя
* ветка на горячем пути при каждом запросе. Один раз вычислив chained_*
* при загрузке расширения, горячий путь делает один unconditional indirect call.
*/
static ExecutorRun_hook_type prev_ExecutorRun = NULL;
static ExecutorRun_hook_type chained_ExecutorRun = NULL;
static ExecutorFinish_hook_type prev_ExecutorFinish = NULL;
static ExecutorFinish_hook_type chained_ExecutorFinish = NULL;
/* ============================================================================
* Предварительные объявления функций
* ============================================================================ */
void _PG_init(void);
void _PG_fini(void);
static void pg_query_stack_assign_enabled(bool newval, void *extra);
Datum pg_query_stack(PG_FUNCTION_ARGS);
/*
* Сигнатура ExecutorRun: в PG 18 у нее нет параметра execute_once.
*/
#if PG_VERSION_NUM >= 180000
#define PGS_RUN_PARAMS QueryDesc *queryDesc, ScanDirection direction, uint64 count
#define PGS_RUN_ARGS queryDesc, direction, count
#else
#define PGS_RUN_PARAMS QueryDesc *queryDesc, ScanDirection direction, uint64 count, \
bool execute_once
#define PGS_RUN_ARGS queryDesc, direction, count, execute_once
#endif
static void pg_query_stack_ExecutorRun(PGS_RUN_PARAMS);
static void pg_query_stack_ExecutorFinish(QueryDesc *queryDesc);
static void pgs_subxact_cb(SubXactEvent event, SubTransactionId mySubid,
SubTransactionId parentSubid, void *arg);
static void pgs_xact_cb(XactEvent event, void *arg);
/* ============================================================================
* Assign-хук для GUC pg_query_stack.enabled
* ============================================================================ */
/*
* Вызывается PostgreSQL при каждом изменении GUC-параметра.
* Пересчитывает pg_query_stack_active - объединенный флаг горячего пути.
* IsParallelWorker() не меняется за время жизни процесса, поэтому
* пересчитывать его при каждом GUC-изменении достаточно и безопасно.
*/
static void
pg_query_stack_assign_enabled(bool newval, void *extra)
{
pg_query_stack_enabled = newval;
pg_query_stack_active = newval && !IsParallelWorker();
}
/* ============================================================================
* Инициализация и выгрузка расширения
* ============================================================================ */
void
_PG_init(void)
{
DefineCustomBoolVariable(
"pg_query_stack.enabled",
"Включить или отключить трекинг стека запросов",
"При отключении накладные расходы близки к нулю. "
"Можно менять на уровне сеанса без перезапуска сервера.",
&pg_query_stack_enabled,
true,
PGC_USERSET,
0,
NULL, /* check_hook - проверка не нужна */
pg_query_stack_assign_enabled, /* assign_hook - пересчет active */
NULL /* show_hook */
);
pg_query_stack_active = pg_query_stack_enabled && !IsParallelWorker();
/*
* Сохраняем текущие хуки до регистрации наших (они могут быть уже заняты
* другим расширением). Затем вычисляем готовый chained_*-указатель -
* это позволит горячему пути вызывать downstream одним indirect call,
* без проверки "а есть ли prev?" при каждом запросе.
*/
prev_ExecutorRun = ExecutorRun_hook;
prev_ExecutorFinish = ExecutorFinish_hook;
chained_ExecutorRun = prev_ExecutorRun
? prev_ExecutorRun
: standard_ExecutorRun;
chained_ExecutorFinish = prev_ExecutorFinish
? prev_ExecutorFinish
: standard_ExecutorFinish;
ExecutorRun_hook = pg_query_stack_ExecutorRun;
ExecutorFinish_hook = pg_query_stack_ExecutorFinish;
RegisterSubXactCallback(pgs_subxact_cb, NULL);
RegisterXactCallback(pgs_xact_cb, NULL);
/* Убеждаемся, что хуки установлены именно наши (cassert-сборки). */
Assert(ExecutorRun_hook == pg_query_stack_ExecutorRun);
Assert(ExecutorFinish_hook == pg_query_stack_ExecutorFinish);
}
void
_PG_fini(void)
{
/*
* Восстанавливаем хуки только через prev_*, а не через chained_*.
* Почему? chained_* может указывать на standard_ExecutorRun/Finish, а
* запись standard_* обратно в глобальный хук означала бы потерю хуков
* всех других расширений, зарегистрированных после нас.
* Восстановление через prev_* корректно возвращает хук к тому,
* что было до нашей загрузки.
*/
if (ExecutorRun_hook == pg_query_stack_ExecutorRun)
ExecutorRun_hook = prev_ExecutorRun;
if (ExecutorFinish_hook == pg_query_stack_ExecutorFinish)
ExecutorFinish_hook = prev_ExecutorFinish;
chained_ExecutorRun = NULL;
chained_ExecutorFinish = NULL;
UnregisterXactCallback(pgs_xact_cb, NULL);
UnregisterSubXactCallback(pgs_subxact_cb, NULL);
pgs_ring_head = 0;
pgs_ring_subxact_top = 0;
pgs_ring_unreliable = false;
}
/* ============================================================================
* Хуки горячего пути: ExecutorRun и ExecutorFinish.
* push перед вызовом downstream, pop после возврата. Никаких аллокаций
* в heap, никаких PG_TRY/PG_CATCH: если downstream уходит через longjmp,
* фрейм убирают колбэки субтранзакции и транзакции, восстанавливая ring_head
* по снимку.
* ============================================================================ */
/*
* Хук ExecutorRun - вызывается PostgreSQL на каждый прогон исполнителя:
* один раз для обычного запроса, по разу на каждый FETCH или порцию строк
* для курсора и приостановленного портала.
*
* Последовательность действий:
* 1. Проверяем флаг pg_query_stack_active (трекинг включен и не параллельный воркер).
* 2. Если кольцо заполнено, передаем управление дальше без записи: снимать
* после возврата тогда тоже нечего.
* 3. Иначе кладем qd в слот head, увеличиваем head и вызываем downstream.
* 4. После возврата записываем в head сохраненный индекс. Именно запись, а не
* head--: вложенные фреймы к этому моменту уже сняты своими хуками, а если
* внутри откатывалась субтранзакция, ее колбэк уже вернул head к снимку,
* сделанному не ниже нашего индекса + 1.
*
* Ошибка внутри downstream (longjmp) до шага 4 не доходит. Фрейм остается в
* кольце до SUBXACT_EVENT_ABORT_SUB или XACT_EVENT_ABORT, где ring_head
* восстанавливается по снимку. Между ошибкой и колбэком пользовательский код
* не выполняется, так что "зависший" фрейм никто не читает. Исключение -
* C-код, перехватывающий ERROR без субтранзакции и продолжающий работу
* (plperl_return_next): такой фрейм живет до возврата внешнего Run/Finish
* или до конца транзакции. PG_TRY в обертках нет намеренно: sigsetjmp на
* каждом запросе дороже, чем эта граница.
*/
static void
pg_query_stack_ExecutorRun(PGS_RUN_PARAMS)
{
int idx;
Assert(queryDesc != NULL); /* A2: queryDesc обязан быть не NULL */
if (unlikely(!pg_query_stack_active) ||
unlikely(pgs_ring_head >= MAX_QUERY_STACK_DEPTH))
{
chained_ExecutorRun(PGS_RUN_ARGS);
return;
}
idx = pgs_ring_head;
Assert(idx >= 0); /* A2: head не может уйти в минус */
pgs_ring[idx].qd = queryDesc;
pgs_ring_head = idx + 1;
chained_ExecutorRun(PGS_RUN_ARGS);
Assert(pgs_ring_unreliable || pgs_ring_head == idx + 1); /* A3: вложенные фреймы сняты симметрично */
pgs_ring_head = idx;
}
/*
* Хук ExecutorFinish - вызывается PostgreSQL после последнего ExecutorRun
* запроса и до ExecutorEnd. Внутри standard_ExecutorFinish срабатывают
* AFTER-триггеры (AfterTriggerEndQuery), поэтому запрос должен быть виден
* в стеке и здесь: аудит-триггер, читающий pg_query_stack(), получает свой
* UPDATE или DELETE как фрейм. Схема та же, что у ExecutorRun.
*/
static void
pg_query_stack_ExecutorFinish(QueryDesc *queryDesc)
{
int idx;
Assert(queryDesc != NULL); /* A2 */
if (unlikely(!pg_query_stack_active) ||
unlikely(pgs_ring_head >= MAX_QUERY_STACK_DEPTH))
{
chained_ExecutorFinish(queryDesc);
return;
}
idx = pgs_ring_head;
Assert(idx >= 0);
pgs_ring[idx].qd = queryDesc;
pgs_ring_head = idx + 1;
chained_ExecutorFinish(queryDesc);
Assert(pgs_ring_unreliable || pgs_ring_head == idx + 1); /* A3 */
pgs_ring_head = idx;
}
/*
* Колбэк субтранзакций - LIFO-снимок/восстановление ring_head.
*
* Зачем этот колбэк нужен?
* Когда PL/pgSQL EXCEPTION WHEN OTHERS перехватывает ошибку из AFTER-триггера
* или вложенного SQL, PostgreSQL откатывает субтранзакцию через longjmp.
* Хук ExecutorRun или ExecutorFinish упавшего запроса не доходит до строки,
* снимающей фрейм после возврата downstream, и фрейм "зависает" в кольце.
*
* SubXactCallback с событием SUBXACT_EVENT_ABORT_SUB срабатывает ДО того,
* как управление перейдет в EXCEPTION-блок. Восстановление ring_head из
* снимка, сделанного при START_SUB, одним присваиванием отсекает все
* "зависшие" фреймы внутри отмененного блока.
*
* PostgreSQL гарантирует вызов SubXactCallback в строгом LIFO-порядке.
* Ассерт A5 проверяет это в cassert-сборках: верхний снимок LIFO-стека
* при ABORT_SUB всегда должен соответствовать mySubid.
*
* Нет аллокаций в heap, нет PG_TRY/PG_CATCH - это колбэк из критического
* пути обработки ошибок, он должен быть простым и предсказуемым.
*/
static void
pgs_subxact_cb(SubXactEvent event, SubTransactionId mySubid,
SubTransactionId parentSubid, void *arg)
{
int i;
int found_index;
(void) parentSubid;
(void) arg;
switch (event)
{
case SUBXACT_EVENT_START_SUB:
/*
* Начало субтранзакции (SAVEPOINT или EXCEPTION-блок).
* Сохраняем снимок: при откате этой субтранзакции мы знаем,
* к какому ring_head нужно вернуться.
* Если LIFO-стек переполнен, снимок сохранить некуда: ABORT_SUB
* этой субтранзакции не сможет снять ее фреймы, поэтому стек
* помечается недостоверным до конца транзакции (тест 032).
*/
if (unlikely(pgs_ring_subxact_top >= MAX_SUBXACT_STACK_DEPTH))
{
pgs_ring_unreliable = true;
break;
}
pgs_ring_subxact_snap[pgs_ring_subxact_top].subxid = mySubid;
pgs_ring_subxact_snap[pgs_ring_subxact_top].ring_head = pgs_ring_head;
pgs_ring_subxact_top++;
break;
case SUBXACT_EVENT_ABORT_SUB:
/*
* A5: PostgreSQL гарантирует LIFO - верхний снимок обязан
* соответствовать текущему mySubid. После переполнения LIFO
* верхний снимок принадлежит внешнему уровню, и проверка
* отключается флагом недостоверности.
*/
Assert(pgs_ring_unreliable ||
pgs_ring_subxact_top == 0 ||
pgs_ring_subxact_snap[pgs_ring_subxact_top - 1].subxid == mySubid);
/*
* Ищем снимок для mySubid сверху вниз (backward scan).
* Defensive scan обрабатывает случай, когда снимок был потерян
* из-за переполнения LIFO-стека (found_index < 0 -> тихий no-op).
*/
found_index = -1;
for (i = pgs_ring_subxact_top - 1; i >= 0; i--)
{
if (pgs_ring_subxact_snap[i].subxid == mySubid)
{
found_index = i;
break;
}
}
if (unlikely(found_index < 0))
break; /* снимок был потерян из-за overflow - ничего не делаем */
/*
* Восстанавливаем ring_head до значения на момент START_SUB.
* Все фреймы, добавленные внутри этой субтранзакции, отсекаются
* одним присваиванием - никакого цикла, никаких pfree.
*/
pgs_ring_head = pgs_ring_subxact_snap[found_index].ring_head;
pgs_ring_subxact_top = found_index;
break;
case SUBXACT_EVENT_COMMIT_SUB:
/*
* Субтранзакция зафиксирована - снимок больше не нужен,
* удаляем его из LIFO-стека. ring_head не трогаем: фреймы,
* поднятые внутри зафиксированной субтранзакции, уже сняты
* своими хуками, в кольце остались только исполняющиеся вызовы.
*
* PRE_COMMIT_SUB не обрабатываем: между ним и COMMIT_SUB
* субтранзакция еще может упасть (например, удаленный RELEASE
* в postgres_fdw), и тогда снимок нужен для ABORT_SUB.
*/
for (i = pgs_ring_subxact_top - 1; i >= 0; i--)
{
if (pgs_ring_subxact_snap[i].subxid == mySubid)
{
pgs_ring_subxact_top = i;
break;
}
}
break;
default:
break;
}
}
/*
* Колбэк завершения транзакции - финальный сброс буфера.
*
* Зачем нужен, если есть SubXactCallback?
* SubXactCallback срабатывает только при наличии субтранзакций (SAVEPOINT).
* Если ошибка произошла в верхнеуровневой транзакции без EXCEPTION-блоков,
* SubXactCallback ABORT_SUB не вызывается. Этот колбэк гарантирует, что
* ring_head вернется в 0 при любом завершении транзакции: COMMIT, ABORT или
* PREPARE TRANSACTION (после PREPARE события COMMIT в этом бэкенде не будет,
* а следующая транзакция должна начаться с чистого состояния, тест 032).
*
* Почему достаточно просто обнулить два int и флаг, а не освобождать память?
* Кольцо хранит только сырые указатели на QueryDesc - не владеет памятью.
* QueryDesc принадлежит PostgreSQL и освобождается им же. Нам достаточно
* забыть об этих указателях, обнуление сбрасывает все состояние к начальному,
* в том числе флаг недостоверности после переполнения LIFO снимков.
*/
static void
pgs_xact_cb(XactEvent event, void *arg)
{
(void) arg;
if (event != XACT_EVENT_COMMIT &&
event != XACT_EVENT_ABORT &&
event != XACT_EVENT_PREPARE)
return;
pgs_ring_head = 0;
pgs_ring_subxact_top = 0;
pgs_ring_unreliable = false;
}
/* ============================================================================
* SQL-функция pg_query_stack() - Set-Returning Function (SRF)
* ============================================================================
*
* Читает кольцевой буфер pgs_ring[0..head-1] и возвращает строки
* (frame_number, query_text) - по одной на каждый активный запрос в стеке.
*
* Параметр skip_count задает, сколько фреймов пропустить с "верхушки" стека
* (т.е. с самого вложенного конца). По умолчанию skip_count=1 - скрывает
* сам вызов pg_query_stack() из результата. pg_self_query() передает 2,
* скрывая дополнительно свою собственную обертку.
*
* Важные принципы реализации:
*
* 1. СНИМОК ОДИН РАЗ (snapshot-once).
* Значение pgs_ring_head читается ОДИН РАЗ в начале FIRSTCALL и замораживается
* в max_emit. Если во время итерации SRF другой запрос изменит ring_head,
* это не повлияет на текущий вызов - мы уже зафиксировали границу.
*
* 2. ЛЕНИВОЕ КОПИРОВАНИЕ ТЕКСТА (lazy pstrdup).
* Текст запроса копируется через pstrdup ТОЛЬКО при вызове pg_query_stack(),
* НЕ в момент ExecutorRun. Это делает горячий путь максимально дешевым:
* push = запись одного указателя. Копирование происходит в контексте
* multi_call_memory_ctx, который живет все время выполнения SRF.
*
* Почему именно multi_call_memory_ctx, а не TopTransactionContext?
* TopTransactionContext уничтожается при COMMIT/ROLLBACK, а тексты нужны
* только во время одного вызова pg_query_stack(). multi_call_memory_ctx
* точно соответствует нужному времени жизни и освобождается автоматически.
*
* 3. ТРИ ЗАЩИТНЫЕ ПРОВЕРКИ (G1, G2, G3).
* Указатели в кольце сырые. Каждый фрейм принадлежит запросу, чей
* ExecutorRun или ExecutorFinish сейчас на C-стеке, поэтому QueryDesc и
* EState под ним живы. Проверки остаются страховкой инварианта и стоят
* по одной загрузке на слот:
*
* G1: qd != NULL - слот без записи (0 из BSS-инициализации). При push и pop
* на одном C-фрейме таких слотов ниже head не бывает.
*
* G2: qd->estate != NULL - standard_ExecutorEnd обнуляет qd->estate до
* освобождения памяти QueryDesc. NULL здесь означает, что в кольце
* оказался уже завершенный запрос, и читать sourceText через qd нельзя.
*
* G3: qd->sourceText != NULL - некоторые внутренние пути SPI передают
* NULL в поле sourceText. Проверяем явно, чтобы не разыменовать NULL.
*
* При срабатывании любой защиты строка в результате заменяется на
* текст-заполнитель. Строка НЕ пропускается - чтобы не нарушить нумерацию
* фреймов и row count.
*
* frame_number = 0 - самый внешний (top-level) запрос (наиболее старый),
* frame_number = max_emit-1 - самый вложенный из видимых.
*/
typedef struct PgsWalkerCtx
{
int max_emit;
char **texts;
} PgsWalkerCtx;
PG_FUNCTION_INFO_V1(pg_query_stack);
Datum
pg_query_stack(PG_FUNCTION_ARGS)
{
FuncCallContext *funcctx;
int skip_count;
/* SQL DEFAULT = 1, что скрывает из результата сам вызов pg_query_stack(). */
skip_count = PG_ARGISNULL(0) ? 0 : PG_GETARG_INT32(0);
if (skip_count < 0)
skip_count = 0;
else if (skip_count > MAX_QUERY_STACK_DEPTH)
skip_count = MAX_QUERY_STACK_DEPTH;
if (SRF_IS_FIRSTCALL())
{
MemoryContext oldcontext;
TupleDesc tupdesc;
PgsWalkerCtx *wctx;
int snapshot_head;
int max_emit;
bool active_snapshot;
int i;
funcctx = SRF_FIRSTCALL_INIT();
/*
* Переключаемся в multi_call_memory_ctx - все, что нужно сохранить
* между вызовами PERCALL (tupdesc, wctx, массив texts[]), аллоцируется
* именно здесь. Этот контекст живет все время выполнения SRF.
*/
oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
tupdesc = CreateTemplateTupleDesc(2);
TupleDescInitEntry(tupdesc, (AttrNumber) 1, "frame_number", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 2, "query_text", TEXTOID, -1, 0);
funcctx->tuple_desc = BlessTupleDesc(tupdesc);
/*
* Читаем флаги один раз. Если расширение отключено, это параллельный
* воркер или стек помечен недостоверным (переполнение LIFO снимков),
* возвращаем пустой результат: при выключенном флаге хуки ничего не
* пишут и не снимают, и head не отражает реальную вложенность, а при
* недостоверном стеке в кольце может лежать освобожденный QueryDesc.
*/
active_snapshot = pg_query_stack_active && !pgs_ring_unreliable;
/*
* Снимок-один-раз: читаем pgs_ring_head ОДИН РАЗ и фиксируем в
* snapshot_head. Все последующие PERCALL-вызовы используют max_emit,
* вычисленный из этого снимка. Изменения ring_head в других хуках
* во время итерации SRF не влияют на этот вызов.
*/
snapshot_head = active_snapshot ? pgs_ring_head : 0;
max_emit = snapshot_head - skip_count;
if (max_emit < 0)
max_emit = 0;
wctx = (PgsWalkerCtx *) palloc(sizeof(PgsWalkerCtx));
wctx->max_emit = max_emit;
wctx->texts = NULL;
funcctx->user_fctx = wctx;
funcctx->max_calls = max_emit;
if (max_emit > 0)
{
wctx->texts = (char **) palloc(sizeof(char *) * max_emit);
/*
* Прямой перебор: кольцо хранит фреймы в хронологическом порядке
* вставки, индекс 0 - самый внешний (top-level) запрос,
* индекс max_emit-1 - самый вложенный из видимых.
*/
for (i = 0; i < max_emit; i++)
{
QueryDesc *qd = pgs_ring[i].qd;
EState *est;
const char *src;
Assert(i >= 0 && i < MAX_QUERY_STACK_DEPTH);
/*
* ЗАЩИТА G1: qd != NULL.
* Слот без записи (0 из BSS-инициализации). Ниже head таких
* слотов не бывает: push и pop делаются на одном C-фрейме, а
* восстановление по снимку никогда не поднимает head.
*/
if (qd == NULL)
{
wctx->texts[i] = pstrdup("<null queryDesc>");
continue;
}
/*
* ЗАЩИТА G2: qd->estate != NULL.
* Читаем estate ОДИН РАЗ в локальную переменную.
* standard_ExecutorEnd обнуляет qd->estate сразу после
* завершения исполнителя, ДО освобождения памяти QueryDesc.
* Фрейм в кольце принадлежит запросу в середине его
* ExecutorRun или ExecutorFinish, так что NULL здесь означает
* нарушение инварианта. Читать sourceText через такой qd нельзя.
*/
est = qd->estate;
if (est == NULL)
{
wctx->texts[i] = pstrdup("<estate cleaned up>");
continue;
}
/*
* ЗАЩИТА G3: qd->sourceText != NULL.
* Читаем sourceText ОДИН РАЗ в локальную переменную.
* На стандартных путях sourceText жив, пока жив estate (G2).
* Тем не менее проверяем явно: некоторые внутренние пути SPI
* (prepared-plan executions) могут передавать NULL в sourceText.
*/
src = qd->sourceText;
if (src == NULL)
{
wctx->texts[i] = pstrdup("<null sourceText>");
continue;
}
/*
* Все проверки прошли - копируем текст через pstrdup.
* Копия живет в multi_call_memory_ctx и доступна во всех
* последующих PERCALL-вызовах. Сырой указатель src больше
* нигде не сохраняется.
*/
wctx->texts[i] = pstrdup(src);
}
}
MemoryContextSwitchTo(oldcontext);
}
funcctx = SRF_PERCALL_SETUP();
if (funcctx->call_cntr < funcctx->max_calls)
{
Datum values[2];
bool nulls[2] = {false, false};
HeapTuple tuple;
PgsWalkerCtx *wctx = (PgsWalkerCtx *) funcctx->user_fctx;
int idx = (int) funcctx->call_cntr;
const char *qtext;
qtext = wctx->texts[idx];
if (qtext == NULL || qtext[0] == '\0')
qtext = "<unnamed query>";
values[0] = Int32GetDatum((int32) idx);
values[1] = CStringGetTextDatum(qtext);
tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
}
else
{
SRF_RETURN_DONE(funcctx);
}
}