From 2e93bd37ca3c8add1f8e3e44a9f3906c332b83f2 Mon Sep 17 00:00:00 2001
From: Ilya Gladyshev <ilya.v.gladyshev@gmail.com>
Date: Fri, 9 Dec 2022 23:17:29 +0400
Subject: [PATCH] report top parent progress for CREATE INDEX

! add asserts, avoid global var
! do not count intermediate/nested/sub-partitions
---
 src/backend/commands/indexcmds.c              | 30 ++++++-
 src/backend/utils/activity/backend_progress.c | 79 +++++++++++++++++++
 src/include/nodes/parsenodes.h                |  2 +
 3 files changed, 107 insertions(+), 4 deletions(-)

diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index b5b860c3abf..6e6bba9d3a9 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -1218,8 +1218,28 @@ DefineIndex(Oid relationId,
 			Relation	parentIndex;
 			TupleDesc	parentDesc;
 
-			pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_TOTAL,
-										 nparts);
+			if (!OidIsValid(parentIndexId))
+			{
+				int			nleaves = 0;
+				List		*childs;
+				ListCell	*lc;
+
+				/*
+				 * Count the number of physical children, excluding foreign
+				 * tables, intermediate partitioned tables, as well as the
+				 * partitioned index itself.
+				 */
+				childs = find_all_inheritors(relationId, NoLock, NULL);
+				foreach(lc, childs)
+				{
+					Oid		partrelid = lfirst_oid(lc);
+					if (RELKIND_HAS_STORAGE(get_rel_relkind(partrelid)))
+						nleaves++;
+				}
+
+				pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_TOTAL,
+											 nleaves);
+			}
 
 			/* Make a local copy of partdesc->oids[], just for safety */
 			memcpy(part_oids, partdesc->oids, sizeof(Oid) * nparts);
@@ -1431,8 +1451,10 @@ DefineIndex(Oid relationId,
 										   child_save_sec_context);
 				}
 
-				pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_DONE,
-											 i + 1);
+				if (RELKIND_HAS_STORAGE(get_rel_relkind(childRelid)))
+					pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_DONE,
+												 ++stmt->nparts_done);
+
 				free_attrmap(attmap);
 			}
 
diff --git a/src/backend/utils/activity/backend_progress.c b/src/backend/utils/activity/backend_progress.c
index f29199725b7..c661ad94782 100644
--- a/src/backend/utils/activity/backend_progress.c
+++ b/src/backend/utils/activity/backend_progress.c
@@ -10,6 +10,7 @@
  */
 #include "postgres.h"
 
+#include "commands/progress.h"
 #include "port/atomics.h"		/* for memory barriers */
 #include "utils/backend_progress.h"
 #include "utils/backend_status.h"
@@ -37,6 +38,82 @@ pgstat_progress_start_command(ProgressCommandType cmdtype, Oid relid)
 	PGSTAT_END_WRITE_ACTIVITY(beentry);
 }
 
+static void
+pgstat_progress_asserts(void)
+{
+	volatile PgBackendStatus *beentry = MyBEEntry;
+	volatile int64		*a = beentry->st_progress_param;
+
+	/*
+	 * If the command fails due to interrupt or error, the values may be
+	 * less than rather than equal to expected, final value.
+	 */
+
+	switch (beentry->st_progress_command)
+	{
+	case PROGRESS_COMMAND_VACUUM:
+		Assert(a[PROGRESS_VACUUM_HEAP_BLKS_SCANNED] <=
+				a[PROGRESS_VACUUM_TOTAL_HEAP_BLKS]);
+		Assert(a[PROGRESS_VACUUM_HEAP_BLKS_VACUUMED] <=
+				a[PROGRESS_VACUUM_TOTAL_HEAP_BLKS]);
+		Assert(a[PROGRESS_VACUUM_NUM_DEAD_TUPLES] <=
+				a[PROGRESS_VACUUM_MAX_DEAD_TUPLES]);
+		break;
+
+	case PROGRESS_COMMAND_ANALYZE:
+		Assert(a[PROGRESS_ANALYZE_BLOCKS_DONE] <=
+				a[PROGRESS_ANALYZE_BLOCKS_TOTAL]);
+		/* Extended stats can be skipped */
+		Assert(a[PROGRESS_ANALYZE_EXT_STATS_COMPUTED] <=
+				a[PROGRESS_ANALYZE_EXT_STATS_TOTAL]);
+		Assert(a[PROGRESS_ANALYZE_CHILD_TABLES_DONE] <=
+				a[PROGRESS_ANALYZE_CHILD_TABLES_TOTAL]);
+		break;
+
+	case PROGRESS_COMMAND_CLUSTER:
+		Assert(a[PROGRESS_CLUSTER_HEAP_BLKS_SCANNED] <=
+				a[PROGRESS_CLUSTER_TOTAL_HEAP_BLKS]);
+		break;
+
+	case PROGRESS_COMMAND_CREATE_INDEX:
+		Assert(a[PROGRESS_CREATEIDX_TUPLES_DONE] <=
+				a[PROGRESS_CREATEIDX_TUPLES_TOTAL]);
+		/* The command can fail before finishing indexes */
+		Assert(a[PROGRESS_CREATEIDX_PARTITIONS_DONE] <=
+				a[PROGRESS_CREATEIDX_PARTITIONS_TOTAL]);
+		break;
+
+	case PROGRESS_COMMAND_BASEBACKUP:
+		/* progress reporting is optional for these */
+		if (a[PROGRESS_BASEBACKUP_BACKUP_TOTAL] >= 0)
+		{
+			Assert(a[PROGRESS_BASEBACKUP_BACKUP_STREAMED] <=
+					a[PROGRESS_BASEBACKUP_BACKUP_TOTAL]);
+			Assert(a[PROGRESS_BASEBACKUP_TBLSPC_STREAMED] <=
+					a[PROGRESS_BASEBACKUP_TBLSPC_TOTAL]);
+		}
+		break;
+
+	case PROGRESS_COMMAND_COPY:
+#if 0
+// This currently fails file_fdw tests, since pgstat_prorgress evidently fails
+// to support simultaneous copy commands, as happens during JOIN.
+		/* bytes progress is not available in all cases */
+		if (a[PROGRESS_COPY_BYTES_TOTAL] > 0)
+			// Assert(a[PROGRESS_COPY_BYTES_PROCESSED] <= a[PROGRESS_COPY_BYTES_TOTAL]);
+			if (a[PROGRESS_COPY_BYTES_PROCESSED] > a[PROGRESS_COPY_BYTES_TOTAL])
+				elog(WARNING, "PROGRESS_COPY_BYTES_PROCESSED %ld %ld",
+							a[PROGRESS_COPY_BYTES_PROCESSED],
+							a[PROGRESS_COPY_BYTES_TOTAL]);
+
+		break;
+#endif
+
+	case PROGRESS_COMMAND_INVALID:
+		break; /* Do nothing */
+	}
+}
+
 /*-----------
  * pgstat_progress_update_param() -
  *
@@ -105,6 +182,8 @@ pgstat_progress_end_command(void)
 	if (beentry->st_progress_command == PROGRESS_COMMAND_INVALID)
 		return;
 
+	pgstat_progress_asserts();
+
 	PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
 	beentry->st_progress_command = PROGRESS_COMMAND_INVALID;
 	beentry->st_progress_command_target = InvalidOid;
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index bebb9620b27..e07527d80f2 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3015,6 +3015,8 @@ typedef struct IndexStmt
 	bool		if_not_exists;	/* just do nothing if index already exists? */
 	bool		reset_default_tblspc;	/* reset default_tablespace prior to
 										 * executing */
+	int			nparts_done;	/* number of partitions processed during
+								 * index creation */
 } IndexStmt;
 
 /* ----------------------
-- 
2.25.1

