Fix REPACK (CONCURRENTLY) for columns added without a table rewrite

When applying the concurrent data changes, REPACK deformed the decoded
tuples with the descriptor of the transient relation.  That descriptor
has no "missing" values, because make_new_heap() gives the transient
relation none of the defaults and constraints of the source relation.  A
tuple written before an ALTER TABLE ... ADD COLUMN that did not rewrite
the table has fewer attributes than the descriptor, so the missing
values came out as NULL, even in a column declared NOT NULL.  If such a
value belongs to the replica identity, the key used to find the target
tuple was NULL too, so the tuple to modify cannot not be found.  The
scan key does not remember that a value is NULL, so a pass-by-reference
key would even crash the comparison function.  Such tuples will be rare,
because new row versions are normally formed with the current
descriptor.  A BEFORE ROW UPDATE trigger returning OLD is one way to
produce one.

Fix by deforming the decoded tuples with the descriptor of the source
relation, which does have the missing values.  Tuples formed with it are
still valid for the transient relation, whose attributes are a copy of
the source relation ones.  Commit 20d3fe9009 took the same approach
for INSERT and UPDATE in the executor.  Expanding the decoded tuple with
heap_expand_tuple() would be the other option, but that is the
workaround of commit ba9f18abd that 20d3fe9009 got rid of, so do not
bring it back.

Add an isolation test.

Author: Sami Imseih <samimseih.pg@gmail.com>
Reported-by: Shihao Zhong <zhong950419@gmail.com>
Reviewed-by: Kirill Reshke <reshkekirill@gmail.com>
Backpatch-through: 19
Discussion: https://postgr.es/m/CAN12+Y+NJwr5EqKVrHpPD=5+b_NRAOzU9fhrVDO3H46x2WRxoA@mail.gmail.com

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

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

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 \

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;
+ <waiting ...>
+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)
+

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',

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