Report actual wire bytes in COPY (cumulative) Transfer column

src/bin/pgcopydb/catalog.c

@@ -9533,6 +9533,74 @@ catalog_count_summary_done_fetch(SQLiteQuery *query)
 }
 
 
+/*
+ * catalog_count_bytes returns byte totals used by list progress to show
+ * transfer volume.  total is the sum of source catalog sizes (s_table.bytes);
+ * transferred is the sum of summary.bytes for all table rows — done rows
+ * hold their final bytesTransmitted, in-progress rows hold the last value
+ * flushed by the periodic stats hook (every ~5 s with PID-based jitter).
+ */
+bool
+catalog_count_bytes(DatabaseCatalog *catalog, CatalogBytesCounts *count)
+{
+	sqlite3 *db = catalog->db;
+
+	if (db == NULL)
+	{
+		log_error("BUG: catalog_count_bytes: db is NULL");
+		return false;
+	}
+
+	char *sql =
+		"select "
+		"  coalesce((select sum(bytes) from s_table), 0) as total, "
+		"  coalesce((select sum(bytes) from summary"
+		"             where tableoid is not null and done_time_epoch is not null), 0)"
+		"    as done, "
+		"  coalesce((select sum(bytes) from summary"
+		"             where tableoid is not null and done_time_epoch is null), 0)"
+		"    as in_progress";
+
+	SQLiteQuery query = {
+		.context = count,
+		.fetchFunction = &catalog_count_bytes_fetch
+	};
+
+	if (!catalog_sql_prepare(db, sql, &query))
+	{
+		/* errors have already been logged */
+		return false;
+	}
+
+	/* now execute the query, which returns exactly one row */
+	if (!catalog_sql_execute_once(&query))
+	{
+		/* errors have already been logged */
+		return false;
+	}
+
+	return true;
+}
+
+
+/*
+ * catalog_count_bytes_fetch fetches a CatalogBytesCounts from a query result.
+ */
+bool
+catalog_count_bytes_fetch(SQLiteQuery *query)
+{
+	CatalogBytesCounts *count = (CatalogBytesCounts *) query->context;
+
+	bzero(count, sizeof(CatalogBytesCounts));
+
+	count->total = sqlite3_column_int64(query->ppStmt, 0);
+	count->done = sqlite3_column_int64(query->ppStmt, 1);
+	count->inProgress = sqlite3_column_int64(query->ppStmt, 2);
+
+	return true;
+}
+
+
 /*
  * catalog_add_timeline_history inserts a timeline history entry to our
  * internal catalogs database.

src/bin/pgcopydb/catalog.h

@@ -737,6 +737,17 @@ bool catalog_count_summary_done(DatabaseCatalog *catalog,
 bool catalog_count_summary_done_fetch(SQLiteQuery *query);
 
 
+typedef struct CatalogBytesCounts
+{
+	uint64_t total;       /* sum of s_table.bytes (source catalog sizes) */
+	uint64_t done;        /* sum of summary.bytes for completed tables */
+	uint64_t inProgress;  /* sum of summary.bytes for in-progress tables (last flush) */
+} CatalogBytesCounts;
+
+bool catalog_count_bytes(DatabaseCatalog *catalog, CatalogBytesCounts *count);
+bool catalog_count_bytes_fetch(SQLiteQuery *query);
+
+
 /*
  * Logical decoding
  */

src/bin/pgcopydb/cli_list.c

@@ -2202,6 +2202,21 @@ cli_list_progress(int argc, char **argv)
 				progress.indexCount,
 				progress.indexInProgress.count,
 				progress.indexDoneCount);
+
+		char totalBytesPretty[BUFSIZE] = { 0 };
+		char doneBytesPretty[BUFSIZE] = { 0 };
+		char inProgressBytesPretty[BUFSIZE] = { 0 };
+
+		pretty_print_bytes(totalBytesPretty, BUFSIZE, progress.totalBytes);
+		pretty_print_bytes(doneBytesPretty, BUFSIZE, progress.doneBytes);
+		pretty_print_bytes(inProgressBytesPretty, BUFSIZE,
+						   progress.inProgressBytes);
+
+		fformat(stdout, "%12s | %12s | %12s | %12s\n",
+				"Bytes",
+				totalBytesPretty,
+				inProgressBytesPretty,
+				doneBytesPretty);
 	}
 }
 

src/bin/pgcopydb/progress.c

@@ -719,6 +719,18 @@ copydb_update_progress(CopyDataSpec *copySpecs, CopyProgress *progress)
 		return false;
 	}
 
+	CatalogBytesCounts bytes = { 0 };
+
+	if (!catalog_count_bytes(sourceDB, &bytes))
+	{
+		/* errors have already been logged */
+		return false;
+	}
+
+	progress->totalBytes = bytes.total;
+	progress->doneBytes = bytes.done;
+	progress->inProgressBytes = bytes.inProgress;
+
 	return true;
 }
 
@@ -974,5 +986,29 @@ copydb_progress_as_json(CopyDataSpec *copySpecs,
 
 	json_object_set_value(jsobj, "indexes", jsIndex);
 
+	/* byte counts */
+	JSON_Value *jsBytes = json_value_init_object();
+	JSON_Object *jsBytesObj = json_value_get_object(jsBytes);
+
+	json_object_set_number(jsBytesObj, "total", (double) progress->totalBytes);
+	json_object_set_number(jsBytesObj, "done", (double) progress->doneBytes);
+	json_object_set_number(jsBytesObj, "in-progress",
+						   (double) progress->inProgressBytes);
+
+	char totalBytesPretty[BUFSIZE] = { 0 };
+	char doneBytesPretty[BUFSIZE] = { 0 };
+	char inProgressBytesPretty[BUFSIZE] = { 0 };
+
+	pretty_print_bytes(totalBytesPretty, BUFSIZE, progress->totalBytes);
+	pretty_print_bytes(doneBytesPretty, BUFSIZE, progress->doneBytes);
+	pretty_print_bytes(inProgressBytesPretty, BUFSIZE, progress->inProgressBytes);
+
+	json_object_set_string(jsBytesObj, "total-pretty", totalBytesPretty);
+	json_object_set_string(jsBytesObj, "done-pretty", doneBytesPretty);
+	json_object_set_string(jsBytesObj, "in-progress-pretty",
+						   inProgressBytesPretty);
+
+	json_object_set_value(jsobj, "bytes", jsBytes);
+
 	return true;
 }

src/bin/pgcopydb/progress.h

@@ -40,6 +40,10 @@ typedef struct CopyProgress
 	int indexDoneCount;
 	SourceIndexArray indexInProgress;
 	CopyIndexSummaryArray indexSummaryArray;
+
+	uint64_t totalBytes;      /* source catalog bytes, all tables */
+	uint64_t doneBytes;       /* bytes from completed tables (final) */
+	uint64_t inProgressBytes; /* bytes from in-progress tables (last flush) */
 } CopyProgress;
 
 

src/bin/pgcopydb/table-data.c

@@ -1304,7 +1304,7 @@ copydb_mark_table_as_done(CopyDataSpec *specs,
 	if (!summary_increment_timing(sourceDB,
 								  TIMING_SECTION_COPY_DATA,
 								  1, /* count */
-								  tableSpecs->sourceTable->bytes,
+								  tableSpecs->summary.bytesTransmitted,
 								  tableSpecs->summary.durationMs))
 	{
 		/* errors have already been logged */
@@ -1526,14 +1526,20 @@ copydb_update_copy_stats_hook(void *ctx, CopyStats *stats)
 
 	uint64_t now = time(NULL);
 
+	/* refrain from updating the statistics too often */
+	const uint64_t WRITE_INTERVAL_SECS = 5;
+
 	if (context->lastWrite == 0)
 	{
-		context->lastWrite = now;
+		/*
+		 * Spread the first flush across [1, WRITE_INTERVAL_SECS] seconds
+		 * using the worker's PID as a deterministic jitter source.  Without
+		 * this, all N table workers initialize at roughly the same instant and
+		 * then hammer SQLite in lock-step every WRITE_INTERVAL_SECS seconds.
+		 */
+		context->lastWrite = now - (getpid() % WRITE_INTERVAL_SECS);
 	}
 
-	/* refrain from updating the statistics too often */
-	const uint64_t WRITE_INTERVAL_SECS = 5;
-
 	if ((now - context->lastWrite) < WRITE_INTERVAL_SECS)
 	{
 		return true;