-
Notifications
You must be signed in to change notification settings - Fork 29
/
pg_squeeze.c
3438 lines (3010 loc) · 97.4 KB
/
pg_squeeze.c
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
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*-----------------------------------------------------
*
* pg_squeeze.c
* A tool to eliminate table bloat.
*
* Copyright (c) 2016-2023, CYBERTEC PostgreSQL International GmbH
*
*-----------------------------------------------------
*/
#include "pg_squeeze.h"
#if PG_VERSION_NUM >= 130000
#include "access/heaptoast.h"
#endif
#include "access/multixact.h"
#include "access/sysattr.h"
#if PG_VERSION_NUM >= 130000
#include "access/xlogutils.h"
#endif
#if PG_VERSION_NUM >= 150000
#include "access/xloginsert.h"
#endif
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/heap.h"
#include "catalog/index.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaddress.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_am.h"
#include "catalog/pg_control.h"
#include "catalog/pg_type.h"
#include "catalog/pg_tablespace.h"
#include "catalog/toasting.h"
#include "commands/cluster.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "executor/executor.h"
#include "lib/stringinfo.h"
#include "nodes/primnodes.h"
#include "nodes/makefuncs.h"
#include "optimizer/optimizer.h"
#include "storage/bufmgr.h"
#include "storage/freespace.h"
#include "storage/lmgr.h"
#include "storage/smgr.h"
#include "storage/standbydefs.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/syscache.h"
#if PG_VERSION_NUM < 150000
extern PGDLLIMPORT int wal_segment_size;
extern PGDLLIMPORT bool FirstSnapshotSet;
#endif
#if PG_VERSION_NUM < 120000
#error "PostgreSQL version 12 or higher is required"
#endif
PG_MODULE_MAGIC;
static void squeeze_table_internal(Name relschema, Name relname, Name indname,
Name tbspname, ArrayType *ind_tbsp);
static int index_cat_info_compare(const void *arg1, const void *arg2);
/* Index-to-tablespace mapping. */
typedef struct IndexTablespace
{
Oid index;
Oid tablespace;
} IndexTablespace;
/* Where should the new table and its indexes be located? */
typedef struct TablespaceInfo
{
Oid table;
int nindexes;
IndexTablespace *indexes;
} TablespaceInfo;
/* The WAL segment being decoded. */
XLogSegNo squeeze_current_segment = 0;
static void check_prerequisites(Relation rel);
static LogicalDecodingContext *setup_decoding(Oid relid, TupleDesc tup_desc,
Snapshot *snap_hist);
static void decoding_cleanup(LogicalDecodingContext *ctx);
static CatalogState *get_catalog_state(Oid relid);
static void get_pg_class_info(Oid relid, TransactionId *xmin,
Form_pg_class *form_p, TupleDesc *desc_p);
static void get_attribute_info(Oid relid, int relnatts,
TransactionId **xmins_p,
CatalogState *cat_state);
static void cache_composite_type_info(CatalogState *cat_state, Oid typid);
static void get_composite_type_info(TypeCatInfo *tinfo);
static IndexCatInfo *get_index_info(Oid relid, int *relninds,
bool *found_invalid,
bool invalid_check_only,
bool *found_pk);
static void check_attribute_changes(CatalogState *cat_state);
static void check_index_changes(CatalogState *state);
static void check_composite_type_changes(CatalogState *cat_state);
static void free_catalog_state(CatalogState *state);
static void check_pg_class_changes(CatalogState *state);
static void free_tablespace_info(TablespaceInfo *tbsp_info);
static void resolve_index_tablepaces(TablespaceInfo *tbsp_info,
CatalogState *cat_state,
ArrayType *ind_tbsp_a);
static void perform_initial_load(Relation rel_src, RangeVar *cluster_idx_rv,
Snapshot snap_hist, Relation rel_dst,
LogicalDecodingContext *ctx);
static Oid create_transient_table(CatalogState *cat_state, TupleDesc tup_desc,
Oid tablespace, Oid relowner);
static Oid *build_transient_indexes(Relation rel_dst, Relation rel_src,
Oid *indexes_src, int nindexes,
TablespaceInfo *tbsp_info,
CatalogState *cat_state,
LogicalDecodingContext *ctx);
static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src,
int *nentries);
static bool perform_final_merge(Oid relid_src, Oid *indexes_src, int nindexes,
Relation rel_dst, ScanKey ident_key,
int ident_key_nentries,
IndexInsertState *iistate,
CatalogState *cat_state,
LogicalDecodingContext *ctx);
static void swap_relation_files(Oid r1, Oid r2);
static void swap_toast_names(Oid relid1, Oid toastrelid1, Oid relid2,
Oid toastrelid2);
static Oid get_toast_index(Oid toastrelid);
/*
* The maximum time to hold AccessExclusiveLock during the final
* processing. Note that it only process_concurrent_changes() execution time
* is included here. The very last steps like swap_relation_files() and
* swap_toast_names() shouldn't get blocked and it'd be wrong to consider them
* a reason to abort otherwise completed processing.
*/
int squeeze_max_xlock_time = 0;
/*
* List of database names for which the background worker should start started
* during cluster startup. (We require OIDs because there seems to be now good
* way to pass list of database name w/o adding restrictions on character set
* characters.)
*/
char *squeeze_worker_autostart = NULL;
/*
* Role on behalf of which automatically-started worker connects to
* database(s).
*/
char *squeeze_worker_role = NULL;
/* The number of squeeze workers per database. */
int squeeze_workers_per_database = 1;
void
_PG_init(void)
{
if (!process_shared_preload_libraries_in_progress)
ereport(ERROR,
(errmsg("pg_squeeze must be loaded via shared_preload_libraries")));
#if PG_VERSION_NUM >= 150000
squeeze_save_prev_shmem_request_hook();
shmem_request_hook = squeeze_worker_shmem_request;
#else
squeeze_worker_shmem_request();
#endif
squeeze_save_prev_shmem_startup_hook();
shmem_startup_hook = squeeze_worker_shmem_startup;
DefineCustomStringVariable(
"squeeze.worker_autostart",
"Names of databases for which background workers start automatically.",
"Comma-separated list for of databases which squeeze worker starts as soon as "
"the cluster startup has completed.",
&squeeze_worker_autostart,
NULL,
PGC_POSTMASTER,
0,
NULL, NULL, NULL);
DefineCustomStringVariable(
"squeeze.worker_role",
"Role that background workers use to connect to database.",
"If background worker was launched automatically on cluster startup, "
"it uses this role to initiate database connection(s).",
&squeeze_worker_role,
NULL,
PGC_POSTMASTER,
0,
NULL, NULL, NULL);
DefineCustomIntVariable(
"squeeze.workers_per_database",
"Maximum number of squeeze worker processes launched for each database.",
NULL,
&squeeze_workers_per_database,
1, 1, max_worker_processes,
PGC_POSTMASTER,
0,
/*
* Assume that the in-core GUC max_worker_processes should already be
* assigned and checked before the loading of the modules starts. Since
* the context of both this GUC and the max_worker_processes is
* PGC_POSTMASTER, no future check should be needed. (Some in-core GUCs
* that reference other ones have the hooks despite being PGC_POSTMASTER,
* but the reason seems to be that those cannot assume anything about the
* order of checking.)
*/
NULL, NULL, NULL);
if (squeeze_worker_autostart)
{
List *dbnames = NIL;
char *dbname,
*c;
int len;
ListCell *lc;
if (squeeze_worker_role == NULL)
ereport(ERROR,
(errcode(ERRCODE_ZERO_LENGTH_CHARACTER_STRING),
(errmsg("\"squeeze.worker_role\" parameter is invalid or not set"))));
c = squeeze_worker_autostart;
len = 0;
dbname = NULL;
while (true)
{
bool done;
done = *c == '\0';
if (done || isspace(*c))
{
if (dbname != NULL)
{
/* The current item ends here. */
Assert(len > 0);
dbnames = lappend(dbnames, pnstrdup(dbname, len));
dbname = NULL;
len = 0;
}
if (done)
break;
}
else
{
/*
* Start a new item or add the character to the current one.
*/
if (dbname == NULL)
{
dbname = c;
len = 1;
}
else
len++;
}
c++;
}
if (list_length(dbnames) == 0)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
(errmsg("\"squeeze.worker_autostart\" parameter is empty"))));
foreach(lc, dbnames)
{
WorkerConInit *con;
BackgroundWorker worker;
dbname = lfirst(lc);
con = allocate_worker_con_info(dbname, squeeze_worker_role);
squeeze_initialize_bgworker(&worker, con, NULL, 0);
RegisterBackgroundWorker(&worker);
}
list_free_deep(dbnames);
}
DefineCustomIntVariable(
"squeeze.max_xlock_time",
"The maximum time the processed table may be locked exclusively.",
"The source table is locked exclusively during the final stage of "
"processing. If the lock time should exceed this value, the lock is "
"released and the final stage is retried a few more times.",
&squeeze_max_xlock_time,
0, 0, INT_MAX,
PGC_USERSET,
GUC_UNIT_MS,
NULL, NULL, NULL);
}
#define REPLORIGIN_NAME_PATTERN "pg_squeeze_%u_%u"
/*
* The original implementation would certainly fail on PG 16 and higher, due
* to the commit 240e0dbacd (in the master branch). It's not worth supporting
* lower versions of pg_squeeze on lower versions of PG server.
*/
extern Datum squeeze_table(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1(squeeze_table);
Datum
squeeze_table(PG_FUNCTION_ARGS)
{
ereport(ERROR, (errmsg("the old implementation of the function is no longer supported"),
errhint("please run \"ALTER EXTENSION pg_squeeze UPDATE\"")));
PG_RETURN_VOID();
}
/*
* A substitute for CHECK_FOR_INTERRUPRS.
*
* procsignal_sigusr1_handler does not support signaling from a backend to a
* non-parallel worker (see the values of ProcSignalReason), and an extension
* has no other way to set the flags checked by ProcessInterrupts(), so the
* worker cannot use CHECK_FOR_INTERRUPTS. Let's use shared memory to tell the
* worker that it should exit. (SIGTERM would terminate the worker easily,
* but due to race conditions we could terminate another backend / worker
* which already managed to reuse this worker's PID.)
*/
void
exit_if_requested(void)
{
bool exit_requested;
/*
* This function only terminates workers launched by the squeeze_table()
* function. Those launched by the scheduler worker need SIGTERM.
*/
if (MyWorkerTask == NULL)
return;
SpinLockAcquire(&MyWorkerTask->mutex);
exit_requested = MyWorkerTask->exit_requested;
SpinLockRelease(&MyWorkerTask->mutex);
if (!exit_requested)
return;
/*
* Message similar to that in ProcessInterrupts(), but ERROR is
* sufficient here. squeeze_table_impl() should catch it.
*/
ereport(ERROR,
(errcode(ERRCODE_ADMIN_SHUTDOWN),
errmsg("terminating pg_squeeze background worker due to administrator command")));
}
/*
* Introduced in pg_squeeze 1.6, to be called directly as opposed to calling
* via the postgres executor.
*
* Return true if succeeded. If failed, copy useful information into *edata_p
* and return false.
*/
bool
squeeze_table_impl(Name relschema, Name relname, Name indname,
Name tbspname, ArrayType *ind_tbsp, ErrorData **edata_p,
MemoryContext edata_cxt)
{
bool result;
PG_TRY();
{
squeeze_table_internal(relschema, relname, indname, tbspname,
ind_tbsp);
result = true;
}
PG_CATCH();
{
squeeze_handle_error_db(edata_p, edata_cxt);
result = false;
}
PG_END_TRY();
return result;
}
static void
squeeze_table_internal(Name relschema, Name relname, Name indname,
Name tbspname, ArrayType *ind_tbsp)
{
RangeVar *relrv_src;
RangeVar *relrv_cl_idx = NULL;
Relation rel_src,
rel_dst;
Oid rel_src_owner;
Oid ident_idx_src,
ident_idx_dst;
Oid relid_src,
relid_dst;
Oid toastrelid_src,
toastrelid_dst;
char replident;
ScanKey ident_key;
int i,
ident_key_nentries;
IndexInsertState *iistate;
LogicalDecodingContext *ctx;
ReplicationSlot *slot;
Snapshot snap_hist;
TupleDesc tup_desc;
CatalogState *cat_state;
XLogRecPtr end_of_wal;
XLogRecPtr xlog_insert_ptr;
int nindexes;
Oid *indexes_src = NULL,
*indexes_dst = NULL;
bool invalid_index = false;
IndexCatInfo *ind_info;
TablespaceInfo *tbsp_info;
ObjectAddress object;
bool source_finalized;
bool xmin_valid;
relrv_src = makeRangeVar(NameStr(*relschema), NameStr(*relname), -1);
rel_src = table_openrv(relrv_src, AccessShareLock);
check_prerequisites(rel_src);
/*
* Retrieve the useful info while holding lock on the relation.
*/
ident_idx_src = RelationGetReplicaIndex(rel_src);
replident = rel_src->rd_rel->relreplident;
/* The table can have PK although the replica identity is FULL. */
if (ident_idx_src == InvalidOid && rel_src->rd_pkindex != InvalidOid)
ident_idx_src = rel_src->rd_pkindex;
relid_src = RelationGetRelid(rel_src);
rel_src_owner = RelationGetForm(rel_src)->relowner;
toastrelid_src = rel_src->rd_rel->reltoastrelid;
/*
* Info to create transient table and to initialize tuplestore we'll use
* during logical decoding.
*/
tup_desc = CreateTupleDescCopy(RelationGetDescr(rel_src));
/*
* Get ready for the subsequent calls of check_catalog_changes().
*
* Not all index changes do conflict with the AccessShareLock - see
* get_index_info() for explanation.
*
* XXX It'd still be correct to start the check a bit later, i.e. just
* before CreateInitDecodingContext(), but the gain is not worth making
* the code less readable.
*/
cat_state = get_catalog_state(relid_src);
/* Give up if it's clear enough to do so. */
if (cat_state->invalid_index)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
(errmsg("At least one index is invalid"))));
/*
* The relation shouldn't be locked during the call of setup_decoding(),
* otherwise another transaction could write XLOG records before the
* slots' data.restart_lsn and we'd have to wait for it to finish. If such
* a transaction requested exclusive lock on our relation (e.g. ALTER
* TABLE), it'd result in a deadlock.
*
* We can't keep the lock till the end of transaction anyway - that's why
* check_catalog_changes() exists.
*
* XXX Now that the squeeze worker launched by the scheduler worker no
* longer needs to call DecodingContextFindStartpoint(), it should not see
* running transactions that started before the restart_lsn, so it's
* probably no longer necessary to close the relation here. (The worker
* launched by the squeeze_table() function does call
* DecodingContextFindStartpoint(), however it does so before the current
* transaction is started.) Reconsider.
*/
table_close(rel_src, AccessShareLock);
/*
* Check if we're ready to capture changes that possibly take place during
* the initial load.
*
* Concurrent DDL causes ERROR in any case, so don't worry about validity
* of this test during the next steps.
*
* Note: we let the plugin do this check on per-change basis, and allow
* processing of tables with no identity if only INSERT changes are
* decoded. However it seems inconsistent.
*
* XXX Although ERRCODE_UNIQUE_VIOLATION is no actual "unique violation",
* this error code seems to be the best match.
* (ERRCODE_TRIGGERED_ACTION_EXCEPTION might be worth consideration as
* well.)
*/
if (replident == REPLICA_IDENTITY_NOTHING ||
(replident == REPLICA_IDENTITY_DEFAULT && !OidIsValid(ident_idx_src)))
ereport(ERROR,
(errcode(ERRCODE_UNIQUE_VIOLATION),
(errmsg("Table \"%s\".\"%s\" has no identity index",
NameStr(*relschema), NameStr(*relname)))));
/*
* Change processing w/o PK index is not a good idea.
*
* Note that some users need the "full identity" although the table does
* have PK. ("full identity" + UNIQUE constraint is also a valid setup,
* but it's harder to check).
*/
if (replident == REPLICA_IDENTITY_FULL && !cat_state->have_pk_index)
ereport(ERROR,
(errcode(ERRCODE_UNIQUE_VIOLATION),
(errmsg("Replica identity \"full\" not supported"))));
/*
* Clustering index, if any.
*
* Do not lock the index so far, e.g. just to retrieve OID and to keep it
* valid. Neither the relation can be locked continuously, so by keeping
* the index locked alone we'd introduce incorrect order of locking.
* Although we use only share locks in most cases (so I'm not aware of
* particular deadlock scenario), it doesn't seem wise. The worst
* consequence of not locking is that perform_initial_load() will error
* out.
*/
if (indname)
{
ereport(DEBUG1,
(errmsg("clustering index: %s", NameStr(*indname))));
relrv_cl_idx = makeRangeVar(NameStr(*relschema),
NameStr(*indname), -1);
}
/*
* Process tablespace arguments, if provided.
*
* XXX Currently we consider tablespace DDLs rather infrequent, so we let
* such a DDL to break transient table or index creation. As we can't
* keep the source table locked all the time, it's possible for tablespace
* to disappear even if it contains the source table. Is it worth locking
* the tablespaces here? Since concurrent renaming of a tablespace is
* disruptive too, we'd probably need AccessExclusiveLock. Or are such
* changes worth making check_catalog_changes() more expensive?
*/
tbsp_info = (TablespaceInfo *) palloc0(sizeof(TablespaceInfo));
if (tbspname)
tbsp_info->table = get_tablespace_oid(pstrdup(NameStr(*tbspname)),
false);
else
tbsp_info->table = cat_state->form_class->reltablespace;
/* Index-to-tablespace mappings. */
if (ind_tbsp)
resolve_index_tablepaces(tbsp_info, cat_state, ind_tbsp);
nindexes = cat_state->relninds;
/*
* Existence of identity index was checked above, so number of indexes and
* attributes are both non-zero.
*/
Assert(cat_state->form_class->relnatts >= 1);
Assert(nindexes > 0);
/* Copy the OIDs into a separate array, for convenient use later. */
indexes_src = (Oid *) palloc(nindexes * sizeof(Oid));
for (i = 0; i < nindexes; i++)
indexes_src[i] = cat_state->indexes[i].oid;
ctx = setup_decoding(relid_src, tup_desc, &snap_hist);
relid_dst = create_transient_table(cat_state, tup_desc, tbsp_info->table,
rel_src_owner);
/* The source relation will be needed for the initial load. */
rel_src = table_open(relid_src, AccessShareLock);
/*
* The new relation should not be visible for other transactions until we
* commit, but exclusive lock just makes sense.
*/
rel_dst = table_open(relid_dst, AccessExclusiveLock);
toastrelid_dst = rel_dst->rd_rel->reltoastrelid;
/*
* We need to know whether that no DDL took place that allows for data
* inconsistency. The relation was unlocked for some time since last
* check, so pass NoLock.
*/
check_catalog_changes(cat_state, NoLock);
/*
* This is to satisfy the check introduced by the commit 2776922201f in PG
* core. (Per HeapTupleSatisfiesToast() the snapshot shouldn't actually be
* used for visibility checks of the TOAST values.)
*/
PushActiveSnapshot(snap_hist);
/*
* The historic snapshot is used to retrieve data w/o concurrent changes.
*/
perform_initial_load(rel_src, relrv_cl_idx, snap_hist, rel_dst, ctx);
/*
* We no longer need to preserve the rows processed during the initial
* load from VACUUM. (User should not run VACUUM on a table that we
* currently process, but our stale effective_xmin would also restrict
* VACUUM on other tables.)
*/
slot = ctx->slot;
SpinLockAcquire(&slot->mutex);
xmin_valid = TransactionIdIsValid(slot->effective_xmin);
slot->effective_xmin = InvalidTransactionId;
SpinLockRelease(&slot->mutex);
/*
* This should not happen, but it's critical, therefore use ereport()
* rather than Assert(). If the value got lost somehow due to releasing
* and acquiring the slot, VACUUM could have removed some rows from the
* source table that the historic snapshot was still supposed to see.
*/
if (!xmin_valid)
ereport(ERROR,
(errmsg("effective_xmin of the replication slot \"%s\" is invalid",
NameStr(slot->data.name))));
/*
* The historic snapshot won't be needed anymore.
*/
PopActiveSnapshot();
/*
* This is rather paranoia than anything else --- perform_initial_load()
* uses each snapshot to access different table, and it does not cause
* catalog changes.
*/
InvalidateSystemCaches();
/*
* Check for concurrent changes that would make us stop working later.
* Index build can take quite some effort and we don't want to waste it.
*
* Note: By still holding the share lock we only ensure that the source
* relation is not altered underneath index build, but we'll have to
* release the lock for a short time at some point. So while we can't
* prevent anyone from forcing us to cancel our work, such cancellation
* must happen at well-defined moment.
*/
check_catalog_changes(cat_state, AccessShareLock);
/*
* Make sure the contents of the transient table is visible for the
* scan(s) during index build.
*/
CommandCounterIncrement();
/*
* Create indexes on the temporary table - that might take a while.
* (Unlike the concurrent changes, which we insert into existing indexes.)
*/
PushActiveSnapshot(GetTransactionSnapshot());
indexes_dst = build_transient_indexes(rel_dst, rel_src, indexes_src,
nindexes, tbsp_info, cat_state,
ctx);
PopActiveSnapshot();
/*
* Make the identity index of the transient table visible, for the sake of
* concurrent UPDATEs and DELETEs.
*/
CommandCounterIncrement();
/* Tablespace info is no longer needed. */
free_tablespace_info(tbsp_info);
/*
* Build scan key that we'll use to look for rows to be updated / deleted
* during logical decoding.
*/
ident_key = build_identity_key(ident_idx_src, rel_src,
&ident_key_nentries);
/*
* As we'll need to take exclusive lock later, release the shared one.
*
* Note: PG core code shouldn't actually participate in such a deadlock,
* as it (supposedly) does not raise lock level. Nor should concurrent
* call of the squeeze_table() function participate in the deadlock,
* because it should have failed much earlier when creating an existing
* logical replication slot again. Nevertheless, these circumstances still
* don't justify generally bad practice.
*
* (As we haven't changed the catalog entry yet, there's no need to send
* invalidation messages.)
*/
table_close(rel_src, AccessShareLock);
/*
* Valid identity index should exist now, see the identity checks above.
*/
Assert(OidIsValid(ident_idx_src));
/* Find "identity index" of the transient relation. */
ident_idx_dst = InvalidOid;
for (i = 0; i < nindexes; i++)
{
if (ident_idx_src == indexes_src[i])
{
ident_idx_dst = indexes_dst[i];
break;
}
}
if (!OidIsValid(ident_idx_dst))
/*
* Should not happen, concurrent DDLs should have been noticed short
* ago.
*/
elog(ERROR, "Identity index missing on the transient relation");
/* Executor state to update indexes. */
iistate = get_index_insert_state(rel_dst, ident_idx_dst);
/*
* Flush all WAL records inserted so far (possibly except for the last
* incomplete page, see GetInsertRecPtr), to minimize the amount of data
* we need to flush while holding exclusive lock on the source table.
*/
xlog_insert_ptr = GetInsertRecPtr();
XLogFlush(xlog_insert_ptr);
/*
* Since we'll do some more changes, all the WAL records flushed so far
* need to be decoded for sure.
*/
#if PG_VERSION_NUM >= 150000
end_of_wal = GetFlushRecPtr(NULL);
#else
end_of_wal = GetFlushRecPtr();
#endif
/*
* Decode and apply the data changes that occurred while the initial load
* was in progress. The XLOG reader should continue where setup_decoding()
* has left it.
*
* Even if the amount of concurrent changes of our source table might not
* be significant, both initial load and index build could have produced
* many XLOG records that we need to read. Do so before requesting
* exclusive lock on the source relation.
*/
process_concurrent_changes(ctx, end_of_wal, cat_state, rel_dst,
ident_key, ident_key_nentries, iistate,
NoLock, NULL);
/*
* This (supposedly cheap) special check should avoid one particular
* deadlock scenario: another transaction, performing index DDL
* concurrenly (e.g. DROP INDEX CONCURRENTLY) committed change of
* indisvalid, indisready, ... and called WaitForLockers() before we
* unlocked both source table and its indexes above. WaitForLockers()
* waits till the end of the holding (our) transaction as opposed to the
* end of our locks, and the other transaction holds (non-exclusive) lock
* on both relation and index. In this situation we'd cause deadlock by
* requesting exclusive lock. We should recognize this scenario by
* checking pg_index alone.
*/
ind_info = get_index_info(relid_src, NULL, &invalid_index, true, NULL);
if (invalid_index)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_IN_USE),
errmsg("Concurrent change of index detected")));
else
pfree(ind_info);
/*
* Try a few times to perform the stage that requires exclusive lock on
* the source relation.
*
* XXX Not sure the number of attempts should be configurable. If it fails
* several times, admin should either increase squeeze_max_xlock_time or
* disable it.
*/
source_finalized = false;
for (i = 0; i < 4; i++)
{
if (perform_final_merge(relid_src, indexes_src, nindexes,
rel_dst, ident_key, ident_key_nentries,
iistate, cat_state, ctx))
{
source_finalized = true;
break;
}
else
elog(DEBUG1,
"pg_squeeze: exclusive lock on table %u had to be released.",
relid_src);
}
if (!source_finalized)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_IN_USE),
errmsg("\"squeeze_max_xlock_time\" prevented squeeze from completion")));
/*
* Done with decoding.
*
* XXX decoding_cleanup() frees tup_desc, although we've used it not only
* for the decoding.
*/
decoding_cleanup(ctx);
ReplicationSlotRelease();
pfree(ident_key);
free_index_insert_state(iistate);
/* The destination table is no longer necessary, so close it. */
/*
* XXX (Should have been closed right after process_concurrent_changes()?)
*/
table_close(rel_dst, AccessExclusiveLock);
/*
* Exchange storage (including TOAST) and indexes between the source and
* destination tables.
*/
swap_relation_files(relid_src, relid_dst);
CommandCounterIncrement();
/*
* As swap_relation_files() already changed pg_class(reltoastrelid), we
* pass toastrelid_dst for relid_src and vice versa.
*/
swap_toast_names(relid_src, toastrelid_dst, relid_dst, toastrelid_src);
for (i = 0; i < nindexes; i++)
swap_relation_files(indexes_src[i], indexes_dst[i]);
CommandCounterIncrement();
if (nindexes > 0)
{
pfree(indexes_src);
pfree(indexes_dst);
}
/* State not needed anymore. */
free_catalog_state(cat_state);
/*
* Drop the transient table including indexes (and possibly constraints on
* those indexes).
*/
object.classId = RelationRelationId;
object.objectSubId = 0;
object.objectId = relid_dst;
performDeletion(&object, DROP_RESTRICT, PERFORM_DELETION_INTERNAL);
}
static int
index_cat_info_compare(const void *arg1, const void *arg2)
{
IndexCatInfo *i1 = (IndexCatInfo *) arg1;
IndexCatInfo *i2 = (IndexCatInfo *) arg2;
if (i1->oid > i2->oid)
return 1;
else if (i1->oid < i2->oid)
return -1;
else
return 0;
}
/*
* Raise error if the relation is not eligible for squeezing or any adverse
* conditions exist.
*
* Some of the checks may be redundant (e.g. heap_open() checks relkind) but
* its safer to have them all listed here.
*/
static void
check_prerequisites(Relation rel)
{
Form_pg_class form = RelationGetForm(rel);
/* Check the relation first. */
if (form->relkind == RELKIND_PARTITIONED_TABLE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot squeeze partitioned table")));
if (form->relkind != RELKIND_RELATION)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is not a table",
RelationGetRelationName(rel))));
if (form->relpersistence != RELPERSISTENCE_PERMANENT)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is not a regular table",
RelationGetRelationName(rel))));
if (form->relisshared)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is shared relation",
RelationGetRelationName(rel))));
if (RelationIsMapped(rel))
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is mapped relation",
RelationGetRelationName(rel))));
/*
* There's no urgent need to process catalog tables.
*
* Should this limitation be relaxed someday, consider if we need to write
* xl_heap_rewrite_mapping records. (Probably not because the whole
* "decoding session" takes place within a call of squeeze_table() and our
* catalog checks should not allow for a concurrent rewrite that could
* make snapmgr.c:tuplecid_data obsolete. Furthermore, such a rewrite
* would have to take place before perform_initial_load(), but this is
* called before any transactions could have been decoded, so tuplecid
* should still be empty anyway.)
*/
if (RelationGetRelid(rel) < FirstNormalObjectId)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is not user relation",
RelationGetRelationName(rel))));
/*
* While AFTER trigger should not be an issue (to generate an event must
* have got XID assigned, causing setup_decoding() to fail later), open
* cursor might be. See comments of the function for details.
*/
CheckTableNotInUse(rel, "squeeze_table()");
}
/*
* Acquire logical replication slot which created either by the scheduler
* worker or by a backend executing the squeeze_table() function.
*/
static LogicalDecodingContext *
setup_decoding(Oid relid, TupleDesc tup_desc, Snapshot *snap_hist)
{
ReplSlotStatus *repl_slot = &MyWorkerTask->repl_slot;
DecodingOutputState *dstate;
MemoryContext oldcontext;
LogicalDecodingContext *ctx;
XLogRecPtr restart_lsn;
dsm_segment *seg = NULL;
char *snap_src;
/*
* Use the slot initialized by the scheduler worker (or by the backend
* running the squeeze_table() function ).
*/
ReplicationSlotAcquire(NameStr(repl_slot->name), true);
/*
* It's pretty unlikely for some client to have consumed data changes
* (accidentally?) before this worker could acquire the slot, but it's
* easy enough to check.
*/
if (MyReplicationSlot->data.confirmed_flush != repl_slot->confirmed_flush)
ereport(ERROR,
(errmsg("replication slot \"%s\" has incorrect confirm position",
NameStr(repl_slot->name))));
/*
* Wasn't effective_xmin lost due to releasing and re-acquiring the slot?
* (ReplicationSlotRelease() does clear it in some cases. We try to avoid
* that, but checking makes sense as this slot field is critical.).
*/
if (!TransactionIdIsValid(MyReplicationSlot->effective_xmin))
ereport(ERROR,
(errmsg("replication slot \"%s\" has invalid effective_xmin",
NameStr(MyReplicationSlot->data.name))));