Copilot commented on code in PR #1757:
URL: https://github.com/apache/cloudberry/pull/1757#discussion_r3279570792
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
+/* Additional headers for extension functionality */
+#include "access/xact.h"
+#include "executor/spi.h"
+#include "fmgr.h"
+#include "lib/stringinfo.h"
+#include "pgstat.h"
+#include "tcop/utility.h"
+
+#include "catalog/namespace.h"
+#include "cdb/cdbvars.h"
+#include "commands/defrem.h"
+#include "funcapi.h"
+
+#include "utils/builtins.h"
+#include "utils/datum.h"
+#include "utils/guc.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+
+#include <assert.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+#include <limits.h>
+
+#define FILEINFO_ARGS_CNT 5
+#define HOUR_TIME 3600000 /* milliseconds in hour */
+#define MINUTE_TIME 60000 /* milliseconds in minute */
+#define FILE_NAPTIME 1 /* default naptime between file processing in
milliseconds */
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(get_stats_for_database);
+PG_FUNCTION_INFO_V1(relsizes_collect_stats_once);
+Datum get_stats_for_database(PG_FUNCTION_ARGS);
+Datum relsizes_collect_stats_once(PG_FUNCTION_ARGS);
+
+static void worker_sigterm(SIGNAL_ARGS);
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction);
+static int update_segment_file_map_table(void);
+static int update_table_sizes_history(void);
+static void get_stats_for_databases(Oid *databases_oids, int databases_cnt,
bool fast);
+static void run_database_stats_worker(bool fast, Oid db);
+static int plugin_created(void);
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle);
+static int delete_data_in_history(void);
+static int put_data_into_history(void);
+void _PG_init(void);
+
+void relsizes_collect_stats(Datum main_arg);
+void relsizes_database_stats_job(Datum args);
+
+/* Global variables */
+static int worker_restart_naptime = 0;
+static int worker_database_naptime = 0;
+static int worker_file_naptime = 0;
+static bool enabled = false;
+
+static volatile sig_atomic_t got_sigterm = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+static_assert(sizeof(Datum) == sizeof(DbWorkerArg), "Invalid size of structure
in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM in background worker processes.
+ *
+ * This handler is called when the postmaster requests the background worker
+ * to shut down. It sets the got_sigterm flag and wakes up the main worker
+ * loop by setting the process latch.
+ *
+ * The function follows PostgreSQL signal handling conventions:
+ * - Saves and restores errno
+ * - Uses only async-signal-safe operations
+ * - Sets a flag that the main loop can check
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Wait for a background worker to stop with timeout and error handling.
+ *
+ * This is a modified version that adds timeout functionality and improved
+ * error handling to prevent infinite loops in case of hung workers.
+ * Returns BGWH_STOPPED on success, BGWH_POSTMASTER_DIED on error/timeout.
+ */
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle) {
+ BgwHandleStatus status = BGWH_NOT_YET_STARTED;
+ int rc;
+ int attempts = 0;
+ const int max_attempts = 5 * HOUR_TIME / 100; /* maximum 5 hours wait time
*/
+
+ PG_TRY();
+ {
+ while (attempts < max_attempts) {
+ pid_t pid;
+
+ status = GetBackgroundWorkerPid(handle, &pid);
+ if (status == BGWH_STOPPED) {
+ return status;
+ }
+
+ /* Add 100ms timeout instead of infinite wait */
+ rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT |
WL_POSTMASTER_DEATH, 100L, WAIT_EVENT_BGWORKER_SHUTDOWN);
+
+ ResetLatch(&MyProc->procLatch);
+
+ if (rc & WL_POSTMASTER_DEATH) {
+ status = BGWH_POSTMASTER_DIED;
+ break;
+ }
+
+ /* Check for interrupts but don't let them break the entire
process */
+ if (QueryCancelPending || ProcDiePending) {
+ ereport(WARNING,
(errmsg("WaitForBackgroundWorkerShutdownSafely: received interrupt signal,
stopping wait")));
+ status = BGWH_POSTMASTER_DIED; /* Return status as if
postmaster died */
+ break;
+ }
+
+ attempts++;
+ }
+
+ /* If maximum attempts reached */
+ if (attempts >= max_attempts) {
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
timeout after %d attempts", max_attempts)));
+ status = BGWH_POSTMASTER_DIED; /* Return error status */
+ }
+ }
+ PG_CATCH();
+ {
+ /* Log error but do NOT re-throw exception */
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
caught exception, returning error status")));
+ /* Return error status instead of PG_RE_THROW() */
+ return BGWH_POSTMASTER_DIED;
+ }
+ PG_END_TRY();
+ return status;
+}
+
+/*
+ * Retrieve list of database OIDs from the catalog.
+ *
+ * This function queries pg_database to get all user databases (excluding
+ * system databases like template0, template1, diskquota, and gpperfmon).
+ *
+ * Parameters:
+ * databases_cnt - Output parameter, set to number of databases found
+ * ctx - Memory context to allocate result in (for cross-call persistence)
+ * create_transaction - Whether to create a new transaction for the query
+ *
+ * Returns:
+ * Array of OIDs allocated in ctx, or NULL on error.
+ * The array length is databases_cnt.
+ *
+ * Note: Caller is responsible for freeing the returned memory when done.
+ */
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction) {
+ const char *sql =
+ "SELECT oid"
+ " FROM pg_database"
+ " WHERE datname NOT IN ('template0', 'template1', 'diskquota',
'gpperfmon')";
+ const char *error = NULL;
+
+ Oid *databases_oids = NULL;
+ *databases_cnt = 0;
+
+ if (create_transaction) {
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+ if (create_transaction) {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_execute(sql, true, 0) != SPI_OK_SELECT) {
+ error = "get_databases_oids: SPI_execute failed (select datname, oid)";
+ goto finish_spi;
+ }
+
+ /* Prepare tuple processing variables */
+
+ *databases_cnt = SPI_processed;
+ MemoryContext old_context = MemoryContextSwitchTo(ctx);
+ databases_oids = palloc((*databases_cnt) * sizeof(*databases_oids));
+ MemoryContextSwitchTo(old_context);
+
+ for (int i = 0; i < SPI_processed; ++i) {
+ Datum oid_datum;
+ bool oid_nullable;
+
+ heap_deform_tuple(SPI_tuptable->vals[i], SPI_tuptable->tupdesc,
&oid_datum, &oid_nullable);
Review Comment:
`heap_deform_tuple()` is being called with pointers to a single
`Datum`/`bool` instead of arrays sized to the tuple descriptor’s natts. This
will write past the provided variables and can corrupt memory. Use
`SPI_getbinval()` (col 1) or allocate `Datum values[Natts]`/`bool nulls[Natts]`
arrays before calling `heap_deform_tuple()`.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
+/* Additional headers for extension functionality */
+#include "access/xact.h"
+#include "executor/spi.h"
+#include "fmgr.h"
+#include "lib/stringinfo.h"
+#include "pgstat.h"
+#include "tcop/utility.h"
+
+#include "catalog/namespace.h"
+#include "cdb/cdbvars.h"
+#include "commands/defrem.h"
+#include "funcapi.h"
+
+#include "utils/builtins.h"
+#include "utils/datum.h"
+#include "utils/guc.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+
+#include <assert.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+#include <limits.h>
+
+#define FILEINFO_ARGS_CNT 5
+#define HOUR_TIME 3600000 /* milliseconds in hour */
+#define MINUTE_TIME 60000 /* milliseconds in minute */
+#define FILE_NAPTIME 1 /* default naptime between file processing in
milliseconds */
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(get_stats_for_database);
+PG_FUNCTION_INFO_V1(relsizes_collect_stats_once);
+Datum get_stats_for_database(PG_FUNCTION_ARGS);
+Datum relsizes_collect_stats_once(PG_FUNCTION_ARGS);
+
+static void worker_sigterm(SIGNAL_ARGS);
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction);
+static int update_segment_file_map_table(void);
+static int update_table_sizes_history(void);
+static void get_stats_for_databases(Oid *databases_oids, int databases_cnt,
bool fast);
+static void run_database_stats_worker(bool fast, Oid db);
+static int plugin_created(void);
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle);
+static int delete_data_in_history(void);
+static int put_data_into_history(void);
+void _PG_init(void);
+
+void relsizes_collect_stats(Datum main_arg);
+void relsizes_database_stats_job(Datum args);
+
+/* Global variables */
+static int worker_restart_naptime = 0;
+static int worker_database_naptime = 0;
+static int worker_file_naptime = 0;
+static bool enabled = false;
+
+static volatile sig_atomic_t got_sigterm = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+static_assert(sizeof(Datum) == sizeof(DbWorkerArg), "Invalid size of structure
in DbWorkerArg");
Review Comment:
`static_assert(...)` is used in this `.c` file. Cloudberry’s
`src/include/c.h` provides `StaticAssertDecl/StaticAssertStmt` for C builds;
`static_assert` is not guaranteed to be available (and can fail to compile
under the project’s C standard). Please switch this to the project-provided
StaticAssert macro appropriate for a file-scope declaration.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
Review Comment:
This file uses `DIR`, `struct dirent`, `AllocateDir()`, `ReadDir()`, and
`FreeDir()` but does not include `storage/fd.h` (which provides the prototypes
and includes `<dirent.h>`). This can fail to compile under stricter warning
levels. Add the required include rather than relying on indirect includes.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
+/* Additional headers for extension functionality */
+#include "access/xact.h"
+#include "executor/spi.h"
+#include "fmgr.h"
+#include "lib/stringinfo.h"
+#include "pgstat.h"
+#include "tcop/utility.h"
+
+#include "catalog/namespace.h"
+#include "cdb/cdbvars.h"
+#include "commands/defrem.h"
+#include "funcapi.h"
+
+#include "utils/builtins.h"
+#include "utils/datum.h"
+#include "utils/guc.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+
+#include <assert.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+#include <limits.h>
+
+#define FILEINFO_ARGS_CNT 5
+#define HOUR_TIME 3600000 /* milliseconds in hour */
+#define MINUTE_TIME 60000 /* milliseconds in minute */
+#define FILE_NAPTIME 1 /* default naptime between file processing in
milliseconds */
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(get_stats_for_database);
+PG_FUNCTION_INFO_V1(relsizes_collect_stats_once);
+Datum get_stats_for_database(PG_FUNCTION_ARGS);
+Datum relsizes_collect_stats_once(PG_FUNCTION_ARGS);
+
+static void worker_sigterm(SIGNAL_ARGS);
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction);
+static int update_segment_file_map_table(void);
+static int update_table_sizes_history(void);
+static void get_stats_for_databases(Oid *databases_oids, int databases_cnt,
bool fast);
+static void run_database_stats_worker(bool fast, Oid db);
+static int plugin_created(void);
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle);
+static int delete_data_in_history(void);
+static int put_data_into_history(void);
+void _PG_init(void);
+
+void relsizes_collect_stats(Datum main_arg);
+void relsizes_database_stats_job(Datum args);
+
+/* Global variables */
+static int worker_restart_naptime = 0;
+static int worker_database_naptime = 0;
+static int worker_file_naptime = 0;
+static bool enabled = false;
+
+static volatile sig_atomic_t got_sigterm = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+static_assert(sizeof(Datum) == sizeof(DbWorkerArg), "Invalid size of structure
in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM in background worker processes.
+ *
+ * This handler is called when the postmaster requests the background worker
+ * to shut down. It sets the got_sigterm flag and wakes up the main worker
+ * loop by setting the process latch.
+ *
+ * The function follows PostgreSQL signal handling conventions:
+ * - Saves and restores errno
+ * - Uses only async-signal-safe operations
+ * - Sets a flag that the main loop can check
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Wait for a background worker to stop with timeout and error handling.
+ *
+ * This is a modified version that adds timeout functionality and improved
+ * error handling to prevent infinite loops in case of hung workers.
+ * Returns BGWH_STOPPED on success, BGWH_POSTMASTER_DIED on error/timeout.
+ */
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle) {
+ BgwHandleStatus status = BGWH_NOT_YET_STARTED;
+ int rc;
+ int attempts = 0;
+ const int max_attempts = 5 * HOUR_TIME / 100; /* maximum 5 hours wait time
*/
+
+ PG_TRY();
+ {
+ while (attempts < max_attempts) {
+ pid_t pid;
+
+ status = GetBackgroundWorkerPid(handle, &pid);
+ if (status == BGWH_STOPPED) {
+ return status;
+ }
+
+ /* Add 100ms timeout instead of infinite wait */
+ rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT |
WL_POSTMASTER_DEATH, 100L, WAIT_EVENT_BGWORKER_SHUTDOWN);
+
+ ResetLatch(&MyProc->procLatch);
+
+ if (rc & WL_POSTMASTER_DEATH) {
+ status = BGWH_POSTMASTER_DIED;
+ break;
+ }
+
+ /* Check for interrupts but don't let them break the entire
process */
+ if (QueryCancelPending || ProcDiePending) {
+ ereport(WARNING,
(errmsg("WaitForBackgroundWorkerShutdownSafely: received interrupt signal,
stopping wait")));
+ status = BGWH_POSTMASTER_DIED; /* Return status as if
postmaster died */
+ break;
+ }
+
+ attempts++;
+ }
+
+ /* If maximum attempts reached */
+ if (attempts >= max_attempts) {
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
timeout after %d attempts", max_attempts)));
+ status = BGWH_POSTMASTER_DIED; /* Return error status */
+ }
+ }
+ PG_CATCH();
+ {
+ /* Log error but do NOT re-throw exception */
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
caught exception, returning error status")));
+ /* Return error status instead of PG_RE_THROW() */
+ return BGWH_POSTMASTER_DIED;
+ }
+ PG_END_TRY();
+ return status;
+}
+
+/*
+ * Retrieve list of database OIDs from the catalog.
+ *
+ * This function queries pg_database to get all user databases (excluding
+ * system databases like template0, template1, diskquota, and gpperfmon).
+ *
+ * Parameters:
+ * databases_cnt - Output parameter, set to number of databases found
+ * ctx - Memory context to allocate result in (for cross-call persistence)
+ * create_transaction - Whether to create a new transaction for the query
+ *
+ * Returns:
+ * Array of OIDs allocated in ctx, or NULL on error.
+ * The array length is databases_cnt.
+ *
+ * Note: Caller is responsible for freeing the returned memory when done.
+ */
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction) {
+ const char *sql =
+ "SELECT oid"
+ " FROM pg_database"
+ " WHERE datname NOT IN ('template0', 'template1', 'diskquota',
'gpperfmon')";
+ const char *error = NULL;
+
+ Oid *databases_oids = NULL;
+ *databases_cnt = 0;
+
+ if (create_transaction) {
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+ if (create_transaction) {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_execute(sql, true, 0) != SPI_OK_SELECT) {
+ error = "get_databases_oids: SPI_execute failed (select datname, oid)";
+ goto finish_spi;
+ }
+
+ /* Prepare tuple processing variables */
+
+ *databases_cnt = SPI_processed;
+ MemoryContext old_context = MemoryContextSwitchTo(ctx);
+ databases_oids = palloc((*databases_cnt) * sizeof(*databases_oids));
+ MemoryContextSwitchTo(old_context);
+
+ for (int i = 0; i < SPI_processed; ++i) {
+ Datum oid_datum;
+ bool oid_nullable;
+
+ heap_deform_tuple(SPI_tuptable->vals[i], SPI_tuptable->tupdesc,
&oid_datum, &oid_nullable);
+
+ databases_oids[i] = DatumGetObjectId(oid_datum);
+ }
+
+finish_spi:
+ SPI_finish();
+finish_transaction:
+ if (create_transaction) {
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+ pgstat_report_activity(STATE_IDLE, NULL);
+ }
+
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ return NULL; /* Return NULL on error */
+ }
+
+ return databases_oids;
+}
+
+/*
+ * Update the segment_file_map table with current relation file mappings.
+ *
+ * This function refreshes the mapping between relation OIDs and their
+ * physical file nodes across all segments. It first deletes the existing
+ * data and then repopulates it by querying pg_class on all segments.
+ *
+ * The mapping is essential for correlating file statistics collected
+ * from the filesystem with actual database relations.
+ *
+ * Returns:
+ * 0 on success, negative value on error
+ *
+ * Note: This function assumes it's running within an active SPI context.
+ */
+static int update_segment_file_map_table() {
+ int retcode = 0;
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_map";
+ char *sql_insert = "INSERT INTO relsizes_stats_schema.segment_file_map
SELECT gp_segment_id, oid, relfilenode FROM "
+ "gp_dist_random('pg_class')";
+ char *error = NULL;
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "update_segment_file_map_table: failed to delete from table";
+ goto cleanup;
+ }
+
+ pgstat_report_activity(STATE_RUNNING, sql_insert);
+ retcode = SPI_execute(sql_insert, false, 0);
+ if (retcode != SPI_OK_INSERT) {
+ error = "update_segment_file_map_table: failed to insert new rows into
table";
+ goto cleanup;
+ }
+
+cleanup:
+ pgstat_report_activity(STATE_IDLE, NULL);
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ }
+ return retcode;
+}
+
+/*
+ * Check if a character is a digit (0-9).
+ *
+ * Simple utility function used by fill_relfilenode() to parse
+ * numeric portions of filenames.
+ *
+ * Returns:
+ * true if character is a digit, false otherwise
+ */
+static bool is_number(char symbol) { return '0' <= symbol && symbol <= '9'; }
+
+/*
+ * Extract relfilenode from filename by finding the first sequence of digits
+ * in the filename and converting it to numeric value
+ */
+static unsigned int fill_relfilenode(char *name) {
+ unsigned int result = 0, pos = 0;
+ size_t name_len = strlen(name);
+
+ while (pos < name_len && !is_number(name[pos])) {
+ ++pos;
+ }
+ while (pos < name_len && is_number(name[pos])) {
+ /* Check for overflow to prevent integer overflow */
+ if (result > (UINT_MAX - (name[pos] - '0')) / 10) {
+ break; /* Stop on potential overflow */
+ }
+ result = (result * 10 + (name[pos] - '0'));
+ ++pos;
+ }
+ return result;
+}
+
+/*
+ * Background worker entry point for database-specific statistics collection.
+ *
+ * This function is executed by dynamically spawned background workers to
+ * collect file size statistics for a specific database. Each worker:
+ * 1. Connects to the target database
+ * 2. Verifies the extension is installed
+ * 3. Updates the segment file mapping
+ * 4. Collects file size statistics from all segments
+ * 5. Updates the historical statistics table
+ *
+ * The function runs within its own transaction and handles errors gracefully
+ * by logging warnings rather than aborting the entire collection process.
+ *
+ * Parameters:
+ * args - Background worker argument which contains database OID and the flag
+ * which indicates make pauses or not
+ *
+ * Note: This function is called via the background worker framework and
+ * should not be called directly.
+ */
+void relsizes_database_stats_job(Datum args) {
+ int retcode = 0;
+ char *error = NULL;
+ DbWorkerArg wa = { .d = args };
+
+ optimizer = false;
+ pqsignal(SIGTERM, worker_sigterm);
+ BackgroundWorkerUnblockSignals();
+
+ BackgroundWorkerInitializeConnectionByOid(wa.s.db, InvalidOid, 0);
+
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+
+ retcode = SPI_connect();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: SPI_connect failed";
+ goto finish_transaction;
+ }
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /* Verify extension is installed */
+ int created = plugin_created();
+ if (created < 0) {
+ error = "relsizes_database_stats_job: SPI execute failed while looking
for plugin";
+ goto finish_spi;
+ } else if (created == 0) {
+ goto finish_spi;
+ }
+
+ retcode = update_segment_file_map_table();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating segment_file_map
failed";
+ goto finish_spi;
+ }
+
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_sizes";
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "relsizes_database_stats_job: SPI_execute failed (delete from
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ /* Remove this condition after decision how to upgrade extensions is made.
*/
+ if (SearchSysCacheExists3(PROCNAMEARGSNSP,
+ CStringGetDatum("get_stats_for_database"),
+ PointerGetDatum(buildoidvector((Oid[]){INT4OID}, 1)),
+ ObjectIdGetDatum(get_namespace_oid("relsizes_stats_schema",
true))))
+ {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 1,
+ (Oid[]){INT4OID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId)},
+ NULL, false, 0);
+ } else {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1,
$2)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 2,
+ (Oid[]){OIDOID, BOOLOID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId),
BoolGetDatum(wa.s.fast)},
+ NULL, false, 0);
+ }
+ if (retcode != SPI_OK_INSERT) {
+ error = "relsizes_database_stats_job: SPI_execute failed (insert into
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ retcode = update_table_sizes_history();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating tables sizes history
table failed";
+ goto finish_spi;
+ }
+
+finish_spi:
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ /* Don't abort execution, continue with cleanup */
+ }
+ SPI_finish();
+finish_transaction:
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+ pgstat_report_activity(STATE_IDLE, NULL);
+}
+
+/*
+ * Spawn and manage a background worker for database statistics collection.
+ *
+ * This function creates a new background worker to collect statistics for
+ * a specific database.
+ *
+ * The function:
+ * 1. Configures a new background worker with appropriate settings
+ * 2. Registers and starts the worker
+ * 3. Waits for the worker to complete
+ * 4. Handles any errors during worker execution
+ *
+ * If the worker fails to start or encounters errors during execution,
+ * warnings are logged but the function returns normally to allow
+ * processing of remaining databases.
+ *
+ * Parameters:
+ * fast - Don't make pauses
+ * db - OID of the database which worker will collect statistics from
+ *
+ * Note: This function may take significant time to complete as it waits
+ * for the background worker to finish processing the entire database.
+ */
+static void run_database_stats_worker(bool fast, Oid db) {
+ bool ret;
+ MemoryContext old_ctx;
+ BackgroundWorkerHandle *handle;
+ BgwHandleStatus status;
+
+ /* Configure background worker */
+ BackgroundWorker database_worker = {
+ .bgw_flags = BGWORKER_SHMEM_ACCESS |
BGWORKER_BACKEND_DATABASE_CONNECTION,
+ .bgw_start_time = BgWorkerStart_RecoveryFinished,
+ .bgw_restart_time = BGW_NEVER_RESTART,
+ .bgw_library_name = "gp_relsizes_stats",
+ .bgw_function_name = "relsizes_database_stats_job",
+ .bgw_notify_pid = MyProcPid,
+ .bgw_main_arg = ((DbWorkerArg){ .s.db = db, .s.fast = fast }).d,
+ .bgw_start_rule = NULL,
+ };
+ snprintf(database_worker.bgw_name, BGW_MAXLEN,
"database_relsizes_collector_worker for %u", db);
+ old_ctx = MemoryContextSwitchTo(TopMemoryContext);
+ ret = RegisterDynamicBackgroundWorker(&database_worker, &handle);
+ MemoryContextSwitchTo(old_ctx);
+ if (!ret) {
+ ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES), errmsg("could
not register background process"),
+ errhint("You may need to increase
max_worker_processes.")));
+ }
+ pid_t pid;
+ status = WaitForBackgroundWorkerStartup(handle, &pid);
+ if (status == BGWH_STOPPED)
+ return;
+ if (status != BGWH_STARTED) {
+ ereport(WARNING, (errmsg("Failed to start background worker [%s],
skipping", database_worker.bgw_name)));
+ return;
+ }
+ status = WaitForBackgroundWorkerShutdownSafely(handle);
+ if (status != BGWH_STOPPED) {
+ ereport(WARNING, (errmsg("Failure during background worker execution
[%s], continuing", database_worker.bgw_name)));
+ /* Don't abort execution, just log and continue */
+ }
+}
+
+/*
+ * SQL-callable function to collect file statistics for a database.
+ *
+ * This function scans the filesystem directory corresponding to a database
+ * and returns statistics for all regular files found. It's designed to run
+ * on individual segments to collect local file information.
+ *
+ * The function:
+ * 1. Validates the function call context (must support returning a set)
+ * 2. Sets up a tuplestore for result collection
+ * 3. Scans the database directory (base/<dboid>/)
+ * 4. For each regular file, extracts relfilenode from filename
+ * 5. Collects file size and modification time via lstat()
+ * 6. Returns results as a set of tuples
+ *
+ * Parameters:
+ * Database OID (oid) - identifies which database directory to scan
+ * Fast (bool) - When true, don't sleep between each collect-phase for files
+ *
+ * Returns:
+ * Set of tuples containing:
+ * - segment: current segment ID
+ * - relfilenode: extracted from filename
+ * - filepath: full path to the file
+ * - size: file size in bytes
+ * - mtime: modification time as Unix timestamp
+ *
+ * Note: Includes configurable delays between file processing to reduce I/O
load
+ */
+Datum get_stats_for_database(PG_FUNCTION_ARGS) {
+ int segment_id = GpIdentity.segindex;
+ Oid dboid = PG_GETARG_OID(0);
+ bool fast = (PG_NARGS() < 2) ? false : PG_GETARG_BOOL(1);
+
+ char cwd[PATH_MAX];
+ char *data_dir = NULL;
+ char *error = NULL;
+ char *file_path = NULL;
+
+ if (getcwd(cwd, sizeof(cwd)) == NULL) {
+ error = "get_stats_for_database: failed to get current working
directory";
+ goto finish_data;
+ }
+ data_dir = psprintf("%s/base/%u", cwd, dboid);
+ ReturnSetInfo *rsinfo = (ReturnSetInfo *)fcinfo->resultinfo;
+ /* Validate function call context */
+ if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo)) {
+ error = "get_stats_for_database: set-valued function called in context
that cannot accept a set";
+ goto finish_data;
+ }
+ if (!(rsinfo->allowedModes & SFRM_Materialize)) {
+ error = "get_stats_for_database: materialize mode required, but it is
not allowed in this context";
+ goto finish_data;
+ }
+
+ /* Setup output tuple store */
+ MemoryContext oldcontext =
MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
+ TupleDesc tupdesc;
+ if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) {
+ MemoryContextSwitchTo(oldcontext);
+ error = "get_stats_for_database: incorrect return type in fcinfo (must
be a row type)";
+ goto finish_data;
+ }
+ tupdesc = BlessTupleDesc(tupdesc);
+
+ bool randomAccess = (rsinfo->allowedModes & SFRM_Materialize_Random) != 0;
+ Tuplestorestate *tupstore = tuplestore_begin_heap(randomAccess, false,
work_mem);
+
+ rsinfo->returnMode = SFRM_Materialize;
+ rsinfo->setResult = tupstore;
+ rsinfo->setDesc = tupdesc;
+
+ Datum outputValues[FILEINFO_ARGS_CNT];
+ bool outputNulls[FILEINFO_ARGS_CNT] = { false };
+
+ MemoryContextSwitchTo(oldcontext);
+
+ /* Scan database directory for files */
+ DIR *current_dir = AllocateDir(data_dir);
+ if (!current_dir) {
+ error = "get_stats_for_database: failed to allocate current directory";
+ goto finish_data;
+ }
+
+ struct dirent *file;
+ while ((file = ReadDir(current_dir, data_dir)) != NULL) {
+ char *filename = file->d_name;
+ if (strcmp(filename, ".") == 0 || strcmp(filename, "..") == 0) {
+ continue;
+ }
+
+ file_path = psprintf("%s/%s", data_dir, filename);
+ struct stat stb;
+ if (lstat(file_path, &stb) < 0) {
+ ereport(WARNING,
+ (errmsg("get_stats_for_database: lstat failed with %s file
(unexpected behavior)", file_path)));
+ pfree(file_path);
+ continue;
+ }
+
+ if (S_ISREG(stb.st_mode)) {
+ /* Process regular files and collect size statistics */
+ outputValues[0] = Int32GetDatum(segment_id);
+ outputValues[1] = ObjectIdGetDatum(fill_relfilenode(filename));
+ outputValues[2] = CStringGetTextDatum(file_path);
+ outputValues[3] = Int64GetDatum(stb.st_size);
+ outputValues[4] = Int64GetDatum(stb.st_mtime);
+
+ tuplestore_putvalues(tupstore, tupdesc, outputValues, outputNulls);
+
+ if (fast)
+ CHECK_FOR_INTERRUPTS();
+ else {
+ /* Brief pause between file processing to reduce system load */
+ int retcode = WaitLatch(&MyProc->procLatch,
+ WL_LATCH_SET | WL_TIMEOUT |
WL_POSTMASTER_DEATH,
+ worker_file_naptime, WAIT_EVENT_BUFFER_IO);
+ ResetLatch(&MyProc->procLatch);
+
+ CHECK_FOR_INTERRUPTS();
+
+ if (retcode & WL_POSTMASTER_DEATH) {
+ proc_exit(1);
+ }
+ }
+ }
+ pfree(file_path);
+ }
+
+ FreeDir(current_dir);
+finish_data:
+ pfree(data_dir);
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ /* Don't abort execution, return result */
+ }
Review Comment:
`pfree(data_dir)` is called unconditionally, but `data_dir` can still be
NULL (e.g., if `getcwd()` fails and the function jumps to `finish_data`).
`pfree(NULL)` will crash (unlike `free(NULL)`). Guard the `pfree()` with a NULL
check or ensure `data_dir` is always assigned before `finish_data` can be
reached.
##########
gpcontrib/gp_relsizes_stats/test/sql/gp_relsizes_stats.sql:
##########
@@ -0,0 +1,96 @@
+CREATE EXTENSION gp_relsizes_stats;
+
+-- start_ignore
+DROP TABLE IF EXISTS employees;
+-- end_ignore
+CREATE TABLE employees (
+ employee_id SERIAL PRIMARY KEY,
+ first_name VARCHAR(50) NOT NULL,
+ last_name VARCHAR(50) NOT NULL,
+ department_id INT,
+ date_of_birth DATE
+);
+
+INSERT INTO employees (first_name, last_name, department_id, date_of_birth)
VALUES
+('John', 'Doe', 1, '1988-06-15'),
+('Jane', 'Smith', 2, '1990-07-20'),
+('Emily', 'Jones', 1, '1985-08-30');
+
+SELECT relsizes_stats_schema.relsizes_collect_stats_once();
+
+SELECT size FROM relsizes_stats_schema.table_sizes_history WHERE relname =
'employees';
+
+-- Fill table with a lot of different rows
+insert into employees (first_name, last_name, department_id, date_of_birth)
+select 'First' || i, 'Last' || i, (i % 10) + 1, DATE '1980-01-01' + (i % 365 *
365 / 30)
+from generate_series(1, 10001)i;
+
+SELECT relsizes_stats_schema.relsizes_collect_stats_once();
+
+-- Check that collected stats are correct
+SELECT size FROM relsizes_stats_schema.table_sizes_history WHERE relname =
'employees';
+
+SELECT relsizes_stats_schema.relsizes_collect_stats_once();
+
+-- Validate that after rerun stats collection size of table has not change
+SELECT size FROM relsizes_stats_schema.table_sizes_history WHERE relname =
'employees';
+
+-- Cleanup
+DROP TABLE employees;
+
+
+--
+-- relsizes_collect_stats_once should collect files sizes without pauses
+-- The naptime value is 1ms, so the pauses take at least 10s to process 10k
files.
+-- Check that relsizes_collect_stats_once completes in significantly less time.
+
+-- start_ignore
+DROP TABLE IF EXISTS t;
+CREATE TABLE t (i int)
+DISTRIBUTED RANDOMLY
+PARTITION BY RANGE (i) (PARTITION a START (0) END (10000) EVERY (1));
+-- end_ignore
+
+SELECT EXTRACT(EPOCH FROM LOCALTIMESTAMP(0)) t1 \gset
+
+SELECT relsizes_stats_schema.relsizes_collect_stats_once();
+
+SELECT (EXTRACT(EPOCH FROM LOCALTIMESTAMP(0)) - :t1) < 5;
+
Review Comment:
This assertion is timing-based (`< 5` seconds) and can be flaky on slower CI
machines or under load, causing nondeterministic failures. Consider asserting
on functional behavior instead (e.g., that the fast mode uses
`file_naptime=0`/skips sleeps), or use a much looser threshold / compute
expected upper bound based on configured naptimes and file count.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
+/* Additional headers for extension functionality */
+#include "access/xact.h"
+#include "executor/spi.h"
+#include "fmgr.h"
+#include "lib/stringinfo.h"
+#include "pgstat.h"
+#include "tcop/utility.h"
+
+#include "catalog/namespace.h"
+#include "cdb/cdbvars.h"
+#include "commands/defrem.h"
+#include "funcapi.h"
+
+#include "utils/builtins.h"
+#include "utils/datum.h"
+#include "utils/guc.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+
+#include <assert.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+#include <limits.h>
+
+#define FILEINFO_ARGS_CNT 5
+#define HOUR_TIME 3600000 /* milliseconds in hour */
+#define MINUTE_TIME 60000 /* milliseconds in minute */
+#define FILE_NAPTIME 1 /* default naptime between file processing in
milliseconds */
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(get_stats_for_database);
+PG_FUNCTION_INFO_V1(relsizes_collect_stats_once);
+Datum get_stats_for_database(PG_FUNCTION_ARGS);
+Datum relsizes_collect_stats_once(PG_FUNCTION_ARGS);
+
+static void worker_sigterm(SIGNAL_ARGS);
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction);
+static int update_segment_file_map_table(void);
+static int update_table_sizes_history(void);
+static void get_stats_for_databases(Oid *databases_oids, int databases_cnt,
bool fast);
+static void run_database_stats_worker(bool fast, Oid db);
+static int plugin_created(void);
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle);
+static int delete_data_in_history(void);
+static int put_data_into_history(void);
+void _PG_init(void);
+
+void relsizes_collect_stats(Datum main_arg);
+void relsizes_database_stats_job(Datum args);
+
+/* Global variables */
+static int worker_restart_naptime = 0;
+static int worker_database_naptime = 0;
+static int worker_file_naptime = 0;
+static bool enabled = false;
+
+static volatile sig_atomic_t got_sigterm = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+static_assert(sizeof(Datum) == sizeof(DbWorkerArg), "Invalid size of structure
in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM in background worker processes.
+ *
+ * This handler is called when the postmaster requests the background worker
+ * to shut down. It sets the got_sigterm flag and wakes up the main worker
+ * loop by setting the process latch.
+ *
+ * The function follows PostgreSQL signal handling conventions:
+ * - Saves and restores errno
+ * - Uses only async-signal-safe operations
+ * - Sets a flag that the main loop can check
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Wait for a background worker to stop with timeout and error handling.
+ *
+ * This is a modified version that adds timeout functionality and improved
+ * error handling to prevent infinite loops in case of hung workers.
+ * Returns BGWH_STOPPED on success, BGWH_POSTMASTER_DIED on error/timeout.
+ */
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle) {
+ BgwHandleStatus status = BGWH_NOT_YET_STARTED;
+ int rc;
+ int attempts = 0;
+ const int max_attempts = 5 * HOUR_TIME / 100; /* maximum 5 hours wait time
*/
+
+ PG_TRY();
+ {
+ while (attempts < max_attempts) {
+ pid_t pid;
+
+ status = GetBackgroundWorkerPid(handle, &pid);
+ if (status == BGWH_STOPPED) {
+ return status;
+ }
+
+ /* Add 100ms timeout instead of infinite wait */
+ rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT |
WL_POSTMASTER_DEATH, 100L, WAIT_EVENT_BGWORKER_SHUTDOWN);
+
+ ResetLatch(&MyProc->procLatch);
+
+ if (rc & WL_POSTMASTER_DEATH) {
+ status = BGWH_POSTMASTER_DIED;
+ break;
+ }
+
+ /* Check for interrupts but don't let them break the entire
process */
+ if (QueryCancelPending || ProcDiePending) {
+ ereport(WARNING,
(errmsg("WaitForBackgroundWorkerShutdownSafely: received interrupt signal,
stopping wait")));
+ status = BGWH_POSTMASTER_DIED; /* Return status as if
postmaster died */
+ break;
+ }
+
+ attempts++;
+ }
+
+ /* If maximum attempts reached */
+ if (attempts >= max_attempts) {
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
timeout after %d attempts", max_attempts)));
+ status = BGWH_POSTMASTER_DIED; /* Return error status */
+ }
+ }
+ PG_CATCH();
+ {
+ /* Log error but do NOT re-throw exception */
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
caught exception, returning error status")));
+ /* Return error status instead of PG_RE_THROW() */
+ return BGWH_POSTMASTER_DIED;
+ }
+ PG_END_TRY();
+ return status;
+}
+
+/*
+ * Retrieve list of database OIDs from the catalog.
+ *
+ * This function queries pg_database to get all user databases (excluding
+ * system databases like template0, template1, diskquota, and gpperfmon).
+ *
+ * Parameters:
+ * databases_cnt - Output parameter, set to number of databases found
+ * ctx - Memory context to allocate result in (for cross-call persistence)
+ * create_transaction - Whether to create a new transaction for the query
+ *
+ * Returns:
+ * Array of OIDs allocated in ctx, or NULL on error.
+ * The array length is databases_cnt.
+ *
+ * Note: Caller is responsible for freeing the returned memory when done.
+ */
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction) {
+ const char *sql =
+ "SELECT oid"
+ " FROM pg_database"
+ " WHERE datname NOT IN ('template0', 'template1', 'diskquota',
'gpperfmon')";
+ const char *error = NULL;
+
+ Oid *databases_oids = NULL;
+ *databases_cnt = 0;
+
+ if (create_transaction) {
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+ if (create_transaction) {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
Review Comment:
`PopActiveSnapshot()` is executed whenever `create_transaction` is true, but
`PushActiveSnapshot()` only happens after a successful `SPI_connect()`. If
`SPI_connect()` fails, the code jumps to `finish_transaction` and will pop a
snapshot that was never pushed. Track whether a snapshot was pushed (or move
snapshot push earlier) and only pop when it was.
##########
gpcontrib/gp_relsizes_stats/gp_relsizes_stats.control:
##########
@@ -0,0 +1,5 @@
+# gp_relsizes_stats extension
+comment = 'gp_relsizes_stats - an extension to track table on-disc sizes in
cloudberry'
+default_version = '1.3'
+module_pathname = '$libdir/gp_relsizes_stats'
+trusted = true
Review Comment:
The control file marks the extension as `trusted = true`, which allows
non-superusers to `CREATE EXTENSION`. This extension exposes C functions that
walk the data directory and return filesystem paths/sizes and also registers
background workers; that should not be available to unprivileged users. Please
remove `trusted = true` (or set it to false) unless there is a strong security
review justifying it.
##########
gpcontrib/gp_relsizes_stats/test/sql/grants.sql:
##########
@@ -0,0 +1,82 @@
+-- Check that user who has created gp_relsizes_stats have privileges to use all
+-- tables, views and functions from the extension.
+-- Check that this user can grant this privileges to others.
+
+-- start_ignore
+DROP DATABASE IF EXISTS db1;
+DROP ROLE IF EXISTS user1, user2;
+-- end_ignore
+
+SELECT '\! cp "' || setting || '/pg_hba.conf" "' || setting ||
'/pg_hba.conf.backup"' as cp_backup
+FROM pg_settings
+WHERE name = 'data_directory' \gset
+
+:cp_backup
+
+SELECT '\! echo "local all user1,user2 trust" >> ' || setting ||
'/pg_hba.conf' as add_users
+FROM pg_settings
+WHERE name = 'data_directory' \gset
+
+:add_users
+
+-- start_ignore
+\! gpstop -u
+-- end_ignore
Review Comment:
This regression test edits `pg_hba.conf` in-place and runs `gpstop -u`.
That’s a high-impact side effect (requires gpstop in PATH, changes cluster
auth, and can interfere with parallel test runs if cleanup is skipped on
failure). Prefer avoiding HBA edits/restarts in `installcheck` tests (e.g., use
`SET ROLE`/`SET SESSION AUTHORIZATION`, or design the privilege test so it
doesn’t require reconnecting as a different OS-authenticated user).
##########
gpcontrib/gp_relsizes_stats/Makefile:
##########
@@ -0,0 +1,20 @@
+MODULE_big = gp_relsizes_stats
+OBJS = ./src/gp_relsizes_stats.o
+EXTENSION = gp_relsizes_stats
+EXTVERSION = 1.3
+DATA = $(wildcard sql/*--*.sql)
+REGRESS = grants gp_relsizes_stats
+REGRESS_OPTS = --inputdir=test/
+PGFILEDESC = "gp_relsizes_stats - an extension to track table on-disc
sizes in greenplum"
Review Comment:
`PGFILEDESC` still says "...sizes in greenplum" even though the extension is
being ported into Cloudberry. Updating this string avoids confusion in
installed extension metadata and aligns with the README/control file.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
+/* Additional headers for extension functionality */
+#include "access/xact.h"
+#include "executor/spi.h"
+#include "fmgr.h"
+#include "lib/stringinfo.h"
+#include "pgstat.h"
+#include "tcop/utility.h"
+
+#include "catalog/namespace.h"
+#include "cdb/cdbvars.h"
+#include "commands/defrem.h"
+#include "funcapi.h"
+
+#include "utils/builtins.h"
+#include "utils/datum.h"
+#include "utils/guc.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+
+#include <assert.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+#include <limits.h>
+
+#define FILEINFO_ARGS_CNT 5
+#define HOUR_TIME 3600000 /* milliseconds in hour */
+#define MINUTE_TIME 60000 /* milliseconds in minute */
+#define FILE_NAPTIME 1 /* default naptime between file processing in
milliseconds */
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(get_stats_for_database);
+PG_FUNCTION_INFO_V1(relsizes_collect_stats_once);
+Datum get_stats_for_database(PG_FUNCTION_ARGS);
+Datum relsizes_collect_stats_once(PG_FUNCTION_ARGS);
+
+static void worker_sigterm(SIGNAL_ARGS);
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction);
+static int update_segment_file_map_table(void);
+static int update_table_sizes_history(void);
+static void get_stats_for_databases(Oid *databases_oids, int databases_cnt,
bool fast);
+static void run_database_stats_worker(bool fast, Oid db);
+static int plugin_created(void);
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle);
+static int delete_data_in_history(void);
+static int put_data_into_history(void);
+void _PG_init(void);
+
+void relsizes_collect_stats(Datum main_arg);
+void relsizes_database_stats_job(Datum args);
+
+/* Global variables */
+static int worker_restart_naptime = 0;
+static int worker_database_naptime = 0;
+static int worker_file_naptime = 0;
+static bool enabled = false;
+
+static volatile sig_atomic_t got_sigterm = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+static_assert(sizeof(Datum) == sizeof(DbWorkerArg), "Invalid size of structure
in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM in background worker processes.
+ *
+ * This handler is called when the postmaster requests the background worker
+ * to shut down. It sets the got_sigterm flag and wakes up the main worker
+ * loop by setting the process latch.
+ *
+ * The function follows PostgreSQL signal handling conventions:
+ * - Saves and restores errno
+ * - Uses only async-signal-safe operations
+ * - Sets a flag that the main loop can check
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Wait for a background worker to stop with timeout and error handling.
+ *
+ * This is a modified version that adds timeout functionality and improved
+ * error handling to prevent infinite loops in case of hung workers.
+ * Returns BGWH_STOPPED on success, BGWH_POSTMASTER_DIED on error/timeout.
+ */
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle) {
+ BgwHandleStatus status = BGWH_NOT_YET_STARTED;
+ int rc;
+ int attempts = 0;
+ const int max_attempts = 5 * HOUR_TIME / 100; /* maximum 5 hours wait time
*/
+
+ PG_TRY();
+ {
+ while (attempts < max_attempts) {
+ pid_t pid;
+
+ status = GetBackgroundWorkerPid(handle, &pid);
+ if (status == BGWH_STOPPED) {
+ return status;
+ }
+
+ /* Add 100ms timeout instead of infinite wait */
+ rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT |
WL_POSTMASTER_DEATH, 100L, WAIT_EVENT_BGWORKER_SHUTDOWN);
+
+ ResetLatch(&MyProc->procLatch);
+
+ if (rc & WL_POSTMASTER_DEATH) {
+ status = BGWH_POSTMASTER_DIED;
+ break;
+ }
+
+ /* Check for interrupts but don't let them break the entire
process */
+ if (QueryCancelPending || ProcDiePending) {
+ ereport(WARNING,
(errmsg("WaitForBackgroundWorkerShutdownSafely: received interrupt signal,
stopping wait")));
+ status = BGWH_POSTMASTER_DIED; /* Return status as if
postmaster died */
+ break;
+ }
+
+ attempts++;
+ }
+
+ /* If maximum attempts reached */
+ if (attempts >= max_attempts) {
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
timeout after %d attempts", max_attempts)));
+ status = BGWH_POSTMASTER_DIED; /* Return error status */
+ }
+ }
+ PG_CATCH();
+ {
+ /* Log error but do NOT re-throw exception */
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
caught exception, returning error status")));
+ /* Return error status instead of PG_RE_THROW() */
+ return BGWH_POSTMASTER_DIED;
+ }
+ PG_END_TRY();
+ return status;
+}
+
+/*
+ * Retrieve list of database OIDs from the catalog.
+ *
+ * This function queries pg_database to get all user databases (excluding
+ * system databases like template0, template1, diskquota, and gpperfmon).
+ *
+ * Parameters:
+ * databases_cnt - Output parameter, set to number of databases found
+ * ctx - Memory context to allocate result in (for cross-call persistence)
+ * create_transaction - Whether to create a new transaction for the query
+ *
+ * Returns:
+ * Array of OIDs allocated in ctx, or NULL on error.
+ * The array length is databases_cnt.
+ *
+ * Note: Caller is responsible for freeing the returned memory when done.
+ */
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction) {
+ const char *sql =
+ "SELECT oid"
+ " FROM pg_database"
+ " WHERE datname NOT IN ('template0', 'template1', 'diskquota',
'gpperfmon')";
+ const char *error = NULL;
+
+ Oid *databases_oids = NULL;
+ *databases_cnt = 0;
+
+ if (create_transaction) {
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+ if (create_transaction) {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_execute(sql, true, 0) != SPI_OK_SELECT) {
+ error = "get_databases_oids: SPI_execute failed (select datname, oid)";
+ goto finish_spi;
+ }
+
+ /* Prepare tuple processing variables */
+
+ *databases_cnt = SPI_processed;
+ MemoryContext old_context = MemoryContextSwitchTo(ctx);
+ databases_oids = palloc((*databases_cnt) * sizeof(*databases_oids));
+ MemoryContextSwitchTo(old_context);
+
+ for (int i = 0; i < SPI_processed; ++i) {
+ Datum oid_datum;
+ bool oid_nullable;
+
+ heap_deform_tuple(SPI_tuptable->vals[i], SPI_tuptable->tupdesc,
&oid_datum, &oid_nullable);
+
+ databases_oids[i] = DatumGetObjectId(oid_datum);
+ }
+
+finish_spi:
+ SPI_finish();
+finish_transaction:
+ if (create_transaction) {
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+ pgstat_report_activity(STATE_IDLE, NULL);
+ }
+
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ return NULL; /* Return NULL on error */
+ }
+
+ return databases_oids;
+}
+
+/*
+ * Update the segment_file_map table with current relation file mappings.
+ *
+ * This function refreshes the mapping between relation OIDs and their
+ * physical file nodes across all segments. It first deletes the existing
+ * data and then repopulates it by querying pg_class on all segments.
+ *
+ * The mapping is essential for correlating file statistics collected
+ * from the filesystem with actual database relations.
+ *
+ * Returns:
+ * 0 on success, negative value on error
+ *
+ * Note: This function assumes it's running within an active SPI context.
+ */
+static int update_segment_file_map_table() {
+ int retcode = 0;
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_map";
+ char *sql_insert = "INSERT INTO relsizes_stats_schema.segment_file_map
SELECT gp_segment_id, oid, relfilenode FROM "
+ "gp_dist_random('pg_class')";
+ char *error = NULL;
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "update_segment_file_map_table: failed to delete from table";
+ goto cleanup;
+ }
+
+ pgstat_report_activity(STATE_RUNNING, sql_insert);
+ retcode = SPI_execute(sql_insert, false, 0);
+ if (retcode != SPI_OK_INSERT) {
+ error = "update_segment_file_map_table: failed to insert new rows into
table";
+ goto cleanup;
+ }
+
+cleanup:
+ pgstat_report_activity(STATE_IDLE, NULL);
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ }
+ return retcode;
+}
+
+/*
+ * Check if a character is a digit (0-9).
+ *
+ * Simple utility function used by fill_relfilenode() to parse
+ * numeric portions of filenames.
+ *
+ * Returns:
+ * true if character is a digit, false otherwise
+ */
+static bool is_number(char symbol) { return '0' <= symbol && symbol <= '9'; }
+
+/*
+ * Extract relfilenode from filename by finding the first sequence of digits
+ * in the filename and converting it to numeric value
+ */
+static unsigned int fill_relfilenode(char *name) {
+ unsigned int result = 0, pos = 0;
+ size_t name_len = strlen(name);
+
+ while (pos < name_len && !is_number(name[pos])) {
+ ++pos;
+ }
+ while (pos < name_len && is_number(name[pos])) {
+ /* Check for overflow to prevent integer overflow */
+ if (result > (UINT_MAX - (name[pos] - '0')) / 10) {
+ break; /* Stop on potential overflow */
+ }
+ result = (result * 10 + (name[pos] - '0'));
+ ++pos;
+ }
+ return result;
+}
+
+/*
+ * Background worker entry point for database-specific statistics collection.
+ *
+ * This function is executed by dynamically spawned background workers to
+ * collect file size statistics for a specific database. Each worker:
+ * 1. Connects to the target database
+ * 2. Verifies the extension is installed
+ * 3. Updates the segment file mapping
+ * 4. Collects file size statistics from all segments
+ * 5. Updates the historical statistics table
+ *
+ * The function runs within its own transaction and handles errors gracefully
+ * by logging warnings rather than aborting the entire collection process.
+ *
+ * Parameters:
+ * args - Background worker argument which contains database OID and the flag
+ * which indicates make pauses or not
+ *
+ * Note: This function is called via the background worker framework and
+ * should not be called directly.
+ */
+void relsizes_database_stats_job(Datum args) {
+ int retcode = 0;
+ char *error = NULL;
+ DbWorkerArg wa = { .d = args };
+
+ optimizer = false;
+ pqsignal(SIGTERM, worker_sigterm);
+ BackgroundWorkerUnblockSignals();
+
+ BackgroundWorkerInitializeConnectionByOid(wa.s.db, InvalidOid, 0);
+
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+
+ retcode = SPI_connect();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: SPI_connect failed";
+ goto finish_transaction;
+ }
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /* Verify extension is installed */
+ int created = plugin_created();
+ if (created < 0) {
+ error = "relsizes_database_stats_job: SPI execute failed while looking
for plugin";
+ goto finish_spi;
+ } else if (created == 0) {
+ goto finish_spi;
+ }
+
+ retcode = update_segment_file_map_table();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating segment_file_map
failed";
+ goto finish_spi;
+ }
+
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_sizes";
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "relsizes_database_stats_job: SPI_execute failed (delete from
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ /* Remove this condition after decision how to upgrade extensions is made.
*/
+ if (SearchSysCacheExists3(PROCNAMEARGSNSP,
+ CStringGetDatum("get_stats_for_database"),
+ PointerGetDatum(buildoidvector((Oid[]){INT4OID}, 1)),
+ ObjectIdGetDatum(get_namespace_oid("relsizes_stats_schema",
true))))
+ {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 1,
+ (Oid[]){INT4OID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId)},
+ NULL, false, 0);
+ } else {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1,
$2)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 2,
+ (Oid[]){OIDOID, BOOLOID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId),
BoolGetDatum(wa.s.fast)},
+ NULL, false, 0);
+ }
+ if (retcode != SPI_OK_INSERT) {
+ error = "relsizes_database_stats_job: SPI_execute failed (insert into
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ retcode = update_table_sizes_history();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating tables sizes history
table failed";
+ goto finish_spi;
+ }
+
+finish_spi:
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ /* Don't abort execution, continue with cleanup */
+ }
+ SPI_finish();
+finish_transaction:
+ PopActiveSnapshot();
Review Comment:
`PopActiveSnapshot()` is unconditional in `finish_transaction`, but
`PushActiveSnapshot()` is skipped when `SPI_connect()` fails. This can trigger
an unbalanced snapshot pop and crash the worker. Guard `PopActiveSnapshot()`
behind a flag set only after the snapshot is pushed.
##########
gpcontrib/gp_relsizes_stats/README.md:
##########
@@ -0,0 +1,52 @@
+# gp_relsizes_stats: Table sizes monitoring tool for Cloudberry
+
+### Features
+gp_relsizes_stats is an extension for the Cloudberry database that calculates
and stores statistics on the size of files and tables, occupied space on the
disks of the master and segment hosts.
+
+#### Features include
+- BackgroundWorker support for collecting statistics automatically
+- the ability to fine-tune the timeout values between actions, for example,
between launches for different databases, or during file processing to
distribute the load over time
+
+### Supported versions and platforms
+At the moment, the program is being tested only for Cloudberry and Linux.
+
+### Installation
+Install from source:
+```
+git clone [email protected]:open-gpdb/gp_relsizes_stats.git
+cd gp_relsizes_stats
+# Build it. Building would require GP installed nearby and sourcing
greenplum_path.sh
+source <path_to_gp>/greenplum_path.sh
+make && make install
+```
+
+### Confguration
Review Comment:
Section header typo: "Confguration" should be "Configuration".
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
+/* Additional headers for extension functionality */
+#include "access/xact.h"
+#include "executor/spi.h"
+#include "fmgr.h"
+#include "lib/stringinfo.h"
+#include "pgstat.h"
+#include "tcop/utility.h"
+
+#include "catalog/namespace.h"
+#include "cdb/cdbvars.h"
+#include "commands/defrem.h"
+#include "funcapi.h"
+
+#include "utils/builtins.h"
+#include "utils/datum.h"
+#include "utils/guc.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+
+#include <assert.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+#include <limits.h>
+
+#define FILEINFO_ARGS_CNT 5
+#define HOUR_TIME 3600000 /* milliseconds in hour */
+#define MINUTE_TIME 60000 /* milliseconds in minute */
+#define FILE_NAPTIME 1 /* default naptime between file processing in
milliseconds */
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(get_stats_for_database);
+PG_FUNCTION_INFO_V1(relsizes_collect_stats_once);
+Datum get_stats_for_database(PG_FUNCTION_ARGS);
+Datum relsizes_collect_stats_once(PG_FUNCTION_ARGS);
+
+static void worker_sigterm(SIGNAL_ARGS);
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction);
+static int update_segment_file_map_table(void);
+static int update_table_sizes_history(void);
+static void get_stats_for_databases(Oid *databases_oids, int databases_cnt,
bool fast);
+static void run_database_stats_worker(bool fast, Oid db);
+static int plugin_created(void);
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle);
+static int delete_data_in_history(void);
+static int put_data_into_history(void);
+void _PG_init(void);
+
+void relsizes_collect_stats(Datum main_arg);
+void relsizes_database_stats_job(Datum args);
+
+/* Global variables */
+static int worker_restart_naptime = 0;
+static int worker_database_naptime = 0;
+static int worker_file_naptime = 0;
+static bool enabled = false;
+
+static volatile sig_atomic_t got_sigterm = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+static_assert(sizeof(Datum) == sizeof(DbWorkerArg), "Invalid size of structure
in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM in background worker processes.
+ *
+ * This handler is called when the postmaster requests the background worker
+ * to shut down. It sets the got_sigterm flag and wakes up the main worker
+ * loop by setting the process latch.
+ *
+ * The function follows PostgreSQL signal handling conventions:
+ * - Saves and restores errno
+ * - Uses only async-signal-safe operations
+ * - Sets a flag that the main loop can check
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Wait for a background worker to stop with timeout and error handling.
+ *
+ * This is a modified version that adds timeout functionality and improved
+ * error handling to prevent infinite loops in case of hung workers.
+ * Returns BGWH_STOPPED on success, BGWH_POSTMASTER_DIED on error/timeout.
+ */
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle) {
+ BgwHandleStatus status = BGWH_NOT_YET_STARTED;
+ int rc;
+ int attempts = 0;
+ const int max_attempts = 5 * HOUR_TIME / 100; /* maximum 5 hours wait time
*/
+
+ PG_TRY();
+ {
+ while (attempts < max_attempts) {
+ pid_t pid;
+
+ status = GetBackgroundWorkerPid(handle, &pid);
+ if (status == BGWH_STOPPED) {
+ return status;
+ }
+
+ /* Add 100ms timeout instead of infinite wait */
+ rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT |
WL_POSTMASTER_DEATH, 100L, WAIT_EVENT_BGWORKER_SHUTDOWN);
+
+ ResetLatch(&MyProc->procLatch);
+
+ if (rc & WL_POSTMASTER_DEATH) {
+ status = BGWH_POSTMASTER_DIED;
+ break;
+ }
+
+ /* Check for interrupts but don't let them break the entire
process */
+ if (QueryCancelPending || ProcDiePending) {
+ ereport(WARNING,
(errmsg("WaitForBackgroundWorkerShutdownSafely: received interrupt signal,
stopping wait")));
+ status = BGWH_POSTMASTER_DIED; /* Return status as if
postmaster died */
+ break;
+ }
+
+ attempts++;
+ }
+
+ /* If maximum attempts reached */
+ if (attempts >= max_attempts) {
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
timeout after %d attempts", max_attempts)));
+ status = BGWH_POSTMASTER_DIED; /* Return error status */
+ }
+ }
+ PG_CATCH();
+ {
+ /* Log error but do NOT re-throw exception */
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
caught exception, returning error status")));
+ /* Return error status instead of PG_RE_THROW() */
+ return BGWH_POSTMASTER_DIED;
+ }
+ PG_END_TRY();
+ return status;
+}
+
+/*
+ * Retrieve list of database OIDs from the catalog.
+ *
+ * This function queries pg_database to get all user databases (excluding
+ * system databases like template0, template1, diskquota, and gpperfmon).
+ *
+ * Parameters:
+ * databases_cnt - Output parameter, set to number of databases found
+ * ctx - Memory context to allocate result in (for cross-call persistence)
+ * create_transaction - Whether to create a new transaction for the query
+ *
+ * Returns:
+ * Array of OIDs allocated in ctx, or NULL on error.
+ * The array length is databases_cnt.
+ *
+ * Note: Caller is responsible for freeing the returned memory when done.
+ */
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction) {
+ const char *sql =
+ "SELECT oid"
+ " FROM pg_database"
+ " WHERE datname NOT IN ('template0', 'template1', 'diskquota',
'gpperfmon')";
+ const char *error = NULL;
+
+ Oid *databases_oids = NULL;
+ *databases_cnt = 0;
+
+ if (create_transaction) {
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+ if (create_transaction) {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_execute(sql, true, 0) != SPI_OK_SELECT) {
+ error = "get_databases_oids: SPI_execute failed (select datname, oid)";
+ goto finish_spi;
+ }
+
+ /* Prepare tuple processing variables */
+
+ *databases_cnt = SPI_processed;
+ MemoryContext old_context = MemoryContextSwitchTo(ctx);
+ databases_oids = palloc((*databases_cnt) * sizeof(*databases_oids));
+ MemoryContextSwitchTo(old_context);
+
+ for (int i = 0; i < SPI_processed; ++i) {
+ Datum oid_datum;
+ bool oid_nullable;
+
+ heap_deform_tuple(SPI_tuptable->vals[i], SPI_tuptable->tupdesc,
&oid_datum, &oid_nullable);
+
+ databases_oids[i] = DatumGetObjectId(oid_datum);
+ }
+
+finish_spi:
+ SPI_finish();
+finish_transaction:
+ if (create_transaction) {
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+ pgstat_report_activity(STATE_IDLE, NULL);
+ }
+
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ return NULL; /* Return NULL on error */
+ }
+
+ return databases_oids;
+}
+
+/*
+ * Update the segment_file_map table with current relation file mappings.
+ *
+ * This function refreshes the mapping between relation OIDs and their
+ * physical file nodes across all segments. It first deletes the existing
+ * data and then repopulates it by querying pg_class on all segments.
+ *
+ * The mapping is essential for correlating file statistics collected
+ * from the filesystem with actual database relations.
+ *
+ * Returns:
+ * 0 on success, negative value on error
+ *
+ * Note: This function assumes it's running within an active SPI context.
+ */
+static int update_segment_file_map_table() {
+ int retcode = 0;
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_map";
+ char *sql_insert = "INSERT INTO relsizes_stats_schema.segment_file_map
SELECT gp_segment_id, oid, relfilenode FROM "
+ "gp_dist_random('pg_class')";
+ char *error = NULL;
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "update_segment_file_map_table: failed to delete from table";
+ goto cleanup;
+ }
+
+ pgstat_report_activity(STATE_RUNNING, sql_insert);
+ retcode = SPI_execute(sql_insert, false, 0);
+ if (retcode != SPI_OK_INSERT) {
+ error = "update_segment_file_map_table: failed to insert new rows into
table";
+ goto cleanup;
+ }
+
+cleanup:
+ pgstat_report_activity(STATE_IDLE, NULL);
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ }
+ return retcode;
+}
+
+/*
+ * Check if a character is a digit (0-9).
+ *
+ * Simple utility function used by fill_relfilenode() to parse
+ * numeric portions of filenames.
+ *
+ * Returns:
+ * true if character is a digit, false otherwise
+ */
+static bool is_number(char symbol) { return '0' <= symbol && symbol <= '9'; }
+
+/*
+ * Extract relfilenode from filename by finding the first sequence of digits
+ * in the filename and converting it to numeric value
+ */
+static unsigned int fill_relfilenode(char *name) {
+ unsigned int result = 0, pos = 0;
+ size_t name_len = strlen(name);
+
+ while (pos < name_len && !is_number(name[pos])) {
+ ++pos;
+ }
+ while (pos < name_len && is_number(name[pos])) {
+ /* Check for overflow to prevent integer overflow */
+ if (result > (UINT_MAX - (name[pos] - '0')) / 10) {
+ break; /* Stop on potential overflow */
+ }
+ result = (result * 10 + (name[pos] - '0'));
+ ++pos;
+ }
+ return result;
+}
+
+/*
+ * Background worker entry point for database-specific statistics collection.
+ *
+ * This function is executed by dynamically spawned background workers to
+ * collect file size statistics for a specific database. Each worker:
+ * 1. Connects to the target database
+ * 2. Verifies the extension is installed
+ * 3. Updates the segment file mapping
+ * 4. Collects file size statistics from all segments
+ * 5. Updates the historical statistics table
+ *
+ * The function runs within its own transaction and handles errors gracefully
+ * by logging warnings rather than aborting the entire collection process.
+ *
+ * Parameters:
+ * args - Background worker argument which contains database OID and the flag
+ * which indicates make pauses or not
+ *
+ * Note: This function is called via the background worker framework and
+ * should not be called directly.
+ */
+void relsizes_database_stats_job(Datum args) {
+ int retcode = 0;
+ char *error = NULL;
+ DbWorkerArg wa = { .d = args };
+
+ optimizer = false;
+ pqsignal(SIGTERM, worker_sigterm);
+ BackgroundWorkerUnblockSignals();
+
+ BackgroundWorkerInitializeConnectionByOid(wa.s.db, InvalidOid, 0);
+
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+
+ retcode = SPI_connect();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: SPI_connect failed";
+ goto finish_transaction;
+ }
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /* Verify extension is installed */
+ int created = plugin_created();
+ if (created < 0) {
+ error = "relsizes_database_stats_job: SPI execute failed while looking
for plugin";
+ goto finish_spi;
+ } else if (created == 0) {
+ goto finish_spi;
+ }
+
+ retcode = update_segment_file_map_table();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating segment_file_map
failed";
+ goto finish_spi;
+ }
+
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_sizes";
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "relsizes_database_stats_job: SPI_execute failed (delete from
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ /* Remove this condition after decision how to upgrade extensions is made.
*/
+ if (SearchSysCacheExists3(PROCNAMEARGSNSP,
+ CStringGetDatum("get_stats_for_database"),
+ PointerGetDatum(buildoidvector((Oid[]){INT4OID}, 1)),
+ ObjectIdGetDatum(get_namespace_oid("relsizes_stats_schema",
true))))
+ {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 1,
+ (Oid[]){INT4OID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId)},
+ NULL, false, 0);
+ } else {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1,
$2)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 2,
+ (Oid[]){OIDOID, BOOLOID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId),
BoolGetDatum(wa.s.fast)},
+ NULL, false, 0);
+ }
+ if (retcode != SPI_OK_INSERT) {
+ error = "relsizes_database_stats_job: SPI_execute failed (insert into
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ retcode = update_table_sizes_history();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating tables sizes history
table failed";
+ goto finish_spi;
+ }
+
+finish_spi:
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ /* Don't abort execution, continue with cleanup */
+ }
+ SPI_finish();
+finish_transaction:
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+ pgstat_report_activity(STATE_IDLE, NULL);
+}
+
+/*
+ * Spawn and manage a background worker for database statistics collection.
+ *
+ * This function creates a new background worker to collect statistics for
+ * a specific database.
+ *
+ * The function:
+ * 1. Configures a new background worker with appropriate settings
+ * 2. Registers and starts the worker
+ * 3. Waits for the worker to complete
+ * 4. Handles any errors during worker execution
+ *
+ * If the worker fails to start or encounters errors during execution,
+ * warnings are logged but the function returns normally to allow
+ * processing of remaining databases.
+ *
+ * Parameters:
+ * fast - Don't make pauses
+ * db - OID of the database which worker will collect statistics from
+ *
+ * Note: This function may take significant time to complete as it waits
+ * for the background worker to finish processing the entire database.
+ */
+static void run_database_stats_worker(bool fast, Oid db) {
+ bool ret;
+ MemoryContext old_ctx;
+ BackgroundWorkerHandle *handle;
+ BgwHandleStatus status;
+
+ /* Configure background worker */
+ BackgroundWorker database_worker = {
+ .bgw_flags = BGWORKER_SHMEM_ACCESS |
BGWORKER_BACKEND_DATABASE_CONNECTION,
+ .bgw_start_time = BgWorkerStart_RecoveryFinished,
+ .bgw_restart_time = BGW_NEVER_RESTART,
+ .bgw_library_name = "gp_relsizes_stats",
+ .bgw_function_name = "relsizes_database_stats_job",
+ .bgw_notify_pid = MyProcPid,
+ .bgw_main_arg = ((DbWorkerArg){ .s.db = db, .s.fast = fast }).d,
+ .bgw_start_rule = NULL,
+ };
+ snprintf(database_worker.bgw_name, BGW_MAXLEN,
"database_relsizes_collector_worker for %u", db);
+ old_ctx = MemoryContextSwitchTo(TopMemoryContext);
+ ret = RegisterDynamicBackgroundWorker(&database_worker, &handle);
+ MemoryContextSwitchTo(old_ctx);
+ if (!ret) {
+ ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES), errmsg("could
not register background process"),
+ errhint("You may need to increase
max_worker_processes.")));
+ }
+ pid_t pid;
+ status = WaitForBackgroundWorkerStartup(handle, &pid);
+ if (status == BGWH_STOPPED)
+ return;
+ if (status != BGWH_STARTED) {
+ ereport(WARNING, (errmsg("Failed to start background worker [%s],
skipping", database_worker.bgw_name)));
+ return;
+ }
+ status = WaitForBackgroundWorkerShutdownSafely(handle);
+ if (status != BGWH_STOPPED) {
+ ereport(WARNING, (errmsg("Failure during background worker execution
[%s], continuing", database_worker.bgw_name)));
+ /* Don't abort execution, just log and continue */
+ }
+}
+
+/*
+ * SQL-callable function to collect file statistics for a database.
+ *
+ * This function scans the filesystem directory corresponding to a database
+ * and returns statistics for all regular files found. It's designed to run
+ * on individual segments to collect local file information.
+ *
+ * The function:
+ * 1. Validates the function call context (must support returning a set)
+ * 2. Sets up a tuplestore for result collection
+ * 3. Scans the database directory (base/<dboid>/)
+ * 4. For each regular file, extracts relfilenode from filename
+ * 5. Collects file size and modification time via lstat()
+ * 6. Returns results as a set of tuples
+ *
+ * Parameters:
+ * Database OID (oid) - identifies which database directory to scan
+ * Fast (bool) - When true, don't sleep between each collect-phase for files
+ *
+ * Returns:
+ * Set of tuples containing:
+ * - segment: current segment ID
+ * - relfilenode: extracted from filename
+ * - filepath: full path to the file
+ * - size: file size in bytes
+ * - mtime: modification time as Unix timestamp
+ *
+ * Note: Includes configurable delays between file processing to reduce I/O
load
+ */
+Datum get_stats_for_database(PG_FUNCTION_ARGS) {
+ int segment_id = GpIdentity.segindex;
+ Oid dboid = PG_GETARG_OID(0);
+ bool fast = (PG_NARGS() < 2) ? false : PG_GETARG_BOOL(1);
+
+ char cwd[PATH_MAX];
+ char *data_dir = NULL;
+ char *error = NULL;
+ char *file_path = NULL;
+
+ if (getcwd(cwd, sizeof(cwd)) == NULL) {
+ error = "get_stats_for_database: failed to get current working
directory";
+ goto finish_data;
+ }
+ data_dir = psprintf("%s/base/%u", cwd, dboid);
+ ReturnSetInfo *rsinfo = (ReturnSetInfo *)fcinfo->resultinfo;
+ /* Validate function call context */
+ if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo)) {
+ error = "get_stats_for_database: set-valued function called in context
that cannot accept a set";
+ goto finish_data;
+ }
+ if (!(rsinfo->allowedModes & SFRM_Materialize)) {
+ error = "get_stats_for_database: materialize mode required, but it is
not allowed in this context";
+ goto finish_data;
+ }
+
+ /* Setup output tuple store */
+ MemoryContext oldcontext =
MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
+ TupleDesc tupdesc;
+ if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) {
+ MemoryContextSwitchTo(oldcontext);
+ error = "get_stats_for_database: incorrect return type in fcinfo (must
be a row type)";
+ goto finish_data;
+ }
+ tupdesc = BlessTupleDesc(tupdesc);
+
+ bool randomAccess = (rsinfo->allowedModes & SFRM_Materialize_Random) != 0;
+ Tuplestorestate *tupstore = tuplestore_begin_heap(randomAccess, false,
work_mem);
+
+ rsinfo->returnMode = SFRM_Materialize;
+ rsinfo->setResult = tupstore;
+ rsinfo->setDesc = tupdesc;
+
+ Datum outputValues[FILEINFO_ARGS_CNT];
+ bool outputNulls[FILEINFO_ARGS_CNT] = { false };
+
+ MemoryContextSwitchTo(oldcontext);
+
+ /* Scan database directory for files */
+ DIR *current_dir = AllocateDir(data_dir);
+ if (!current_dir) {
+ error = "get_stats_for_database: failed to allocate current directory";
+ goto finish_data;
+ }
+
+ struct dirent *file;
+ while ((file = ReadDir(current_dir, data_dir)) != NULL) {
+ char *filename = file->d_name;
+ if (strcmp(filename, ".") == 0 || strcmp(filename, "..") == 0) {
+ continue;
+ }
+
+ file_path = psprintf("%s/%s", data_dir, filename);
+ struct stat stb;
+ if (lstat(file_path, &stb) < 0) {
+ ereport(WARNING,
+ (errmsg("get_stats_for_database: lstat failed with %s file
(unexpected behavior)", file_path)));
+ pfree(file_path);
+ continue;
+ }
+
+ if (S_ISREG(stb.st_mode)) {
+ /* Process regular files and collect size statistics */
+ outputValues[0] = Int32GetDatum(segment_id);
+ outputValues[1] = ObjectIdGetDatum(fill_relfilenode(filename));
+ outputValues[2] = CStringGetTextDatum(file_path);
+ outputValues[3] = Int64GetDatum(stb.st_size);
+ outputValues[4] = Int64GetDatum(stb.st_mtime);
+
+ tuplestore_putvalues(tupstore, tupdesc, outputValues, outputNulls);
+
Review Comment:
`fill_relfilenode()` returns 0 when the filename contains no digits (e.g.
files like `PG_VERSION`, `pg_filenode.map`, etc.). The loop currently adds
*all* regular files to `segment_file_sizes`, which can bloat the table with
rows that will never join to `segment_file_map`. Consider skipping files that
don’t start with a digit / don’t contain a relfilenode, or at least skipping
entries where `fill_relfilenode()` returns 0.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
+/* Additional headers for extension functionality */
+#include "access/xact.h"
+#include "executor/spi.h"
+#include "fmgr.h"
+#include "lib/stringinfo.h"
+#include "pgstat.h"
+#include "tcop/utility.h"
+
+#include "catalog/namespace.h"
+#include "cdb/cdbvars.h"
+#include "commands/defrem.h"
+#include "funcapi.h"
+
+#include "utils/builtins.h"
+#include "utils/datum.h"
+#include "utils/guc.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+
+#include <assert.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+#include <limits.h>
+
+#define FILEINFO_ARGS_CNT 5
+#define HOUR_TIME 3600000 /* milliseconds in hour */
+#define MINUTE_TIME 60000 /* milliseconds in minute */
+#define FILE_NAPTIME 1 /* default naptime between file processing in
milliseconds */
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(get_stats_for_database);
+PG_FUNCTION_INFO_V1(relsizes_collect_stats_once);
+Datum get_stats_for_database(PG_FUNCTION_ARGS);
+Datum relsizes_collect_stats_once(PG_FUNCTION_ARGS);
+
+static void worker_sigterm(SIGNAL_ARGS);
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction);
+static int update_segment_file_map_table(void);
+static int update_table_sizes_history(void);
+static void get_stats_for_databases(Oid *databases_oids, int databases_cnt,
bool fast);
+static void run_database_stats_worker(bool fast, Oid db);
+static int plugin_created(void);
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle);
+static int delete_data_in_history(void);
+static int put_data_into_history(void);
+void _PG_init(void);
+
+void relsizes_collect_stats(Datum main_arg);
+void relsizes_database_stats_job(Datum args);
+
+/* Global variables */
+static int worker_restart_naptime = 0;
+static int worker_database_naptime = 0;
+static int worker_file_naptime = 0;
+static bool enabled = false;
+
+static volatile sig_atomic_t got_sigterm = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+static_assert(sizeof(Datum) == sizeof(DbWorkerArg), "Invalid size of structure
in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM in background worker processes.
+ *
+ * This handler is called when the postmaster requests the background worker
+ * to shut down. It sets the got_sigterm flag and wakes up the main worker
+ * loop by setting the process latch.
+ *
+ * The function follows PostgreSQL signal handling conventions:
+ * - Saves and restores errno
+ * - Uses only async-signal-safe operations
+ * - Sets a flag that the main loop can check
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Wait for a background worker to stop with timeout and error handling.
+ *
+ * This is a modified version that adds timeout functionality and improved
+ * error handling to prevent infinite loops in case of hung workers.
+ * Returns BGWH_STOPPED on success, BGWH_POSTMASTER_DIED on error/timeout.
+ */
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle) {
+ BgwHandleStatus status = BGWH_NOT_YET_STARTED;
+ int rc;
+ int attempts = 0;
+ const int max_attempts = 5 * HOUR_TIME / 100; /* maximum 5 hours wait time
*/
+
+ PG_TRY();
+ {
+ while (attempts < max_attempts) {
+ pid_t pid;
+
+ status = GetBackgroundWorkerPid(handle, &pid);
+ if (status == BGWH_STOPPED) {
+ return status;
+ }
+
+ /* Add 100ms timeout instead of infinite wait */
+ rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT |
WL_POSTMASTER_DEATH, 100L, WAIT_EVENT_BGWORKER_SHUTDOWN);
+
+ ResetLatch(&MyProc->procLatch);
+
+ if (rc & WL_POSTMASTER_DEATH) {
+ status = BGWH_POSTMASTER_DIED;
+ break;
+ }
+
+ /* Check for interrupts but don't let them break the entire
process */
+ if (QueryCancelPending || ProcDiePending) {
+ ereport(WARNING,
(errmsg("WaitForBackgroundWorkerShutdownSafely: received interrupt signal,
stopping wait")));
+ status = BGWH_POSTMASTER_DIED; /* Return status as if
postmaster died */
+ break;
+ }
+
+ attempts++;
+ }
+
+ /* If maximum attempts reached */
+ if (attempts >= max_attempts) {
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
timeout after %d attempts", max_attempts)));
+ status = BGWH_POSTMASTER_DIED; /* Return error status */
+ }
+ }
+ PG_CATCH();
+ {
+ /* Log error but do NOT re-throw exception */
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
caught exception, returning error status")));
+ /* Return error status instead of PG_RE_THROW() */
+ return BGWH_POSTMASTER_DIED;
+ }
+ PG_END_TRY();
+ return status;
+}
+
+/*
+ * Retrieve list of database OIDs from the catalog.
+ *
+ * This function queries pg_database to get all user databases (excluding
+ * system databases like template0, template1, diskquota, and gpperfmon).
+ *
+ * Parameters:
+ * databases_cnt - Output parameter, set to number of databases found
+ * ctx - Memory context to allocate result in (for cross-call persistence)
+ * create_transaction - Whether to create a new transaction for the query
+ *
+ * Returns:
+ * Array of OIDs allocated in ctx, or NULL on error.
+ * The array length is databases_cnt.
+ *
+ * Note: Caller is responsible for freeing the returned memory when done.
+ */
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction) {
+ const char *sql =
+ "SELECT oid"
+ " FROM pg_database"
+ " WHERE datname NOT IN ('template0', 'template1', 'diskquota',
'gpperfmon')";
+ const char *error = NULL;
+
+ Oid *databases_oids = NULL;
+ *databases_cnt = 0;
+
+ if (create_transaction) {
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+ if (create_transaction) {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_execute(sql, true, 0) != SPI_OK_SELECT) {
+ error = "get_databases_oids: SPI_execute failed (select datname, oid)";
+ goto finish_spi;
+ }
+
+ /* Prepare tuple processing variables */
+
+ *databases_cnt = SPI_processed;
+ MemoryContext old_context = MemoryContextSwitchTo(ctx);
+ databases_oids = palloc((*databases_cnt) * sizeof(*databases_oids));
+ MemoryContextSwitchTo(old_context);
+
+ for (int i = 0; i < SPI_processed; ++i) {
+ Datum oid_datum;
+ bool oid_nullable;
+
+ heap_deform_tuple(SPI_tuptable->vals[i], SPI_tuptable->tupdesc,
&oid_datum, &oid_nullable);
+
+ databases_oids[i] = DatumGetObjectId(oid_datum);
+ }
+
+finish_spi:
+ SPI_finish();
+finish_transaction:
+ if (create_transaction) {
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+ pgstat_report_activity(STATE_IDLE, NULL);
+ }
+
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ return NULL; /* Return NULL on error */
+ }
+
+ return databases_oids;
+}
+
+/*
+ * Update the segment_file_map table with current relation file mappings.
+ *
+ * This function refreshes the mapping between relation OIDs and their
+ * physical file nodes across all segments. It first deletes the existing
+ * data and then repopulates it by querying pg_class on all segments.
+ *
+ * The mapping is essential for correlating file statistics collected
+ * from the filesystem with actual database relations.
+ *
+ * Returns:
+ * 0 on success, negative value on error
+ *
+ * Note: This function assumes it's running within an active SPI context.
+ */
+static int update_segment_file_map_table() {
+ int retcode = 0;
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_map";
+ char *sql_insert = "INSERT INTO relsizes_stats_schema.segment_file_map
SELECT gp_segment_id, oid, relfilenode FROM "
+ "gp_dist_random('pg_class')";
+ char *error = NULL;
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "update_segment_file_map_table: failed to delete from table";
+ goto cleanup;
+ }
+
+ pgstat_report_activity(STATE_RUNNING, sql_insert);
+ retcode = SPI_execute(sql_insert, false, 0);
+ if (retcode != SPI_OK_INSERT) {
+ error = "update_segment_file_map_table: failed to insert new rows into
table";
+ goto cleanup;
+ }
+
+cleanup:
+ pgstat_report_activity(STATE_IDLE, NULL);
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ }
+ return retcode;
+}
+
+/*
+ * Check if a character is a digit (0-9).
+ *
+ * Simple utility function used by fill_relfilenode() to parse
+ * numeric portions of filenames.
+ *
+ * Returns:
+ * true if character is a digit, false otherwise
+ */
+static bool is_number(char symbol) { return '0' <= symbol && symbol <= '9'; }
+
+/*
+ * Extract relfilenode from filename by finding the first sequence of digits
+ * in the filename and converting it to numeric value
+ */
+static unsigned int fill_relfilenode(char *name) {
+ unsigned int result = 0, pos = 0;
+ size_t name_len = strlen(name);
+
+ while (pos < name_len && !is_number(name[pos])) {
+ ++pos;
+ }
+ while (pos < name_len && is_number(name[pos])) {
+ /* Check for overflow to prevent integer overflow */
+ if (result > (UINT_MAX - (name[pos] - '0')) / 10) {
+ break; /* Stop on potential overflow */
+ }
+ result = (result * 10 + (name[pos] - '0'));
+ ++pos;
+ }
+ return result;
+}
+
+/*
+ * Background worker entry point for database-specific statistics collection.
+ *
+ * This function is executed by dynamically spawned background workers to
+ * collect file size statistics for a specific database. Each worker:
+ * 1. Connects to the target database
+ * 2. Verifies the extension is installed
+ * 3. Updates the segment file mapping
+ * 4. Collects file size statistics from all segments
+ * 5. Updates the historical statistics table
+ *
+ * The function runs within its own transaction and handles errors gracefully
+ * by logging warnings rather than aborting the entire collection process.
+ *
+ * Parameters:
+ * args - Background worker argument which contains database OID and the flag
+ * which indicates make pauses or not
+ *
+ * Note: This function is called via the background worker framework and
+ * should not be called directly.
+ */
+void relsizes_database_stats_job(Datum args) {
+ int retcode = 0;
+ char *error = NULL;
+ DbWorkerArg wa = { .d = args };
+
+ optimizer = false;
+ pqsignal(SIGTERM, worker_sigterm);
+ BackgroundWorkerUnblockSignals();
+
+ BackgroundWorkerInitializeConnectionByOid(wa.s.db, InvalidOid, 0);
+
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+
+ retcode = SPI_connect();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: SPI_connect failed";
+ goto finish_transaction;
+ }
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /* Verify extension is installed */
+ int created = plugin_created();
+ if (created < 0) {
+ error = "relsizes_database_stats_job: SPI execute failed while looking
for plugin";
+ goto finish_spi;
+ } else if (created == 0) {
+ goto finish_spi;
+ }
+
+ retcode = update_segment_file_map_table();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating segment_file_map
failed";
+ goto finish_spi;
+ }
+
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_sizes";
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "relsizes_database_stats_job: SPI_execute failed (delete from
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ /* Remove this condition after decision how to upgrade extensions is made.
*/
+ if (SearchSysCacheExists3(PROCNAMEARGSNSP,
+ CStringGetDatum("get_stats_for_database"),
+ PointerGetDatum(buildoidvector((Oid[]){INT4OID}, 1)),
+ ObjectIdGetDatum(get_namespace_oid("relsizes_stats_schema",
true))))
+ {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 1,
+ (Oid[]){INT4OID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId)},
Review Comment:
In the legacy-signature branch, `SPI_execute_with_args` declares the
argument type as `INT4OID` but passes `ObjectIdGetDatum(MyDatabaseId)`. That
Datum encoding does not match `INT4OID`. Either change the declared type to
`OIDOID` or pass `Int32GetDatum(MyDatabaseId)` (and consider an explicit cast
in SQL) to avoid misinterpretation on some platforms.
##########
gpcontrib/gp_relsizes_stats/README.md:
##########
@@ -0,0 +1,52 @@
+# gp_relsizes_stats: Table sizes monitoring tool for Cloudberry
+
+### Features
+gp_relsizes_stats is an extension for the Cloudberry database that calculates
and stores statistics on the size of files and tables, occupied space on the
disks of the master and segment hosts.
+
+#### Features include
+- BackgroundWorker support for collecting statistics automatically
+- the ability to fine-tune the timeout values between actions, for example,
between launches for different databases, or during file processing to
distribute the load over time
+
+### Supported versions and platforms
+At the moment, the program is being tested only for Cloudberry and Linux.
+
+### Installation
+Install from source:
+```
+git clone [email protected]:open-gpdb/gp_relsizes_stats.git
+cd gp_relsizes_stats
+# Build it. Building would require GP installed nearby and sourcing
greenplum_path.sh
+source <path_to_gp>/greenplum_path.sh
+make && make install
Review Comment:
The installation instructions still describe cloning the standalone upstream
repo and sourcing `greenplum_path.sh`. Since this extension is now in-tree
under `gpcontrib/`, the README should describe the Cloudberry build/install
flow (e.g., building from the monorepo, or `make -C gpcontrib/gp_relsizes_stats
install`, and the correct environment script if needed).
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,934 @@
+#include "postgres.h"
+
+/* Required headers for background workers */
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
+#include "storage/lwlock.h"
+#include "storage/proc.h"
+#include "storage/shmem.h"
+
+/* Additional headers for extension functionality */
+#include "access/xact.h"
+#include "executor/spi.h"
+#include "fmgr.h"
+#include "lib/stringinfo.h"
+#include "pgstat.h"
+#include "tcop/utility.h"
+
+#include "catalog/namespace.h"
+#include "cdb/cdbvars.h"
+#include "commands/defrem.h"
+#include "funcapi.h"
+
+#include "utils/builtins.h"
+#include "utils/datum.h"
+#include "utils/guc.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+
+#include <assert.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+#include <limits.h>
+
+#define FILEINFO_ARGS_CNT 5
+#define HOUR_TIME 3600000 /* milliseconds in hour */
+#define MINUTE_TIME 60000 /* milliseconds in minute */
+#define FILE_NAPTIME 1 /* default naptime between file processing in
milliseconds */
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(get_stats_for_database);
+PG_FUNCTION_INFO_V1(relsizes_collect_stats_once);
+Datum get_stats_for_database(PG_FUNCTION_ARGS);
+Datum relsizes_collect_stats_once(PG_FUNCTION_ARGS);
+
+static void worker_sigterm(SIGNAL_ARGS);
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction);
+static int update_segment_file_map_table(void);
+static int update_table_sizes_history(void);
+static void get_stats_for_databases(Oid *databases_oids, int databases_cnt,
bool fast);
+static void run_database_stats_worker(bool fast, Oid db);
+static int plugin_created(void);
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle);
+static int delete_data_in_history(void);
+static int put_data_into_history(void);
+void _PG_init(void);
+
+void relsizes_collect_stats(Datum main_arg);
+void relsizes_database_stats_job(Datum args);
+
+/* Global variables */
+static int worker_restart_naptime = 0;
+static int worker_database_naptime = 0;
+static int worker_file_naptime = 0;
+static bool enabled = false;
+
+static volatile sig_atomic_t got_sigterm = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+static_assert(sizeof(Datum) == sizeof(DbWorkerArg), "Invalid size of structure
in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM in background worker processes.
+ *
+ * This handler is called when the postmaster requests the background worker
+ * to shut down. It sets the got_sigterm flag and wakes up the main worker
+ * loop by setting the process latch.
+ *
+ * The function follows PostgreSQL signal handling conventions:
+ * - Saves and restores errno
+ * - Uses only async-signal-safe operations
+ * - Sets a flag that the main loop can check
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Wait for a background worker to stop with timeout and error handling.
+ *
+ * This is a modified version that adds timeout functionality and improved
+ * error handling to prevent infinite loops in case of hung workers.
+ * Returns BGWH_STOPPED on success, BGWH_POSTMASTER_DIED on error/timeout.
+ */
+static BgwHandleStatus
WaitForBackgroundWorkerShutdownSafely(BackgroundWorkerHandle *handle) {
+ BgwHandleStatus status = BGWH_NOT_YET_STARTED;
+ int rc;
+ int attempts = 0;
+ const int max_attempts = 5 * HOUR_TIME / 100; /* maximum 5 hours wait time
*/
+
+ PG_TRY();
+ {
+ while (attempts < max_attempts) {
+ pid_t pid;
+
+ status = GetBackgroundWorkerPid(handle, &pid);
+ if (status == BGWH_STOPPED) {
+ return status;
+ }
+
+ /* Add 100ms timeout instead of infinite wait */
+ rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT |
WL_POSTMASTER_DEATH, 100L, WAIT_EVENT_BGWORKER_SHUTDOWN);
+
+ ResetLatch(&MyProc->procLatch);
+
+ if (rc & WL_POSTMASTER_DEATH) {
+ status = BGWH_POSTMASTER_DIED;
+ break;
+ }
+
+ /* Check for interrupts but don't let them break the entire
process */
+ if (QueryCancelPending || ProcDiePending) {
+ ereport(WARNING,
(errmsg("WaitForBackgroundWorkerShutdownSafely: received interrupt signal,
stopping wait")));
+ status = BGWH_POSTMASTER_DIED; /* Return status as if
postmaster died */
+ break;
+ }
+
+ attempts++;
+ }
+
+ /* If maximum attempts reached */
+ if (attempts >= max_attempts) {
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
timeout after %d attempts", max_attempts)));
+ status = BGWH_POSTMASTER_DIED; /* Return error status */
+ }
+ }
+ PG_CATCH();
+ {
+ /* Log error but do NOT re-throw exception */
+ ereport(WARNING, (errmsg("WaitForBackgroundWorkerShutdownSafely:
caught exception, returning error status")));
+ /* Return error status instead of PG_RE_THROW() */
+ return BGWH_POSTMASTER_DIED;
+ }
+ PG_END_TRY();
+ return status;
+}
+
+/*
+ * Retrieve list of database OIDs from the catalog.
+ *
+ * This function queries pg_database to get all user databases (excluding
+ * system databases like template0, template1, diskquota, and gpperfmon).
+ *
+ * Parameters:
+ * databases_cnt - Output parameter, set to number of databases found
+ * ctx - Memory context to allocate result in (for cross-call persistence)
+ * create_transaction - Whether to create a new transaction for the query
+ *
+ * Returns:
+ * Array of OIDs allocated in ctx, or NULL on error.
+ * The array length is databases_cnt.
+ *
+ * Note: Caller is responsible for freeing the returned memory when done.
+ */
+static Oid *get_databases_oids(int *databases_cnt, MemoryContext ctx, bool
create_transaction) {
+ const char *sql =
+ "SELECT oid"
+ " FROM pg_database"
+ " WHERE datname NOT IN ('template0', 'template1', 'diskquota',
'gpperfmon')";
+ const char *error = NULL;
+
+ Oid *databases_oids = NULL;
+ *databases_cnt = 0;
+
+ if (create_transaction) {
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+ if (create_transaction) {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_execute(sql, true, 0) != SPI_OK_SELECT) {
+ error = "get_databases_oids: SPI_execute failed (select datname, oid)";
+ goto finish_spi;
+ }
+
+ /* Prepare tuple processing variables */
+
+ *databases_cnt = SPI_processed;
+ MemoryContext old_context = MemoryContextSwitchTo(ctx);
+ databases_oids = palloc((*databases_cnt) * sizeof(*databases_oids));
+ MemoryContextSwitchTo(old_context);
+
+ for (int i = 0; i < SPI_processed; ++i) {
+ Datum oid_datum;
+ bool oid_nullable;
+
+ heap_deform_tuple(SPI_tuptable->vals[i], SPI_tuptable->tupdesc,
&oid_datum, &oid_nullable);
+
+ databases_oids[i] = DatumGetObjectId(oid_datum);
+ }
+
+finish_spi:
+ SPI_finish();
+finish_transaction:
+ if (create_transaction) {
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+ pgstat_report_activity(STATE_IDLE, NULL);
+ }
+
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ return NULL; /* Return NULL on error */
+ }
+
+ return databases_oids;
+}
+
+/*
+ * Update the segment_file_map table with current relation file mappings.
+ *
+ * This function refreshes the mapping between relation OIDs and their
+ * physical file nodes across all segments. It first deletes the existing
+ * data and then repopulates it by querying pg_class on all segments.
+ *
+ * The mapping is essential for correlating file statistics collected
+ * from the filesystem with actual database relations.
+ *
+ * Returns:
+ * 0 on success, negative value on error
+ *
+ * Note: This function assumes it's running within an active SPI context.
+ */
+static int update_segment_file_map_table() {
+ int retcode = 0;
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_map";
+ char *sql_insert = "INSERT INTO relsizes_stats_schema.segment_file_map
SELECT gp_segment_id, oid, relfilenode FROM "
+ "gp_dist_random('pg_class')";
+ char *error = NULL;
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "update_segment_file_map_table: failed to delete from table";
+ goto cleanup;
+ }
+
+ pgstat_report_activity(STATE_RUNNING, sql_insert);
+ retcode = SPI_execute(sql_insert, false, 0);
+ if (retcode != SPI_OK_INSERT) {
+ error = "update_segment_file_map_table: failed to insert new rows into
table";
+ goto cleanup;
+ }
+
+cleanup:
+ pgstat_report_activity(STATE_IDLE, NULL);
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ }
+ return retcode;
+}
+
+/*
+ * Check if a character is a digit (0-9).
+ *
+ * Simple utility function used by fill_relfilenode() to parse
+ * numeric portions of filenames.
+ *
+ * Returns:
+ * true if character is a digit, false otherwise
+ */
+static bool is_number(char symbol) { return '0' <= symbol && symbol <= '9'; }
+
+/*
+ * Extract relfilenode from filename by finding the first sequence of digits
+ * in the filename and converting it to numeric value
+ */
+static unsigned int fill_relfilenode(char *name) {
+ unsigned int result = 0, pos = 0;
+ size_t name_len = strlen(name);
+
+ while (pos < name_len && !is_number(name[pos])) {
+ ++pos;
+ }
+ while (pos < name_len && is_number(name[pos])) {
+ /* Check for overflow to prevent integer overflow */
+ if (result > (UINT_MAX - (name[pos] - '0')) / 10) {
+ break; /* Stop on potential overflow */
+ }
+ result = (result * 10 + (name[pos] - '0'));
+ ++pos;
+ }
+ return result;
+}
+
+/*
+ * Background worker entry point for database-specific statistics collection.
+ *
+ * This function is executed by dynamically spawned background workers to
+ * collect file size statistics for a specific database. Each worker:
+ * 1. Connects to the target database
+ * 2. Verifies the extension is installed
+ * 3. Updates the segment file mapping
+ * 4. Collects file size statistics from all segments
+ * 5. Updates the historical statistics table
+ *
+ * The function runs within its own transaction and handles errors gracefully
+ * by logging warnings rather than aborting the entire collection process.
+ *
+ * Parameters:
+ * args - Background worker argument which contains database OID and the flag
+ * which indicates make pauses or not
+ *
+ * Note: This function is called via the background worker framework and
+ * should not be called directly.
+ */
+void relsizes_database_stats_job(Datum args) {
+ int retcode = 0;
+ char *error = NULL;
+ DbWorkerArg wa = { .d = args };
+
+ optimizer = false;
+ pqsignal(SIGTERM, worker_sigterm);
+ BackgroundWorkerUnblockSignals();
+
+ BackgroundWorkerInitializeConnectionByOid(wa.s.db, InvalidOid, 0);
+
+ SetCurrentStatementStartTimestamp();
+ StartTransactionCommand();
+
+ retcode = SPI_connect();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: SPI_connect failed";
+ goto finish_transaction;
+ }
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /* Verify extension is installed */
+ int created = plugin_created();
+ if (created < 0) {
+ error = "relsizes_database_stats_job: SPI execute failed while looking
for plugin";
+ goto finish_spi;
+ } else if (created == 0) {
+ goto finish_spi;
+ }
+
+ retcode = update_segment_file_map_table();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating segment_file_map
failed";
+ goto finish_spi;
+ }
+
+ char *sql_delete = "DELETE FROM relsizes_stats_schema.segment_file_sizes";
+ pgstat_report_activity(STATE_RUNNING, sql_delete);
+ retcode = SPI_execute(sql_delete, false, 0);
+ if (retcode != SPI_OK_DELETE) {
+ error = "relsizes_database_stats_job: SPI_execute failed (delete from
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ /* Remove this condition after decision how to upgrade extensions is made.
*/
+ if (SearchSysCacheExists3(PROCNAMEARGSNSP,
+ CStringGetDatum("get_stats_for_database"),
+ PointerGetDatum(buildoidvector((Oid[]){INT4OID}, 1)),
+ ObjectIdGetDatum(get_namespace_oid("relsizes_stats_schema",
true))))
+ {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 1,
+ (Oid[]){INT4OID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId)},
+ NULL, false, 0);
+ } else {
+ const char* sql_get_stats =
+ "INSERT INTO relsizes_stats_schema.segment_file_sizes (segment,
relfilenode, filepath, size, mtime) "
+ "SELECT * FROM relsizes_stats_schema.get_stats_for_database($1,
$2)";
+ pgstat_report_activity(STATE_RUNNING, sql_get_stats);
+ retcode = SPI_execute_with_args(sql_get_stats, 2,
+ (Oid[]){OIDOID, BOOLOID},
+ (Datum[]){ObjectIdGetDatum(MyDatabaseId),
BoolGetDatum(wa.s.fast)},
+ NULL, false, 0);
+ }
+ if (retcode != SPI_OK_INSERT) {
+ error = "relsizes_database_stats_job: SPI_execute failed (insert into
segment_file_sizes)";
+ goto finish_spi;
+ }
+
+ retcode = update_table_sizes_history();
+ if (retcode < 0) {
+ error = "relsizes_database_stats_job: updating tables sizes history
table failed";
+ goto finish_spi;
+ }
+
+finish_spi:
+ if (error != NULL) {
+ ereport(WARNING, (errmsg("%s: %m", error)));
+ /* Don't abort execution, continue with cleanup */
+ }
+ SPI_finish();
+finish_transaction:
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+ pgstat_report_activity(STATE_IDLE, NULL);
+}
+
+/*
+ * Spawn and manage a background worker for database statistics collection.
+ *
+ * This function creates a new background worker to collect statistics for
+ * a specific database.
+ *
+ * The function:
+ * 1. Configures a new background worker with appropriate settings
+ * 2. Registers and starts the worker
+ * 3. Waits for the worker to complete
+ * 4. Handles any errors during worker execution
+ *
+ * If the worker fails to start or encounters errors during execution,
+ * warnings are logged but the function returns normally to allow
+ * processing of remaining databases.
+ *
+ * Parameters:
+ * fast - Don't make pauses
+ * db - OID of the database which worker will collect statistics from
+ *
+ * Note: This function may take significant time to complete as it waits
+ * for the background worker to finish processing the entire database.
+ */
+static void run_database_stats_worker(bool fast, Oid db) {
+ bool ret;
+ MemoryContext old_ctx;
+ BackgroundWorkerHandle *handle;
+ BgwHandleStatus status;
+
+ /* Configure background worker */
+ BackgroundWorker database_worker = {
+ .bgw_flags = BGWORKER_SHMEM_ACCESS |
BGWORKER_BACKEND_DATABASE_CONNECTION,
+ .bgw_start_time = BgWorkerStart_RecoveryFinished,
+ .bgw_restart_time = BGW_NEVER_RESTART,
+ .bgw_library_name = "gp_relsizes_stats",
+ .bgw_function_name = "relsizes_database_stats_job",
+ .bgw_notify_pid = MyProcPid,
+ .bgw_main_arg = ((DbWorkerArg){ .s.db = db, .s.fast = fast }).d,
+ .bgw_start_rule = NULL,
+ };
+ snprintf(database_worker.bgw_name, BGW_MAXLEN,
"database_relsizes_collector_worker for %u", db);
+ old_ctx = MemoryContextSwitchTo(TopMemoryContext);
+ ret = RegisterDynamicBackgroundWorker(&database_worker, &handle);
+ MemoryContextSwitchTo(old_ctx);
+ if (!ret) {
+ ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES), errmsg("could
not register background process"),
+ errhint("You may need to increase
max_worker_processes.")));
+ }
+ pid_t pid;
+ status = WaitForBackgroundWorkerStartup(handle, &pid);
+ if (status == BGWH_STOPPED)
+ return;
+ if (status != BGWH_STARTED) {
+ ereport(WARNING, (errmsg("Failed to start background worker [%s],
skipping", database_worker.bgw_name)));
+ return;
+ }
+ status = WaitForBackgroundWorkerShutdownSafely(handle);
+ if (status != BGWH_STOPPED) {
+ ereport(WARNING, (errmsg("Failure during background worker execution
[%s], continuing", database_worker.bgw_name)));
+ /* Don't abort execution, just log and continue */
+ }
+}
+
+/*
+ * SQL-callable function to collect file statistics for a database.
+ *
+ * This function scans the filesystem directory corresponding to a database
+ * and returns statistics for all regular files found. It's designed to run
+ * on individual segments to collect local file information.
+ *
+ * The function:
+ * 1. Validates the function call context (must support returning a set)
+ * 2. Sets up a tuplestore for result collection
+ * 3. Scans the database directory (base/<dboid>/)
+ * 4. For each regular file, extracts relfilenode from filename
+ * 5. Collects file size and modification time via lstat()
+ * 6. Returns results as a set of tuples
+ *
+ * Parameters:
+ * Database OID (oid) - identifies which database directory to scan
+ * Fast (bool) - When true, don't sleep between each collect-phase for files
+ *
+ * Returns:
+ * Set of tuples containing:
+ * - segment: current segment ID
+ * - relfilenode: extracted from filename
+ * - filepath: full path to the file
+ * - size: file size in bytes
+ * - mtime: modification time as Unix timestamp
+ *
+ * Note: Includes configurable delays between file processing to reduce I/O
load
+ */
+Datum get_stats_for_database(PG_FUNCTION_ARGS) {
+ int segment_id = GpIdentity.segindex;
+ Oid dboid = PG_GETARG_OID(0);
+ bool fast = (PG_NARGS() < 2) ? false : PG_GETARG_BOOL(1);
+
+ char cwd[PATH_MAX];
+ char *data_dir = NULL;
+ char *error = NULL;
+ char *file_path = NULL;
+
+ if (getcwd(cwd, sizeof(cwd)) == NULL) {
+ error = "get_stats_for_database: failed to get current working
directory";
+ goto finish_data;
+ }
+ data_dir = psprintf("%s/base/%u", cwd, dboid);
+ ReturnSetInfo *rsinfo = (ReturnSetInfo *)fcinfo->resultinfo;
Review Comment:
Building the scan path as `getcwd() + "/base/<dboid>"` is brittle (cwd isn’t
guaranteed to be `DataDir`) and it only covers the default tablespace.
Meanwhile the SQL views include relations in non-global tablespaces, so sizes
for user tablespaces will be missing/undercounted. Consider using
`GetDatabasePath(dboid, DEFAULTTABLESPACE_OID)` (and/or iterating `pg_tblspc`)
and anchoring paths off `DataDir` instead of `getcwd()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
