From ffa3fc916becb0b38ea0e5acadaf4e498a629066 Mon Sep 17 00:00:00 2001 From: Dimitri Fontaine Date: Tue, 23 Jun 2026 20:44:08 +0200 Subject: [PATCH] fix: report actual wire bytes in COPY (cumulative) Transfer column (#258) (#1002) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: report actual wire bytes in COPY (cumulative) Transfer column (#258) The top-level summary has always had a Transfer column in the COPY (cumulative) row but it was populated with sourceTable->bytes — the on-disk size from pg_table_size() on the source — rather than the actual bytes sent over the wire during COPY. The real wire byte count is already tracked buffer-by-buffer in CopyTableSummary.bytesTransmitted (pgsql.c) and stored in the summary table after each table copy (table-data.c:1467/1502). The only gap was that summary_increment_timing for TIMING_SECTION_COPY_DATA was reading the catalog size instead of the transmitted count. Changing that one argument makes the COPY (cumulative) row display the true amount of data moved across the wire, e.g.: COPY (cumulative) both 12h01m 47 GB 4 Source catalog size, target relation size, and wire bytes legitimately differ due to bloat, TOAST, protocol framing, and storage version differences, so the catalog value was never a useful proxy. Closes #258 * feat: show bytes transferred in list progress (#258) pgcopydb list progress now prints a Bytes row alongside Tables and Indexes: Tables | 42 | 3 | 39 Indexes | 150 | 12 | 138 Bytes | 100 GB | | 95 GB The 'Done' column accumulates bytes from: - completed tables: their final bytesTransmitted from summary_finish_table - in-progress tables: the last periodic flush from summary_update_table_copy_stats That hook already wrote bytesTransmitted to summary.bytes every 5 s; what was missing was reading it back in copydb_update_progress. Jitter is added to the periodic flush so N table workers do not all wake at the same second. Each worker's first write fires between 1 and WRITE_INTERVAL_SECS seconds after startup, offset by getpid() % WRITE_INTERVAL_SECS. Subsequent writes remain every WRITE_INTERVAL_SECS seconds, but staggered. rand() is not used (it is banned); getpid() is a deterministic, non-repeating source adequate for this purpose. The JSON output gains a 'bytes' object with 'total', 'transferred', 'total-pretty', and 'transferred-pretty' fields. (cherry picked from commit fe9efbcecbc02c333ef393570295ad4fc0b8078a) --- src/bin/pgcopydb/catalog.c | 68 +++++++++++++++++++++++++++++++++++ src/bin/pgcopydb/catalog.h | 11 ++++++ src/bin/pgcopydb/cli_list.c | 15 ++++++++ src/bin/pgcopydb/progress.c | 36 +++++++++++++++++++ src/bin/pgcopydb/progress.h | 4 +++ src/bin/pgcopydb/table-data.c | 16 ++++++--- 6 files changed, 145 insertions(+), 5 deletions(-) diff --git a/src/bin/pgcopydb/catalog.c b/src/bin/pgcopydb/catalog.c index b86b87687..02b025f4d 100644 --- a/src/bin/pgcopydb/catalog.c +++ b/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. diff --git a/src/bin/pgcopydb/catalog.h b/src/bin/pgcopydb/catalog.h index 7d1cccf35..af91f1c70 100644 --- a/src/bin/pgcopydb/catalog.h +++ b/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 */ diff --git a/src/bin/pgcopydb/cli_list.c b/src/bin/pgcopydb/cli_list.c index f836e24d8..446002cf3 100644 --- a/src/bin/pgcopydb/cli_list.c +++ b/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); } } diff --git a/src/bin/pgcopydb/progress.c b/src/bin/pgcopydb/progress.c index 8bcb448ea..61fcab6dd 100644 --- a/src/bin/pgcopydb/progress.c +++ b/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; } diff --git a/src/bin/pgcopydb/progress.h b/src/bin/pgcopydb/progress.h index 4b52527ad..f15815545 100644 --- a/src/bin/pgcopydb/progress.h +++ b/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; diff --git a/src/bin/pgcopydb/table-data.c b/src/bin/pgcopydb/table-data.c index 12f561828..b94a5588e 100644 --- a/src/bin/pgcopydb/table-data.c +++ b/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;