Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * pgrepack.c
4 : : * Logical Replication output plugin for REPACK command
5 : : *
6 : : * Copyright (c) 2026, PostgreSQL Global Development Group
7 : : *
8 : : * IDENTIFICATION
9 : : * src/backend/replication/pgrepack/pgrepack.c
10 : : *
11 : : *-------------------------------------------------------------------------
12 : : */
13 : : #include "postgres.h"
14 : :
15 : : #include "access/detoast.h"
16 : : #include "commands/repack.h"
17 : : #include "commands/repack_internal.h"
18 : : #include "replication/snapbuild.h"
19 : : #include "utils/memutils.h"
20 : :
21 : 8 : PG_MODULE_MAGIC_EXT(
22 : : .name = "pgrepack",
23 : : .version = PG_VERSION
24 : : );
25 : :
26 : : static void repack_startup(LogicalDecodingContext *ctx,
27 : : OutputPluginOptions *opt, bool is_init);
28 : : static void repack_shutdown(LogicalDecodingContext *ctx);
29 : : static void repack_begin_txn(LogicalDecodingContext *ctx,
30 : : ReorderBufferTXN *txn);
31 : : static void repack_commit_txn(LogicalDecodingContext *ctx,
32 : : ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
33 : : static void repack_process_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
34 : : Relation relation, ReorderBufferChange *change);
35 : : static void repack_store_change(LogicalDecodingContext *ctx, Relation relation,
36 : : ConcurrentChangeKind kind, HeapTuple tuple);
37 : :
38 : : void
39 : 8 : _PG_output_plugin_init(OutputPluginCallbacks *cb)
40 : : {
41 : 8 : cb->startup_cb = repack_startup;
42 : 8 : cb->begin_cb = repack_begin_txn;
43 : 8 : cb->change_cb = repack_process_change;
44 : 8 : cb->commit_cb = repack_commit_txn;
45 : 8 : cb->shutdown_cb = repack_shutdown;
46 : 8 : }
47 : :
48 : :
49 : : /* initialize this plugin */
50 : : static void
51 : 8 : repack_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
52 : : bool is_init)
53 : : {
54 : : RepackDecodingState *dstate;
55 : :
56 [ + + ]: 8 : if (!AmRepackWorker())
57 [ + - ]: 1 : ereport(ERROR,
58 : : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
59 : : errmsg("unsupported use of logical decoding plugin \"%s\"",
60 : : "pgrepack"),
61 : : errdetail("This plugin can only be used by %s.",
62 : : "REPACK (CONCURRENTLY)"));
63 : :
64 : : /* Initial setup of our private state */
65 : : Assert(CurrentMemoryContext == ctx->context);
66 : 7 : dstate = palloc0_object(RepackDecodingState);
67 : 7 : dstate->change_cxt = AllocSetContextCreate(ctx->context,
68 : : "REPACK - change",
69 : : ALLOCSET_DEFAULT_SIZES);
70 : : /* repack_setup_logical_decoding fills in the rest */
71 : 7 : ctx->output_writer_private = dstate;
72 : :
73 : : /* Probably unnecessary, as we don't use the SQL interface ... */
74 : 7 : opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
75 : :
76 [ - + ]: 7 : if (ctx->output_plugin_options != NIL)
77 : : {
78 [ # # ]: 0 : ereport(ERROR,
79 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
80 : : errmsg("this plugin does not expect any options"));
81 : : }
82 : 7 : }
83 : :
84 : : static void
85 : 7 : repack_shutdown(LogicalDecodingContext *ctx)
86 : : {
87 : 7 : }
88 : :
89 : : /*
90 : : * As we don't release the slot during processing of particular table, there's
91 : : * no room for SQL interface, even for debugging purposes. Therefore we need
92 : : * neither OutputPluginPrepareWrite() nor OutputPluginWrite() in the plugin
93 : : * callbacks. (Although we might want to write custom callbacks, this API
94 : : * seems to be unnecessarily generic for our purposes.)
95 : : */
96 : :
97 : : /* BEGIN callback */
98 : : static void
99 : 11 : repack_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
100 : : {
101 : 11 : }
102 : :
103 : : /* COMMIT callback */
104 : : static void
105 : 11 : repack_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
106 : : XLogRecPtr commit_lsn)
107 : : {
108 : 11 : }
109 : :
110 : : /*
111 : : * Callback for individual changed tuples
112 : : */
113 : : static void
114 : 32 : repack_process_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
115 : : Relation relation, ReorderBufferChange *change)
116 : : {
117 : 32 : RepackDecodingState *private PG_USED_FOR_ASSERTS_ONLY =
118 : : (RepackDecodingState *) ctx->output_writer_private;
119 : :
120 : : /* Changes of other relation should not have been decoded. */
121 : : Assert(RelationGetRelid(relation) == private->relid);
122 : :
123 : : /* Decode entry depending on its type */
124 [ + + + - ]: 32 : switch (change->action)
125 : : {
126 : 7 : case REORDER_BUFFER_CHANGE_INSERT:
127 : : {
128 : : HeapTuple newtuple;
129 : :
130 : 7 : newtuple = change->data.tp.newtuple;
131 : :
132 : : /*
133 : : * Identity checks in the main function should have made this
134 : : * impossible.
135 : : */
136 [ - + ]: 7 : if (newtuple == NULL)
137 [ # # ]: 0 : elog(ERROR, "incomplete insert info");
138 : :
139 : 7 : repack_store_change(ctx, relation, CHANGE_INSERT, newtuple);
140 : : }
141 : 7 : break;
142 : 22 : case REORDER_BUFFER_CHANGE_UPDATE:
143 : : {
144 : : HeapTuple oldtuple,
145 : : newtuple;
146 : :
147 : 22 : oldtuple = change->data.tp.oldtuple;
148 : 22 : newtuple = change->data.tp.newtuple;
149 : :
150 [ - + ]: 22 : if (newtuple == NULL)
151 [ # # ]: 0 : elog(ERROR, "incomplete update info");
152 : :
153 [ + + ]: 22 : if (oldtuple != NULL)
154 : 8 : repack_store_change(ctx, relation, CHANGE_UPDATE_OLD, oldtuple);
155 : :
156 : 22 : repack_store_change(ctx, relation, CHANGE_UPDATE_NEW, newtuple);
157 : : }
158 : 22 : break;
159 : 3 : case REORDER_BUFFER_CHANGE_DELETE:
160 : : {
161 : : HeapTuple oldtuple;
162 : :
163 : 3 : oldtuple = change->data.tp.oldtuple;
164 : :
165 [ - + ]: 3 : if (oldtuple == NULL)
166 [ # # ]: 0 : elog(ERROR, "incomplete delete info");
167 : :
168 : 3 : repack_store_change(ctx, relation, CHANGE_DELETE, oldtuple);
169 : : }
170 : 3 : break;
171 : 0 : default:
172 : :
173 : : /*
174 : : * Should not come here. This includes TRUNCATE of the table being
175 : : * processed. heap_decode() cannot check the file locator easily,
176 : : * but we assume that TRUNCATE uses AccessExclusiveLock on the
177 : : * table so it should not occur during REPACK (CONCURRENTLY).
178 : : */
179 : : Assert(false);
180 : 0 : break;
181 : : }
182 : 32 : }
183 : :
184 : : /*
185 : : * Write the given tuple, with the given change kind, to the repack spill
186 : : * file. Later, the repack decoding worker can read these and replay
187 : : * the operations on the new copy of the table.
188 : : *
189 : : * For each change affecting the table being repacked, we store enough
190 : : * information about each tuple in it, so that it can be replayed in the
191 : : * new copy of the table.
192 : : */
193 : : static void
194 : 40 : repack_store_change(LogicalDecodingContext *ctx, Relation relation,
195 : : ConcurrentChangeKind kind, HeapTuple tuple)
196 : : {
197 : : RepackDecodingState *dstate;
198 : : MemoryContext oldcxt;
199 : : BufFile *file;
200 : 40 : List *attrs_ext = NIL;
201 : : int natt_ext;
202 : :
203 : 40 : dstate = (RepackDecodingState *) ctx->output_writer_private;
204 : 40 : file = dstate->file;
205 : :
206 : : /* Store the change kind. */
207 : 40 : BufFileWrite(file, &kind, 1);
208 : :
209 : : /* Use a frequently-reset context to avoid dealing with leaks manually */
210 : 40 : oldcxt = MemoryContextSwitchTo(dstate->change_cxt);
211 : :
212 : : /*
213 : : * If the tuple contains "external indirect" attributes, we need to write
214 : : * the contents to the file because we have no control over that memory.
215 : : */
216 [ + + ]: 40 : if (HeapTupleHasExternal(tuple))
217 : : {
218 : 13 : TupleDesc desc = RelationGetDescr(relation);
219 : : TupleTableSlot *slot;
220 : :
221 : : /* Initialize the slot, if not done already */
222 [ + + ]: 13 : if (dstate->slot == NULL)
223 : : {
224 : : ResourceOwner saveResourceOwner;
225 : :
226 : 1 : MemoryContextSwitchTo(dstate->worker_cxt);
227 : 1 : saveResourceOwner = CurrentResourceOwner;
228 : 1 : CurrentResourceOwner = dstate->worker_resowner;
229 : 1 : dstate->slot = MakeSingleTupleTableSlot(desc, &TTSOpsHeapTuple);
230 : 1 : MemoryContextSwitchTo(dstate->change_cxt);
231 : 1 : CurrentResourceOwner = saveResourceOwner;
232 : : }
233 : :
234 : 13 : slot = dstate->slot;
235 : 13 : ExecStoreHeapTuple(tuple, slot, false);
236 : :
237 : : /*
238 : : * Loop over all attributes, and find out which ones we need to spill
239 : : * separately, to wit: each one that's a non-null varlena and stored
240 : : * out of line.
241 : : */
242 [ + + ]: 78 : for (int i = 0; i < desc->natts; i++)
243 : : {
244 : 65 : CompactAttribute *attr = TupleDescCompactAttr(desc, i);
245 : : varlena *varlen;
246 : :
247 [ + + + + : 91 : if (attr->attisdropped || attr->attlen != -1 ||
- + ]
248 : 26 : slot_attisnull(slot, i + 1))
249 : 39 : continue;
250 : :
251 : 26 : slot_getsomeattrs(slot, i + 1);
252 : :
253 : : /*
254 : : * This is a non-null varlena datum, but we only care if it's
255 : : * out-of-line
256 : : */
257 : 26 : varlen = (varlena *) DatumGetPointer(slot->tts_values[i]);
258 [ + + ]: 26 : if (!VARATT_IS_EXTERNAL(varlen))
259 : 9 : continue;
260 : :
261 : : /*
262 : : * We spill any indirect-external attributes separately from the
263 : : * heap tuple. Anything else is written as is.
264 : : */
265 [ + + ]: 17 : if (VARATT_IS_EXTERNAL_INDIRECT(varlen))
266 : 15 : attrs_ext = lappend(attrs_ext, varlen);
267 : : else
268 : : {
269 : : /*
270 : : * Logical decoding should not produce "external expanded"
271 : : * attributes (those actually should never appear on disk), so
272 : : * only TOASTed attribute can be seen here.
273 : : *
274 : : * We get here if the table has external values but only
275 : : * in-line values are being updated now.
276 : : */
277 : : Assert(VARATT_IS_EXTERNAL_ONDISK(varlen));
278 : : }
279 : : }
280 : :
281 : 13 : ExecClearTuple(slot);
282 : : }
283 : :
284 : : /*
285 : : * First, write the original heap tuple, prefixed by its length. Note
286 : : * that the external-toast tag for each toasted attribute will be present
287 : : * in what we write, so that we know where to restore each one later.
288 : : */
289 : 40 : BufFileWrite(file, &tuple->t_len, sizeof(tuple->t_len));
290 : 40 : BufFileWrite(file, tuple->t_data, tuple->t_len);
291 : :
292 : : /* Then, write the number of external attributes we found. */
293 : 40 : natt_ext = list_length(attrs_ext);
294 : 40 : BufFileWrite(file, &natt_ext, sizeof(natt_ext));
295 : :
296 : : /* Finally, the attributes themselves, if any */
297 [ + + + + : 95 : foreach_ptr(varlena, attr_val, attrs_ext)
+ + ]
298 : : {
299 : 15 : attr_val = detoast_external_attr(attr_val);
300 : 15 : BufFileWrite(file, attr_val, VARSIZE_ANY(attr_val));
301 : : /* These attributes could be large, so free them right away */
302 : 15 : pfree(attr_val);
303 : : }
304 : :
305 : : /* Cleanup. */
306 : 40 : MemoryContextSwitchTo(oldcxt);
307 : 40 : MemoryContextReset(dstate->change_cxt);
308 : 40 : }
|
