Copilot commented on code in PR #1936:
URL: https://github.com/apache/cloudberry/pull/1936#discussion_r3873589041
##########
contrib/interconnect/README.md:
##########
@@ -1,6 +1,6 @@
# Intercontect
-This subtree contains interconnect module && test && benchmark that different
with other subtree inside {cbdb_src}/contrib. Other moudles are not part of the
core CloudBerry system, but interconnect module split from `cdb module`, it
**must be preload with CloudBerry**, otherwise CloudBerry system will not work
properly.
+This subtree contains interconnect module && test && benchmark that different
with other subtree inside {cbdb_src}/contrib. Other moudles are not part of the
core Cloudberry system, but interconnect module split from `cdb module`, it
**must be preload with Cloudberry**, otherwise Cloudberry system will not work
properly.
Review Comment:
This edited line still contains typos/grammar issues ('moudles', 'must be
preload'). Since the line is being touched in this PR, please fix them (e.g.,
'modules', 'must be preloaded') to avoid perpetuating documentation errors.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,1046 @@
+/*-------------------------------------------------------------------------
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * gp_relsizes_stats.c
+ *
+ * IDENTIFICATION
+ * gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#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 void worker_sighup(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);
+
+PGDLLEXPORT void relsizes_collect_stats(Datum main_arg);
+PGDLLEXPORT 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 bool save_history = true;
+
+static volatile sig_atomic_t got_sigterm = false;
+static volatile sig_atomic_t got_sighup = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+StaticAssertDecl(sizeof(Datum) == sizeof(DbWorkerArg),
+ "Invalid size of structure in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM
+ * Set a flag to let the main loop to terminate, and set our latch
to wake
+ * it up.
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Signal handler for SIGHUP
+ * Set a flag to tell the main loop to reread the config file, and
set
+ * our latch to wake it up.
+ */
+static void worker_sighup(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sighup = 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();
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+
+ 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_isnull;
+
+ oid_datum = SPI_getbinval(SPI_tuptable->vals[i],
SPI_tuptable->tupdesc, 1, &oid_isnull);
+
+ 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.
+ */
+PGDLLEXPORT void relsizes_database_stats_job(Datum args) {
+ int retcode = 0;
+ char *error = NULL;
+ DbWorkerArg wa = { .d = args };
+
+ optimizer = false;
+ pqsignal(SIGTERM, worker_sigterm);
+ pqsignal(SIGHUP, worker_sighup);
+ BackgroundWorkerUnblockSignals();
+
+ BackgroundWorkerInitializeConnectionByOid(wa.s.db, InvalidOid, 0);
+
+ if (IS_QUERY_DISPATCHER() && !IS_SINGLENODE())
+ Gp_role = GP_ROLE_DISPATCH;
+
+ 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[]){Int32GetDatum((int32) 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[]){Int32GetDatum((int32) MyDatabaseId),
BoolGetDatum(wa.s.fast)},
Review Comment:
The argument is declared as type OIDOID, but the code passes
`Int32GetDatum((int32) MyDatabaseId)`. This should use the Oid/Datum conversion
macro for OIDs (e.g., `ObjectIdGetDatum(MyDatabaseId)`) to avoid
type/representation mismatches and make the intent correct.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,1046 @@
+/*-------------------------------------------------------------------------
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * gp_relsizes_stats.c
+ *
+ * IDENTIFICATION
+ * gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#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 void worker_sighup(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);
+
+PGDLLEXPORT void relsizes_collect_stats(Datum main_arg);
+PGDLLEXPORT 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 bool save_history = true;
+
+static volatile sig_atomic_t got_sigterm = false;
+static volatile sig_atomic_t got_sighup = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+StaticAssertDecl(sizeof(Datum) == sizeof(DbWorkerArg),
+ "Invalid size of structure in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM
+ * Set a flag to let the main loop to terminate, and set our latch
to wake
+ * it up.
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Signal handler for SIGHUP
+ * Set a flag to tell the main loop to reread the config file, and
set
+ * our latch to wake it up.
+ */
+static void worker_sighup(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sighup = 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();
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+
+ 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_isnull;
+
+ oid_datum = SPI_getbinval(SPI_tuptable->vals[i],
SPI_tuptable->tupdesc, 1, &oid_isnull);
+
+ 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() {
Review Comment:
`update_segment_file_map_table()` currently returns raw SPI status codes
(e.g., `SPI_OK_DELETE`, `SPI_OK_INSERT`) rather than `0` on success as the
comment claims. Please update the doc comment to match the real return contract
(or normalize return values) to prevent future callers from misusing it.
##########
gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c:
##########
@@ -0,0 +1,1046 @@
+/*-------------------------------------------------------------------------
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * gp_relsizes_stats.c
+ *
+ * IDENTIFICATION
+ * gpcontrib/gp_relsizes_stats/src/gp_relsizes_stats.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#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 void worker_sighup(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);
+
+PGDLLEXPORT void relsizes_collect_stats(Datum main_arg);
+PGDLLEXPORT 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 bool save_history = true;
+
+static volatile sig_atomic_t got_sigterm = false;
+static volatile sig_atomic_t got_sighup = false;
+
+typedef union DbWorkerArg {
+ Datum d;
+ struct {
+ Oid db;
+ bool fast;
+ } s;
+} DbWorkerArg;
+
+StaticAssertDecl(sizeof(Datum) == sizeof(DbWorkerArg),
+ "Invalid size of structure in DbWorkerArg");
+
+/*
+ * Signal handler for SIGTERM
+ * Set a flag to let the main loop to terminate, and set our latch
to wake
+ * it up.
+ */
+static void worker_sigterm(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sigterm = true;
+ if (MyProc) {
+ SetLatch(&MyProc->procLatch);
+ }
+ errno = save_errno;
+}
+
+/*
+ * Signal handler for SIGHUP
+ * Set a flag to tell the main loop to reread the config file, and
set
+ * our latch to wake it up.
+ */
+static void worker_sighup(SIGNAL_ARGS) {
+ int save_errno = errno;
+ got_sighup = 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();
+ PushActiveSnapshot(GetTransactionSnapshot());
+ pgstat_report_activity(STATE_RUNNING, sql);
+ }
+
+ if (SPI_connect() < 0) {
+ error = "get_databases_oids: SPI_connect failed";
+ goto finish_transaction;
+ }
+
+ 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_isnull;
+
+ oid_datum = SPI_getbinval(SPI_tuptable->vals[i],
SPI_tuptable->tupdesc, 1, &oid_isnull);
+
+ 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)));
Review Comment:
Several WARNING paths format messages as `\"%s: %m\"`, but `%m` is for
`errno`-based system errors and is often meaningless for SPI failures (and can
produce misleading output). Consider emitting SPI-related details explicitly
(or drop `: %m`), and reserve `%m` only for code paths that are directly
reporting a syscall failure that sets `errno`.
##########
.github/workflows/build-deb-cloudberry.yml:
##########
@@ -354,8 +375,13 @@ jobs:
outputs:
build_timestamp: ${{ steps.set_timestamp.outputs.timestamp }}
+ strategy:
+ fail-fast: false
+ matrix:
+ ubuntu_version: ['22.04', '24.04']
+
Review Comment:
This workflow adds a `workflow_dispatch` input for `ubuntu_version` and
computes `UBUNTU_VERSIONS_JSON` in `prepare-test-matrix-deb`, but `build-deb`
(and other jobs with their own matrices) still hardcode `['22.04','24.04']`.
Consider driving the Ubuntu version matrix from the input-derived list to avoid
surprising behavior where selecting a single version still runs both.
##########
.github/workflows/build-deb-cloudberry.yml:
##########
@@ -849,9 +881,9 @@ jobs:
} 2>&1 | tee -a install-logs/details/deb-installation.log
- name: Upload install logs
- uses: actions/upload-artifact@v4
+ uses: actions/upload-artifact@v7
with:
- name: install-logs-${{ matrix.name }}-${{
needs.build-deb.outputs.build_timestamp }}
+ name: install-logs-ubuntu${{ matrix.ubuntu_version }}-${{
needs.build-deb.outputs.build_timestamp }}
Review Comment:
`build-deb` is now a matrix job, but downstream jobs reference a single
`needs.build-deb.outputs.build_timestamp`. Job outputs are not reliably
consumable from matrix jobs as a single scalar, which can lead to
empty/incorrect artifact names. Prefer generating timestamps per job (or
embedding `${{ github.run_id }}` / `${{ github.run_attempt }}`), or restructure
so a non-matrix job produces a single shared timestamp output.
--
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]
