Added infrastructure for progress monitoring.

The pg_rewrite_progress view is the expected user interface.

Author: Antonin Houska

README.md

@@ -74,12 +74,12 @@ CREATE TABLE measurement_y2006m03 PARTITION OF measurement_aux
 ```
 
 *It's essential that both the source (`measurement`) and destination
-(`measurement_aux`) table have an identity index, the easiest way to ensure
+(`measurement_aux`) table have an identity index. The easiest way to ensure
 this is to create `PRIMARY KEY` or `UNIQUE` constraint. Also note that the key
 (i.e. column list) of the identity index of the source and destination table
 must be identical. The identity is needed to process data changes that
-applications make while data is being copied from the source to the destination
-table.*
+applications make while data is being copied from the source to the
+destination table.*
 
 Also, unless you've set `rewrite.check_constraints` to `false`, make sure that
 the destination table has all the constraints that the source table has.
@@ -98,6 +98,19 @@ lock `measurement` exclusively and rename (1) `measurement` to
 ends up to be the partitioned table, while `measurement_old` is the original,
 non-partitioned table.
 
+# Progress monitoring
+
+If `partition_table()` takes long time to finish, you might be interested in
+the progress. The `pg_rewrite_progress` view shows all the pending calls of
+the function in the current database. The `src_table`, `dst_table` and
+`src_table_new` columns contain the arguments of the `partition_table()`
+function. `ins_initial` is the number of tuples inserted into the new table
+storage during the "initial load stage", i.e. the number of tuples present in
+the table before the processing started. On the other hand, `ins`, `upd` and
+`del` are the numbers of tuples inserted, updated and deleted by applications
+during the table processing. (These "concurrent data changes" must also be
+incorporated into the partitioned table, otherwise they'd get lost.)
+
 # Limitations
 
 Please consider the following before you try to use the function:

concurrent.c

@@ -205,9 +205,6 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 {
 	TupleTableSlot *slot;
 	HeapTuple	tup_old = NULL;
-	double		ninserts,
-				nupdates,
-				ndeletes;
 
 	if (dstate->nchanges == 0)
 		return;
@@ -221,9 +218,6 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 	 */
 	PushActiveSnapshot(GetTransactionSnapshot());
 
-	ninserts = 0;
-	nupdates = 0;
-	ndeletes = 0;
 	while (tuplestore_gettupleslot(dstate->tstore, true, false,
 								   dstate->tsslot))
 	{
@@ -305,7 +299,10 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 			list_free(recheck);
 			pfree(tup);
 
-			ninserts++;
+			/* Update the progress information. */
+			SpinLockAcquire(&MyWorkerTask->mutex);
+			MyWorkerTask->progress.ins++;
+			SpinLockRelease(&MyWorkerTask->mutex);
 		}
 		else if (change->kind == CHANGE_UPDATE_NEW ||
 				 change->kind == CHANGE_DELETE)
@@ -342,7 +339,11 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 				find_tuple_in_partition(tup_old, rri_old->ri_RelationDesc,
 										partitions, key, nkeys, &ctid);
 				simple_heap_delete(rri_old->ri_RelationDesc, &ctid);
-				ndeletes++;
+
+				/* Update the progress information. */
+				SpinLockAcquire(&MyWorkerTask->mutex);
+				MyWorkerTask->progress.del++;
+				SpinLockRelease(&MyWorkerTask->mutex);
 
 				/* Insert the new tuple into its partition. */
 				ExecStoreHeapTuple(tup, slot, false);
@@ -370,7 +371,11 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 #endif
 					);
 				ExecClearTuple(slot);
-				ninserts++;
+
+				/* Update the progress information. */
+				SpinLockAcquire(&MyWorkerTask->mutex);
+				MyWorkerTask->progress.ins++;
+				SpinLockRelease(&MyWorkerTask->mutex);
 
 				list_free(recheck);
 			}
@@ -439,12 +444,19 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 						list_free(recheck);
 					}
 
-					nupdates++;
+					/* Update the progress information. */
+					SpinLockAcquire(&MyWorkerTask->mutex);
+					MyWorkerTask->progress.upd++;
+					SpinLockRelease(&MyWorkerTask->mutex);
 				}
 				else
 				{
 					simple_heap_delete(rri->ri_RelationDesc, &ctid);
-					ndeletes++;
+
+					/* Update the progress information. */
+					SpinLockAcquire(&MyWorkerTask->mutex);
+					MyWorkerTask->progress.del++;
+					SpinLockRelease(&MyWorkerTask->mutex);
 				}
 			}
 
@@ -470,10 +482,6 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 		pfree(tup_change);
 	}
 
-	elog(DEBUG1,
-		 "pg_rewrite: concurrent changes applied: %.0f inserts, %.0f updates, %.0f deletes.",
-		 ninserts, nupdates, ndeletes);
-
 	tuplestore_clear(dstate->tstore);
 	dstate->nchanges = 0;
 

pg_rewrite--1.0--1.1.sql

@@ -12,3 +12,26 @@ RETURNS void
 AS 'MODULE_PATHNAME', 'partition_table_new'
 LANGUAGE C
 STRICT;
+
+CREATE FUNCTION pg_rewrite_get_task_list()
+RETURNS TABLE (
+	tabschema_src	name,
+	tabname_src	name,
+	tabschema_dst	name,
+	tabname_dst	name,
+	tabname_src_new	name,
+	ins_initial	bigint,
+	ins		bigint,
+	upd		bigint,
+	del		bigint)
+AS 'MODULE_PATHNAME', 'pg_rewrite_get_task_list'
+LANGUAGE C;
+
+-- The column names should match the arguments of the partition_table()
+-- function.
+CREATE VIEW pg_rewrite_progress AS
+SELECT	COALESCE(tabschema_src || '.', '') || tabname_src AS src_table,
+	COALESCE(tabschema_dst || '.', '') || tabname_dst AS dst_table,
+	tabname_src_new AS src_table_new,
+	ins_initial, ins, upd, del
+FROM pg_rewrite_get_task_list();