diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 71851481a49..5369d417bb8 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -113,6 +113,12 @@ typedef struct ChangeContext ResultRelInfo *cc_rri; EState *cc_estate; + /* + * The tuple descriptor to deform decoded tuples with; it must have the + * attmissingval values from the relation being repacked. + */ + TupleDesc cc_tupdesc; + /* * Existing tuples to UPDATE and DELETE are located via this index. We * keep the scankey in partially initialized state to avoid repeated work. @@ -198,7 +204,8 @@ static void process_concurrent_changes(XLogRecPtr end_of_wal, ChangeContext *chgcxt, bool done); static void initialize_change_context(ChangeContext *chgcxt, - Relation relation, + Relation src_relation, + Relation tgt_relation, Oid ident_index_id); static void release_change_context(ChangeContext *chgcxt); static void rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, @@ -2688,12 +2695,26 @@ apply_concurrent_changes(BufFile *file, ChangeContext *chgcxt) bool have_old_tuple = false; MemoryContext oldcxt; - spilled_tuple = MakeSingleTupleTableSlot(RelationGetDescr(rel), - &TTSOpsVirtual); + /* + * Set up the tuple table slots for the operations. + * + * spilled_tuple is a tuple we read from the file spilled by the decoding + * worker. It must be read using the tuple descriptor of the original + * relation, because it may contain attributes with 'attmissingval'. + * + * old_update_tuple is an update's OLD tuple to extract the tuple's key + * from, read from the spill file, so we also use the original rel's + * tupdesc. (XXX Many places aren't prepared for that tupdesc's attribute + * layout to differ from the transient rel's tupdesc.) + * + * ondisk_tuple is the tuple in the transient relation for UPDATEs and + * DELETEs, as obtained by searching by replication identity. It uses + * tupdesc and tuptable ops appropriate for the transient relation. + */ + spilled_tuple = MakeSingleTupleTableSlot(chgcxt->cc_tupdesc, &TTSOpsVirtual); + old_update_tuple = MakeSingleTupleTableSlot(chgcxt->cc_tupdesc, &TTSOpsVirtual); ondisk_tuple = MakeSingleTupleTableSlot(RelationGetDescr(rel), table_slot_callbacks(rel)); - old_update_tuple = MakeSingleTupleTableSlot(RelationGetDescr(rel), - &TTSOpsVirtual); oldcxt = MemoryContextSwitchTo(GetPerTupleMemoryContext(chgcxt->cc_estate)); @@ -3151,14 +3172,24 @@ process_concurrent_changes(XLogRecPtr end_of_wal, ChangeContext *chgcxt, bool do } /* - * Initialize the ChangeContext struct for the given relation, with - * the given index as identity index. + * Initialize ChangeContext to propagate changes of src_relation (the relation + * being repacked) into tgt_relation (the transient relation), using the given + * index (on tgt_relation) as identity. */ static void initialize_change_context(ChangeContext *chgcxt, - Relation relation, Oid ident_index_id) + Relation src_relation, Relation tgt_relation, + Oid ident_index_id) { - chgcxt->cc_rel = relation; + chgcxt->cc_rel = tgt_relation; + + /* + * Use the descriptor of the source relation as the one to deform the + * decoded tuples with; in particular, this descriptor contains all the + * missing attributes. Tuples formed with it are also valid for the + * transient relation, as the attributes are otherwise identical. + */ + chgcxt->cc_tupdesc = RelationGetDescr(src_relation); /* Only initialize fields needed by ExecInsertIndexTuples(). */ chgcxt->cc_estate = CreateExecutorState(); @@ -3169,7 +3200,7 @@ initialize_change_context(ChangeContext *chgcxt, */ { RangeTblEntry *rte; - TupleDesc desc = RelationGetDescr(relation); + TupleDesc desc = RelationGetDescr(tgt_relation); List *perminfos = NIL; Bitmapset *updatedCols = NULL; RTEPermissionInfo *perminfo; @@ -3181,8 +3212,8 @@ initialize_change_context(ChangeContext *chgcxt, */ rte = makeNode(RangeTblEntry); rte->rtekind = RTE_RELATION; - rte->relid = RelationGetRelid(relation); - rte->relkind = RelationGetForm(relation)->relkind; + rte->relid = RelationGetRelid(tgt_relation); + rte->relkind = RelationGetForm(tgt_relation)->relkind; /* Create the RTEPermissionInfo instance (and set ->perminfoindex). */ addRTEPermissionInfo(&perminfos, rte); @@ -3216,7 +3247,7 @@ initialize_change_context(ChangeContext *chgcxt, /* Set up our ResultRelInfo to use for index updates */ chgcxt->cc_rri = makeNode(ResultRelInfo); - InitResultRelInfo(chgcxt->cc_rri, relation, 1, NULL, 0); + InitResultRelInfo(chgcxt->cc_rri, tgt_relation, 1, NULL, 0); ExecOpenIndices(chgcxt->cc_rri, false); /* @@ -3377,7 +3408,7 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, get_rel_name(identIdx)); /* Gather information to apply concurrent changes. */ - initialize_change_context(&chgcxt, NewHeap, ident_idx_new); + initialize_change_context(&chgcxt, OldHeap, NewHeap, ident_idx_new); /* * During testing, wait for another backend to perform concurrent data diff --git a/src/backend/replication/pgrepack/pgrepack.c b/src/backend/replication/pgrepack/pgrepack.c index a1cbc8db68a..6006120548a 100644 --- a/src/backend/replication/pgrepack/pgrepack.c +++ b/src/backend/replication/pgrepack/pgrepack.c @@ -181,7 +181,7 @@ repack_process_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, /* * Write the given tuple, with the given change kind, to the repack spill - * file. Later, the repack decoding worker can read these and replay + * file. Later, the repack steering process can read these and replay * the operations on the new copy of the table. * * For each change affecting the table being repacked, we store enough diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 1c680abf7dd..136f0f77951 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -20,6 +20,7 @@ ISOLATION = basic \ repack \ repack_commit_race \ repack_decode \ + repack_missingval \ repack_temporal \ repack_temporal_multirange \ repack_toast \ diff --git a/src/test/modules/injection_points/expected/repack_missingval.out b/src/test/modules/injection_points/expected/repack_missingval.out new file mode 100644 index 00000000000..ee2bc55ef97 --- /dev/null +++ b/src/test/modules/injection_points/expected/repack_missingval.out @@ -0,0 +1,38 @@ +Parsed test spec with 2 sessions + +starting permutation: s1_wait_before_lock s2_update s2_wakeup_before_lock s1_check +injection_points_attach +----------------------- + +(1 row) + +step s1_wait_before_lock: + REPACK (CONCURRENTLY) repack_missingval; + +step s2_update: + UPDATE repack_missingval SET j = j WHERE i IN (1, 2); + +step s2_wakeup_before_lock: + SELECT injection_points_wakeup('repack-concurrently-before-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step s1_wait_before_lock: <... completed> +step s1_check: + SELECT i, j, c FROM repack_missingval ORDER BY i; + +i|j|c +-+-+--- +1|1|xyz +2|2|xyz +3|3|xyz +(3 rows) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 45117b6ffec..db6b93a8115 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -49,6 +49,7 @@ tests += { 'repack', 'repack_commit_race', 'repack_decode', + 'repack_missingval', 'repack_temporal', 'repack_temporal_multirange', 'repack_toast', diff --git a/src/test/modules/injection_points/specs/repack_missingval.spec b/src/test/modules/injection_points/specs/repack_missingval.spec new file mode 100644 index 00000000000..4b70d810afd --- /dev/null +++ b/src/test/modules/injection_points/specs/repack_missingval.spec @@ -0,0 +1,74 @@ +# Test interaction of REPACK (CONCURRENTLY) with an ALTER TABLE .. ADD COLUMN +# that doesn't rewrite the table (pg_attribute.attmissingval). +setup +{ + CREATE EXTENSION IF NOT EXISTS injection_points; + + CREATE TABLE repack_missingval(i int PRIMARY KEY, j int); + INSERT INTO repack_missingval(i, j) VALUES (1, 1), (2, 2), (3, 3); + + -- Add a column with default values and don't rewrite the table; use it + -- in the replica identity also. + ALTER TABLE repack_missingval ADD COLUMN c text NOT NULL DEFAULT 'xyz'; + CREATE UNIQUE INDEX repack_missingval_idx ON repack_missingval (i, c); + ALTER TABLE repack_missingval REPLICA IDENTITY USING INDEX repack_missingval_idx; + + -- Set up a trigger on UPDATE that returns the original tuple. This way, + -- the original short tuple is given to logical decoding. + CREATE FUNCTION repack_return_old() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + RETURN OLD; + END; + $$; + CREATE TRIGGER return_old BEFORE UPDATE ON repack_missingval + FOR EACH ROW EXECUTE FUNCTION repack_return_old(); +} + +teardown +{ + DROP TABLE repack_missingval; + DROP FUNCTION repack_return_old(); + DROP EXTENSION injection_points; +} + +session s1 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-before-lock', 'wait'); +} + +# Perform the initial load and wait for s2 to change the data. +step s1_wait_before_lock +{ + REPACK (CONCURRENTLY) repack_missingval; +} + +# The missing values must have survived the concurrent changes. +step s1_check +{ + SELECT i, j, c FROM repack_missingval ORDER BY i; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-before-lock'); +} + +session s2 + +# Update two of the three rows. +step s2_update +{ + UPDATE repack_missingval SET j = j WHERE i IN (1, 2); +} +step s2_wakeup_before_lock +{ + SELECT injection_points_wakeup('repack-concurrently-before-lock'); +} + +permutation + s1_wait_before_lock + s2_update + s2_wakeup_before_lock + s1_check