diff --git a/src/bin/pgcopydb/catalog.c b/src/bin/pgcopydb/catalog.c index b86b8768..02b025f4 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 7d1cccf3..af91f1c7 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 f836e24d..446002cf 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 8bcb448e..61fcab6d 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 4b52527a..f1581554 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 12f56182..b94a5588 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;