Fix rewriting where the destination table column has a different data type.

So far only done for non-partitioned table. The current tests for partitioned
table moved to a separate test.

Author: Antonin Houska

Makefile

@@ -9,7 +9,7 @@ pg_rewrite--1.2--1.3.sql
 DOCS = pg_rewrite.md
 
 REGRESS = pg_rewrite
-#ISOLATION = pg_rewrite_concurrent
+#ISOLATION = pg_rewrite_concurrent pg_rewrite_concurrent_partition
 
 PGXS := $(shell $(PG_CONFIG) --pgxs)
 include $(PGXS)

concurrent.c

@@ -374,7 +374,7 @@ apply_insert(Relation rel, HeapTuple tup, TupleTableSlot *slot,
 	{
 		/* Non-partitioned table. */
 		rri = mtstate->resultRelInfo;
-		rel_ins = rel;
+		rel_ins = rri->ri_RelationDesc;
 		/* Use bistate passed by the caller. */
 	}
 	Assert(bistate != NULL);
@@ -446,21 +446,27 @@ apply_update_or_delete(Relation rel, HeapTuple tup, HeapTuple tup_old,
 		}
 	}
 
-	/* Which partition does the tuple belong to? */
-	ExecStoreHeapTuple(tup, slot, false);
-	rri = ExecFindPartition(mtstate, mtstate->rootResultRelInfo,
-							proute, slot, estate);
-	ExecClearTuple(slot);
-
-	/* Is this a cross-partition update? */
-	if (partitions && change_kind == CHANGE_UPDATE_NEW && tup_old)
+	/* Is the destination table partitioned? */
+	if (proute)
 	{
-		ExecStoreHeapTuple(tup_old, slot, false);
-		rri_old = ExecFindPartition(mtstate, mtstate->rootResultRelInfo,
-									proute, slot, estate);
+		/* Which partition does the tuple belong to? */
+		ExecStoreHeapTuple(tup, slot, false);
+		rri = ExecFindPartition(mtstate, mtstate->rootResultRelInfo,
+								proute, slot, estate);
 		ExecClearTuple(slot);
+
+		if (change_kind == CHANGE_UPDATE_NEW && tup_old)
+		{
+			ExecStoreHeapTuple(tup_old, slot, false);
+			rri_old = ExecFindPartition(mtstate, mtstate->rootResultRelInfo,
+										proute, slot, estate);
+			ExecClearTuple(slot);
+		}
 	}
+	else
+		rri = mtstate->resultRelInfo;
 
+	/* Is this a cross-partition update? */
 	if (rri_old &&
 		RelationGetRelid(rri_old->ri_RelationDesc) !=
 		RelationGetRelid(rri->ri_RelationDesc))

expected/pg_rewrite_concurrent.out

@@ -1,6 +1,6 @@
 Parsed test spec with 2 sessions
 
-starting permutation: do_rewrite wait_for_before_lock_ip do_changes wakeup_before_lock_ip wait_for_after_commit_ip check wakeup_after_commit_ip
+starting permutation: do_rewrite wait_for_before_lock_ip do_changes wakeup_before_lock_ip wait_for_after_commit_ip do_check wakeup_after_commit_ip
 injection_points_attach
 -----------------------
 
@@ -34,23 +34,21 @@ END;
 $$;
 
 step do_changes: 
-	-- Insert one row into each partition.
 	INSERT INTO tbl_src VALUES (2, 20), (3, 30), (5, 50);
 
 	-- Update with no identity change.
 	UPDATE tbl_src SET j=0 WHERE i=1;
 
-	-- Update with identity change but within the same partition.
-	UPDATE tbl_src SET i=6 WHERE i=5;
+	-- Update with identity change.
+	UPDATE tbl_src SET i=6 WHERE i=4;
 
-	-- Cross-partition update.
-	UPDATE tbl_src SET i=7 WHERE i=3;
-
-	-- Update a row we inserted and updated, to check that it's visible.
-	UPDATE tbl_src SET j=4 WHERE i=7;
+	-- Update a row we inserted, to check that the insertion is visible.
+	UPDATE tbl_src SET j=7 WHERE i=2;
+	-- ... and update it again, to check that the update is visible.
+	UPDATE tbl_src SET j=8 WHERE j=7;
 
 	-- Delete.
-	DELETE FROM tbl_src WHERE i=4;
+	DELETE FROM tbl_src WHERE i=7;
 
 step wakeup_before_lock_ip: 
     SELECT injection_points_wakeup('pg_rewrite-before-lock');
@@ -77,25 +75,26 @@ BEGIN
 		PERFORM pg_sleep(.1);
 	END LOOP;
 END;
-$$
+$$;
 
-step check: 
+step do_check: 
     TABLE pg_rewrite_progress;
 
     SELECT i, j FROM tbl_src ORDER BY i, j;
 
 src_table|dst_table|src_table_new|ins_initial|ins|upd|del
 ---------+---------+-------------+-----------+---+---+---
-tbl_src  |tbl_dst  |tbl_src_old  |          2|  4|  3|  2
+tbl_src  |tbl_dst  |tbl_src_old  |          3|  3|  4|  1
 (1 row)
 
 i| j
 -+--
 1| 0
-2|20
-6|50
-7| 4
-(4 rows)
+2| 8
+3|30
+5|50
+6|40
+(5 rows)
 
 step wakeup_after_commit_ip: 
     SELECT injection_points_wakeup('pg_rewrite-after-commit');

expected/pg_rewrite_concurrent_partition.out

@@ -0,0 +1,112 @@
+Parsed test spec with 2 sessions
+
+starting permutation: do_rewrite wait_for_before_lock_ip do_changes wakeup_before_lock_ip wait_for_after_commit_ip do_check wakeup_after_commit_ip
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+step do_rewrite: 
+    SELECT rewrite_table_nowait('tbl_src', 'tbl_dst', 'tbl_src_old');
+
+rewrite_table_nowait
+--------------------
+                    
+(1 row)
+
+step wait_for_before_lock_ip: 
+DO $$
+BEGIN
+        LOOP
+		PERFORM pg_stat_clear_snapshot();
+
+	        PERFORM
+		FROM pg_stat_activity
+		WHERE (wait_event_type, wait_event)=('InjectionPoint', 'pg_rewrite-before-lock');
+
+		IF FOUND THEN
+			EXIT;
+		END IF;
+
+		PERFORM pg_sleep(.1);
+	END LOOP;
+END;
+$$;
+
+step do_changes: 
+	-- Insert one row into each partition.
+	INSERT INTO tbl_src VALUES (2, 20), (3, 30), (5, 50);
+
+	-- Update with no identity change.
+	UPDATE tbl_src SET j=0 WHERE i=1;
+
+	-- Update with identity change but within the same partition.
+	UPDATE tbl_src SET i=6 WHERE i=5;
+
+	-- Cross-partition update.
+	UPDATE tbl_src SET i=7 WHERE i=3;
+
+	-- Update a row we inserted and updated, to check that it's visible.
+	UPDATE tbl_src SET j=4 WHERE i=7;
+
+	-- Delete.
+	DELETE FROM tbl_src WHERE i=4;
+
+step wakeup_before_lock_ip: 
+    SELECT injection_points_wakeup('pg_rewrite-before-lock');
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step wait_for_after_commit_ip: 
+DO $$
+BEGIN
+        LOOP
+		PERFORM pg_stat_clear_snapshot();
+
+	        PERFORM
+		FROM pg_stat_activity
+		WHERE (wait_event_type, wait_event)=('InjectionPoint', 'pg_rewrite-after-commit');
+
+		IF FOUND THEN
+			EXIT;
+		END IF;
+
+		PERFORM pg_sleep(.1);
+	END LOOP;
+END;
+$$
+
+step do_check: 
+    TABLE pg_rewrite_progress;
+
+    SELECT i, j FROM tbl_src ORDER BY i, j;
+
+src_table|dst_table|src_table_new|ins_initial|ins|upd|del
+---------+---------+-------------+-----------+---+---+---
+tbl_src  |tbl_dst  |tbl_src_old  |          2|  4|  3|  2
+(1 row)
+
+i| j
+-+--
+1| 0
+2|20
+6|50
+7| 4
+(4 rows)
+
+step wakeup_after_commit_ip: 
+    SELECT injection_points_wakeup('pg_rewrite-after-commit');
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+

pg_rewrite.c

@@ -103,7 +103,7 @@ static void run_worker(BackgroundWorker *worker, WorkerTask *task,
 					   bool nowait);
 
 static void check_prerequisites(Relation rel);
-static LogicalDecodingContext *setup_decoding(Oid relid, TupleDesc tup_desc);
+static LogicalDecodingContext *setup_decoding(Oid relid);
 static void decoding_cleanup(LogicalDecodingContext *ctx);
 static CatalogState *get_catalog_state(Oid relid);
 static void get_pg_class_info(Oid relid, TransactionId *xmin,
@@ -736,6 +736,13 @@ rewrite_worker_main(Datum main_arg)
 		MemoryContext old_context = CurrentMemoryContext;
 		ErrorData	*edata;
 
+		/*
+		 * If the backend is not waiting for our exit, make sure the error is
+		 * logged.
+		 */
+		if (MyWorkerTask->nowait)
+			PG_RE_THROW();
+
 		HOLD_INTERRUPTS();
 
 		/*
@@ -846,7 +853,6 @@ rewrite_table_impl(char *relschema_src, char *relname_src,
 	LogicalDecodingContext *ctx;
 	ReplicationSlot *slot;
 	Snapshot	snap_hist;
-	TupleDesc	tup_desc_src;
 	CatalogState *cat_state;
 	XLogRecPtr	end_of_wal;
 	XLogRecPtr	xlog_insert_ptr;
@@ -906,14 +912,6 @@ rewrite_table_impl(char *relschema_src, char *relname_src,
 
 	relid_src = RelationGetRelid(rel_src);
 
-	/*
-	 * Info to initialize the tuplestore we'll use during logical decoding.
-	 *
-	 * Copy is needed because we'll close the relation before using the tuple
-	 * descriptor.
-	 */
-	tup_desc_src = CreateTupleDescCopy(RelationGetDescr(rel_src));
-
 	/*
 	 * Get ready for the subsequent calls of
 	 * pg_rewrite_check_catalog_changes().
@@ -959,7 +957,7 @@ rewrite_table_impl(char *relschema_src, char *relname_src,
 	for (i = 0; i < nindexes; i++)
 		indexes_src[i] = cat_state->indexes[i].oid;
 
-	ctx = setup_decoding(relid_src, tup_desc_src);
+	ctx = setup_decoding(relid_src);
 
 	/*
 	 * The destination table should not be accessed by anyone during our
@@ -969,7 +967,8 @@ rewrite_table_impl(char *relschema_src, char *relname_src,
 	 * until the processing is done.
 	 *
 	 * This cannot be done before the call of setup_decoding() as the
-	 * exclusive lock does assign XID.
+	 * exclusive lock does assign XID. (setup_decoding() would then wait for
+	 * our transaction to complete.)
 	 */
 	relrv = makeRangeVar(relschema_dst, relname_dst, -1);
 	rel_dst = table_openrv(relrv, AccessExclusiveLock);
@@ -984,6 +983,12 @@ rewrite_table_impl(char *relschema_src, char *relname_src,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("\"%s\" is not a permanent table", relname_dst)));
 
+	/*
+	 * Now that we have locked rel_dst, get its tuple descriptor.
+	 */
+	((DecodingOutputState *) ctx->output_writer_private)->tupdesc =
+		RelationGetDescr(rel_dst);
+
 	/*
 	 * Build a "historic snapshot", i.e. one that reflect the table state at
 	 * the moment the snapshot builder reached SNAPBUILD_CONSISTENT state.
@@ -1527,7 +1532,7 @@ check_prerequisites(Relation rel)
  * and FATAL should lead to cleanup even before the cluster goes down.)
  */
 static LogicalDecodingContext *
-setup_decoding(Oid relid, TupleDesc tup_desc)
+setup_decoding(Oid relid)
 {
 	StringInfo	buf;
 	LogicalDecodingContext *ctx;
@@ -1598,7 +1603,6 @@ setup_decoding(Oid relid, TupleDesc tup_desc)
 	dstate->relid = relid;
 	dstate->tstore = tuplestore_begin_heap(false, false,
 										   maintenance_work_mem);
-	dstate->tupdesc = tup_desc;
 
 	/* Initialize the descriptor to store the changes ... */
 	dstate->tupdesc_change = CreateTemplateTupleDesc(1);
@@ -1626,7 +1630,6 @@ decoding_cleanup(LogicalDecodingContext *ctx)
 
 	ExecDropSingleTupleTableSlot(dstate->tsslot);
 	FreeTupleDesc(dstate->tupdesc_change);
-	FreeTupleDesc(dstate->tupdesc);
 	tuplestore_end(dstate->tstore);
 
 	FreeDecodingContext(ctx);
@@ -2104,6 +2107,7 @@ get_modify_table_state(EState *estate, Relation rel, CmdType operation)
 	ResultRelInfo *rri = makeNode(ResultRelInfo);
 
 	InitResultRelInfo(rri, rel, 0, NULL, 0);
+	ExecOpenIndices(rri, false);
 
 	result->ps.plan = NULL;
 	result->ps.state = estate;
@@ -2119,6 +2123,7 @@ get_modify_table_state(EState *estate, Relation rel, CmdType operation)
 static void
 free_modify_table_state(ModifyTableState *mtstate)
 {
+	ExecCloseIndices(mtstate->resultRelInfo);
 	pfree(mtstate->resultRelInfo);
 	pfree(mtstate);
 }

pg_rewrite.h

@@ -77,7 +77,9 @@ typedef struct DecodingOutputState
 	 */
 	TupleDesc	tupdesc_change;
 
-	/* Tuple descriptor needed to update indexes. */
+	/*
+	 * Tuple descriptor needed to apply the concurrent changes.
+	 */
 	TupleDesc	tupdesc;
 
 	/* Slot to retrieve data from tstore. */