From 2f0096bbf4e96b29ba51215fc295f351c6b104c5 Mon Sep 17 00:00:00 2001 From: Xiao__Ma Date: Thu, 29 Jul 2021 19:20:49 +0800 Subject: [PATCH] clean --- src/bin/Makefile | 1 - src/bin/gs_guc/cluster_guc.conf | 1 + src/bin/pgxc_clean/pgxc_clean.cpp | 45 +- src/common/backend/utils/init/miscinit.cpp | 20 + src/common/backend/utils/init/postinit.cpp | 11 + src/common/backend/utils/misc/guc.cpp | 15 +- .../process/postmaster/twophasecleaner.cpp | 504 +++++++++++++++++- 7 files changed, 541 insertions(+), 56 deletions(-) diff --git a/src/bin/Makefile b/src/bin/Makefile index 3561373b7..fc0e256d0 100644 --- a/src/bin/Makefile +++ b/src/bin/Makefile @@ -52,7 +52,6 @@ SUBDIRS = \ pg_controldata \ pg_resetxlog \ gs_guc \ - pgxc_clean \ gsqlerr \ pg_basebackup \ pg_probackup diff --git a/src/bin/gs_guc/cluster_guc.conf b/src/bin/gs_guc/cluster_guc.conf index 071e338e2..4105a2d0b 100755 --- a/src/bin/gs_guc/cluster_guc.conf +++ b/src/bin/gs_guc/cluster_guc.conf @@ -296,6 +296,7 @@ gtm_port4|int|1,65535|NULL|NULL| gtm_port5|int|1,65535|NULL|NULL| gtm_port6|int|1,65535|NULL|NULL| gtm_port7|int|1,65535|NULL|NULL| +gs_clean_timeout|int|0,2147483|s|NULL| hashagg_table_size|int|0,1073741823|NULL|NULL| hba_file|string|0,0|NULL|NULL| hot_standby|bool|0,0|NULL|When hot_standby set to on, wal_level must be set to hot_standby. Otherwise it will cause the database can not be started. In the dual-system environments, hot_standby can not be set to off.| diff --git a/src/bin/pgxc_clean/pgxc_clean.cpp b/src/bin/pgxc_clean/pgxc_clean.cpp index 4ee2fa392..8efc7364b 100644 --- a/src/bin/pgxc_clean/pgxc_clean.cpp +++ b/src/bin/pgxc_clean/pgxc_clean.cpp @@ -145,16 +145,12 @@ static void usage(void); static void showVersion(void); static PGconn* loginDatabase(char* host, int port, char* user, char* password, char* dbname, const char* progname, char* encoding, const char* password_prompt); -#ifdef ENABLE_MULTIPLE_NODES static void getMyNodename(PGconn* conn); static void getMyGtmMode(PGconn* conn); -#endif static void recover2PCForDatabase(database_info* db_info, CleanWorkerInfo* wkinfo); static void recover2PC(PGconn* conn, txn_info* txn, int num); static void getDatabaseList(PGconn* conn); -#ifdef ENABLE_MULTIPLE_NODES static void getNodeList(PGconn* conn); -#endif static void getPreparedTxnList(PGconn* conn); static void getTxnInfoOnOtherNodesAll(CleanWorkerInfo* wkinfo); static void do_commit(PGconn* conn, txn_info* txn, int num); @@ -170,10 +166,8 @@ static void dropTempSchemas(PGconn* conn, bool missing_ok); static void dropTempSchema(PGconn* conn, char* nspname, bool missing_ok); static void getTempSchemaListOnCN(PGconn* conn); -#ifdef ENABLE_MULTIPLE_NODES static void getTempSchemaOnOneDN(PGconn* conn, int no); static int getTempSchemaListOnDN(PGconn* conn); -#endif static void checkAllocMem(void* p); static void cleanTempSchemaList(); static void cleanBackendList(); @@ -670,7 +664,6 @@ int main(int argc, char* argv[]) exit(1); } -#ifdef ENABLE_MULTIPLE_NODES /* * Get my nodename (connected Coordinator) and gtm_mode */ @@ -684,7 +677,6 @@ int main(int argc, char* argv[]) transfor_gtm_optoin(gtm_option, gtm_mode), gtm_option_num); } -#endif /* * Get available databases @@ -712,7 +704,6 @@ int main(int argc, char* argv[]) } } -#ifdef ENABLE_MULTIPLE_NODES /* * Get list of Nodes * @@ -737,7 +728,6 @@ int main(int argc, char* argv[]) pgxc_clean_node_info[ii].type == NODE_TYPE_COORD ? "coordinator" : "datanode"); } } -#endif /* * Only clean temp schema if clean_temp_schema_only == true, @@ -751,15 +741,6 @@ int main(int argc, char* argv[]) } exit(0); } -#ifndef ENABLE_MULTIPLE_NODES - else { - PQfinish(coord_conn); - coord_conn = NULL; - write_stderr("%s %s: We only clean temp schema in openGauss, please specify '-e' option.\n", - formatLogTime(), progname); - exit(0); - } -#endif /* * Only clean plan_table if clean_plan_table_only == true, @@ -1133,9 +1114,7 @@ static void cleanPlanTableSessidList() static void dropTempNamespace() { database_info* cur_database = NULL; -#ifdef ENABLE_MULTIPLE_NODES int datanode_no = -1; -#endif if (head_database_info != NULL) { for (cur_database = head_database_info; cur_database != NULL; cur_database = cur_database->next) { @@ -1187,12 +1166,11 @@ static void dropTempNamespace() (void)dropTempSchemas(coord_conn, true); (void)cleanBackendList(); -#ifdef ENABLE_MULTIPLE_NODES datanode_no = getTempSchemaListOnDN(coord_conn); (void)getActiveBackendListOnCN(coord_conn, datanode_no); (void)dropTempSchemas(coord_conn, true); (void)cleanBackendList(); -#endif + if (verbose_opt) write_stderr("%s %s: drop temp namespace for database \"%s\" finished\n", @@ -1209,12 +1187,10 @@ static void dropTempSchemas(PGconn* conn, bool missing_ok) { tempschema_info* cell1 = temp_schema_info; activebackend_info* cell2 = NULL; + char node_name[NAMEDATALEN]; char temp_buffer[NAMEDATALEN] = {0}; char temp_buffer2[NAMEDATALEN] = {0}; -#ifdef ENABLE_MULTIPLE_NODES - char node_name[NAMEDATALEN]; bool onOtherCNNode = false; -#endif errno_t rc; if (temp_schema_info == NULL || active_backend_info == NULL) @@ -1254,7 +1230,6 @@ static void dropTempSchemas(PGconn* conn, bool missing_ok) } if (cell2 == NULL) { -#ifdef ENABLE_MULTIPLE_NODES /* * now it is not the active backend, furthermore, we need to * check whether the schema on the other node: @@ -1329,9 +1304,6 @@ static void dropTempSchemas(PGconn* conn, bool missing_ok) if (!onOtherCNNode) dropTempSchema(conn, cell1->tempschema_name, missing_ok); } -#else - dropTempSchema(conn, cell1->tempschema_name, missing_ok); -#endif } cell1 = cell1->next; @@ -1443,7 +1415,6 @@ static char* pg_strdup(const char* string) return tmp; } -#ifdef ENABLE_MULTIPLE_NODES static void getMyNodename(PGconn* conn) { static const char* stmt = "SELECT pgxc_node_str()"; @@ -1487,7 +1458,6 @@ static void getMyGtmMode(PGconn* conn) } PQclear(res); } -#endif static void recover2PCForDatabase(database_info* db_info, CleanWorkerInfo* wkinfo) { @@ -2352,7 +2322,6 @@ static void add_active_backend_info(int64 sessionID, uint32 tempID, uint32 timeL } } -#ifdef ENABLE_MULTIPLE_NODES /* * @Description: get temp schema list of a random DN. * @param[IN] conn: node connection handle @@ -2446,7 +2415,6 @@ static void getTempSchemaOnOneDN(PGconn* conn, int no) PQclear(res); } -#endif static void getTempSchemaListOnCN(PGconn* conn) { @@ -2458,18 +2426,11 @@ static void getTempSchemaListOnCN(PGconn* conn) char STMT_GET_TEMP_SCHEMA_LIST[NAMEDATALEN + 128] = {0}; -#ifdef ENABLE_MULTIPLE_NODES rc = snprintf_s(STMT_GET_TEMP_SCHEMA_LIST, sizeof(STMT_GET_TEMP_SCHEMA_LIST), sizeof(STMT_GET_TEMP_SCHEMA_LIST) - 1, "SELECT NSPNAME FROM PG_NAMESPACE WHERE NSPNAME LIKE 'pg_temp_%s_%%'", my_nodename); -#else - rc = snprintf_s(STMT_GET_TEMP_SCHEMA_LIST, - sizeof(STMT_GET_TEMP_SCHEMA_LIST), - sizeof(STMT_GET_TEMP_SCHEMA_LIST) - 1, - "SELECT NSPNAME FROM PG_NAMESPACE WHERE NSPNAME LIKE 'pg_temp_%%'"); -#endif securec_check_ss_c(rc, "\0", "\0"); if (verbose_opt) @@ -2589,7 +2550,6 @@ static void checkAllocMem(void* p) } } -#ifdef ENABLE_MULTIPLE_NODES static void getNodeList(PGconn* conn) { int ii; @@ -2672,7 +2632,6 @@ error_exit: PQclear(res); exit(1); } -#endif static char* transfor_gtm_optoin(const char* option, const char* gtm_free_mode) { diff --git a/src/common/backend/utils/init/miscinit.cpp b/src/common/backend/utils/init/miscinit.cpp index fa26cb2a6..2c9ca47cd 100644 --- a/src/common/backend/utils/init/miscinit.cpp +++ b/src/common/backend/utils/init/miscinit.cpp @@ -32,6 +32,7 @@ #include #include "catalog/pg_authid.h" +#include "commands/user.h" #include "job/job_scheduler.h" #include "job/job_worker.h" #include "mb/pg_wchar.h" @@ -790,6 +791,25 @@ void InitializeSessionUserId(const char* rolename) roleTup = SearchSysCache1(AUTHNAME, PointerGetDatum(rolename)); +#ifndef ENABLE_MULTIPLE_NODES + /* + * In opengauss, we allow twophasecleaner to connect to database + * as superusers for cleaning up temporary tables. + */ + if (!HeapTupleIsValid(roleTup) && u_sess->proc_cxt.IsInnerMaintenanceTools) { + roleTup = SearchSysCache1(AUTHOID, UInt32GetDatum(BOOTSTRAP_SUPERUSERID)); + + char userName[NAMEDATALEN]; + MemoryContext oldcontext = NULL; + oldcontext = MemoryContextSwitchTo(SESS_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_EXECUTOR)); + if (u_sess->proc_cxt.MyProcPort->user_name) + pfree_ext(u_sess->proc_cxt.MyProcPort->user_name); + u_sess->proc_cxt.MyProcPort->user_name = pstrdup((char*)GetSuperUserName((char*)userName)); + (void)MemoryContextSwitchTo(oldcontext); + rolename = u_sess->proc_cxt.MyProcPort->user_name; + } +#endif + if (!HeapTupleIsValid(roleTup)) { /* * Audit user login diff --git a/src/common/backend/utils/init/postinit.cpp b/src/common/backend/utils/init/postinit.cpp index 78fe07f28..eaac93742 100644 --- a/src/common/backend/utils/init/postinit.cpp +++ b/src/common/backend/utils/init/postinit.cpp @@ -1868,6 +1868,17 @@ void PostgresInitializer::InitUser() InitializeSessionUserId(m_username); m_isSuperUser = superuser(); u_sess->misc_cxt.CurrentUserName = u_sess->proc_cxt.MyProcPort->user_name; +#ifndef ENABLE_MULTIPLE_NODES + /* + * In opengauss, we allow twophasecleaner to connect to database as superusers + * for cleaning up temporary tables. During the cleanup of temporary tables, + * m_username and u_sess->proc_cxt.MyProcPort->user_name point to the same memory + * address. We have freed and reinitialized u_sess->proc_cxt.MyProcPort->user_name + * in function InitializeSessionUserId, and we need to initialize m_username here. + */ + if (u_sess->proc_cxt.IsInnerMaintenanceTools) + m_username = u_sess->proc_cxt.MyProcPort->user_name; +#endif } void PostgresInitializer::CheckConnPermission() diff --git a/src/common/backend/utils/misc/guc.cpp b/src/common/backend/utils/misc/guc.cpp index 68fcac28e..c15898025 100644 --- a/src/common/backend/utils/misc/guc.cpp +++ b/src/common/backend/utils/misc/guc.cpp @@ -6936,6 +6936,20 @@ static void InitConfigureNamesInt() NULL, NULL}, + {{"gs_clean_timeout", + PGC_SIGHUP, + STATS_MONITORING, + gettext_noop("Sets the timeout to call gs_clean."), + gettext_noop("A value of 0 turns off the timeout."), + GUC_UNIT_S}, + &u_sess->attr.attr_storage.gs_clean_timeout, + 60, + 0, + INT_MAX / 1000, + NULL, + NULL, + NULL}, + /* End-of-list marker */ {{NULL, (GucContext)0, (config_group)0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL}}; @@ -9566,7 +9580,6 @@ static void InitSingleNodeUnsupportGuc() u_sess->attr.attr_common.max_datanode_for_plan = 0; u_sess->attr.attr_common.transaction_sync_naptime = 30; u_sess->attr.attr_common.transaction_sync_timeout = 600; - u_sess->attr.attr_storage.gs_clean_timeout = 300; u_sess->attr.attr_storage.twophase_clean_workers = 3; /* for Double Guc Variables */ u_sess->attr.attr_sql.stream_multiple = DEFAULT_STREAM_MULTIPLE; diff --git a/src/gausskernel/process/postmaster/twophasecleaner.cpp b/src/gausskernel/process/postmaster/twophasecleaner.cpp index ec0597528..35ef23b3f 100644 --- a/src/gausskernel/process/postmaster/twophasecleaner.cpp +++ b/src/gausskernel/process/postmaster/twophasecleaner.cpp @@ -31,7 +31,9 @@ #include +#include "catalog/pg_database.h" #include "gssignal/gs_signal.h" +#include "libpq/libpq-fe.h" #include "libpq/pqsignal.h" #include "miscadmin.h" #include "pgstat.h" @@ -52,13 +54,38 @@ bool bSyncXactsCallGsclean = false; PGPROC* twoPhaseCleanerProc = NULL; +#ifndef ENABLE_MULTIPLE_NODES +typedef struct DatabaseNames { + struct DatabaseNames* next; + char* databaseName; +} DatabaseNames; + +typedef struct TempSchemaInfo { + struct TempSchemaInfo* next; + char* tempSchemaName; +} TempSchemaInfo; + +typedef struct ActiveBackendInfo { + struct ActiveBackendInfo* next; + int64 sessionID; + uint32 tempID; + uint32 timeLineID; +} ActiveBackendInfo; +#endif + /* Signal handlers */ static void TwoPCSigHupHandler(SIGNAL_ARGS); static void TwoPCShutdownHandler(SIGNAL_ARGS); +#ifdef ENABLE_MULTIPLE_NODES static int get_prog_path(const char* argv0); +#endif +#ifndef ENABLE_MULTIPLE_NODES +static bool DropTempNamespace(); +#endif NON_EXEC_STATIC void TwoPhaseCleanerMain() { + sigjmp_buf local_sigjmp_buf; MemoryContext twopc_context; bool clean_successed = false; int rc; @@ -91,6 +118,31 @@ NON_EXEC_STATIC void TwoPhaseCleanerMain() /* We allow SIGQUIT (quickdie) at all times */ (void)sigdelset(&t_thrd.libpq_cxt.BlockSig, SIGQUIT); + /* + * If an exception is encountered, processing resumes here. + * + * See notes in postgres.c about the design of this coding. + */ + int curTryCounter; + int* oldTryCounter = NULL; + if (sigsetjmp(local_sigjmp_buf, 1) != 0) { + gstrace_tryblock_exit(true, oldTryCounter); + /* Prevents interrupts while cleaning up */ + HOLD_INTERRUPTS(); + + /* Report the error to the server log */ + EmitErrorReport(); + + FlushErrorState(); + + /* Now we can allow interrupts again */ + RESUME_INTERRUPTS(); + } + oldTryCounter = gstrace_tryblock_entry(&curTryCounter); + + /* We can now handle ereport(ERROR) */ + t_thrd.log_cxt.PG_exception_stack = &local_sigjmp_buf; + /* * Create a memory context that we will do all our work in. We do this so * that we can reset the context during error recovery and thereby avoid @@ -132,12 +184,15 @@ NON_EXEC_STATIC void TwoPhaseCleanerMain() } /* - * Call gs_clean to clean the prepared transaction every - * gs_clean_timeout second. + * In distributed Gaussdb, we call gs_clean to clean the prepared + * transaction every gs_clean_timeout second. * Remark: After the process is started, wait 60s to call * gs_clean firstly, if successed, then make the clean_successed * to true, if failed, wait 60s to recall gs_clean until it * success. + * + * In opengauss, instead of calling gs_clean, we use twophasecleaner + * to clean up the temporary table every gs_clean_timeout second. */ if (bSyncXactsCallGsclean || (u_sess->attr.attr_storage.gs_clean_timeout && @@ -146,6 +201,7 @@ NON_EXEC_STATIC void TwoPhaseCleanerMain() TimestampDifferenceExceeds(gs_clean_start_time, gs_clean_current_time, u_sess->attr.attr_storage.gs_clean_timeout * 1000)))) { +#ifdef ENABLE_MULTIPLE_NODES int status = 0; char cmd[MAX_PATH_LEN]; @@ -154,7 +210,6 @@ NON_EXEC_STATIC void TwoPhaseCleanerMain() ereport(DEBUG5, (errmsg("failed to invoke get_prog_path()"))); } else { /* if we find explicit cn listen address, we use tcp connection instead of unix socket */ -#ifdef ENABLE_MULTIPLE_NODES #ifdef USE_ASSERT_CHECKING rc = sprintf_s(cmd, sizeof(cmd), @@ -170,14 +225,7 @@ NON_EXEC_STATIC void TwoPhaseCleanerMain() g_instance.attr.attr_network.PoolerPort, u_sess->attr.attr_storage.twophase_clean_workers); securec_check_ss(rc, "\0", "\0"); -#endif -#else - rc = sprintf_s(cmd, - sizeof(cmd), - "gs_clean -a -p %d -h localhost -e -v -r -j %d > /dev/null 2>&1", - g_instance.attr.attr_network.PostPortNumber, - u_sess->attr.attr_storage.twophase_clean_workers); - securec_check_ss(rc, "\0", "\0"); + #endif socket_close_on_exec(); @@ -193,6 +241,16 @@ NON_EXEC_STATIC void TwoPhaseCleanerMain() ereport(WARNING, (errmsg("clean up 2pc transactions failed"))); } } +#else + if (DropTempNamespace()) { + ereport(DEBUG5, (errmsg("clean up temp schemas succeed"))); + clean_successed = true; + bSyncXactsCallGsclean = false; + } else { + clean_successed = false; + ereport(WARNING, (errmsg("clean up temp schemas failed"))); + } +#endif gs_clean_start_time = GetCurrentTimestamp(); } @@ -248,6 +306,7 @@ static void TwoPCShutdownHandler(SIGNAL_ARGS) errno = save_errno; } +#ifdef ENABLE_MULTIPLE_NODES static int get_prog_path(const char* argv0) { char* exec_path = NULL; @@ -318,3 +377,426 @@ static int get_prog_path(const char* argv0) return 0; } +#endif + +#ifndef ENABLE_MULTIPLE_NODES +static PGconn* LoginDatabase(char* host, int port, char* user, char* password, + char* dbname, const char* progname, char* encoding) +{ + PGconn* conn = NULL; + char portValue[32]; +#define PARAMS_ARRAY_SIZE 10 + const char* keywords[PARAMS_ARRAY_SIZE]; + const char* values[PARAMS_ARRAY_SIZE]; + int count = 0; + int retryNum = 10; + int rc; + + rc = sprintf_s(portValue, sizeof(portValue), "%d", port); + securec_check_ss_c(rc, "\0", "\0"); + + keywords[0] = "host"; + values[0] = host; + keywords[1] = "port"; + values[1] = portValue; + keywords[2] = "user"; + values[2] = user; + keywords[3] = "password"; + values[3] = password; + keywords[4] = "dbname"; + values[4] = dbname; + keywords[5] = "fallback_application_name"; + values[5] = progname; + keywords[6] = "client_encoding"; + values[6] = encoding; + keywords[7] = "connect_timeout"; + values[7] = "5"; + keywords[8] = "options"; + /* this mode: remove timeout */ + values[8] = "-c xc_maintenance_mode=on"; + keywords[9] = NULL; + values[9] = NULL; + +retry: + /* try to connect to database */ + conn = PQconnectdbParams(keywords, values, true); + if (PQstatus(conn) != CONNECTION_OK) { + if (++count < retryNum) { + ereport(LOG, (errmsg("Could not connect to the %s, the connection info : %s", + dbname, PQerrorMessage(conn)))); + PQfinish(conn); + conn = NULL; + + /* sleep 0.1 s */ + pg_usleep(100000L); + goto retry; + } + + PQfinish(conn); + conn = NULL; + ereport(ERROR, (errcode(ERRCODE_CONNECTION_TIMED_OUT), + (errmsg("Could not connect to the %s, " + "we have tried %d times, the connection info : %s", + dbname, count, PQerrorMessage(conn))))); + } + + return (conn); +} + +static void AddDatabaseInfo(DatabaseNames** dbList, char* dbName) +{ + DatabaseNames* tempDatabase = NULL; + + tempDatabase = (DatabaseNames*)palloc(sizeof(DatabaseNames)); + + tempDatabase->next = NULL; + tempDatabase->databaseName = pstrdup(dbName); + + if (*dbList == NULL) { + *dbList = tempDatabase; + } else { + tempDatabase->next = (*dbList)->next; + (*dbList)->next = tempDatabase; + } +} + +static void GetDatabaseList(PGconn* conn, DatabaseNames** dbList) +{ + int databaseCount; + PGresult* res = NULL; + char* dbName = NULL; + + /* SQL Statement */ + static const char* STMT_GET_DATABASE_LIST = "SELECT DATNAME FROM PG_DATABASE;"; + + /* Get database list. */ + res = PQexec(conn, STMT_GET_DATABASE_LIST); + if (res == NULL || PQresultStatus(res) != PGRES_TUPLES_OK) { + PQclear(res); + ereport(ERROR, (errcode(ERRCODE_INVALID_STATUS), + errmsg("Could not obtain database list : %s", + PQerrorMessage(conn)))); + return; + } + + databaseCount = PQntuples(res); + for (int i = 0; i < databaseCount; i++) { + dbName = PQgetvalue(res, i, 0); + if (strcmp(dbName, "template0") == 0 || strcmp(dbName, "template1") == 0) { + /* Skip template0 and template1 database */ + continue; + } + + AddDatabaseInfo(dbList, dbName); + } + + PQclear(res); +} + +static void CleanDatabaseList(DatabaseNames** dbList) +{ + DatabaseNames* database = *dbList; + + while (database != NULL) { + DatabaseNames* savedDatabase = database; + database = database->next; + pfree_ext(savedDatabase->databaseName); + pfree_ext(savedDatabase); + } + + *dbList = NULL; +} + +static void AddTempSchemaInfo(TempSchemaInfo** tempSchemaList, char* tempSchemaName) +{ + TempSchemaInfo* tempSchema = NULL; + + tempSchema = (TempSchemaInfo*)palloc(sizeof(TempSchemaInfo)); + + tempSchema->next = NULL; + tempSchema->tempSchemaName = pstrdup(tempSchemaName); + + if (*tempSchemaList == NULL) { + *tempSchemaList = tempSchema; + } else { + tempSchema->next = (*tempSchemaList)->next; + (*tempSchemaList)->next = tempSchema; + } +} + +static void GetTempSchemaList(PGconn* conn, TempSchemaInfo** tempSchemaList) +{ + int tempSchemaCount; + PGresult* res = NULL; + char* nspname = NULL; + int rc; + + /* SQL Statement */ + char STMT_GET_TEMP_SCHEMA_LIST[NAMEDATALEN + 128] = {0}; + + rc = snprintf_s(STMT_GET_TEMP_SCHEMA_LIST, + sizeof(STMT_GET_TEMP_SCHEMA_LIST), + sizeof(STMT_GET_TEMP_SCHEMA_LIST) - 1, + "SELECT NSPNAME FROM PG_NAMESPACE WHERE NSPNAME LIKE 'pg_temp_%%'"); + securec_check_ss_c(rc, "\0", "\0"); + + /* Get temp schema list. */ + res = PQexec(conn, STMT_GET_TEMP_SCHEMA_LIST); + if (res == NULL || PQresultStatus(res) != PGRES_TUPLES_OK) { + PQclear(res); + ereport(ERROR, (errcode(ERRCODE_INVALID_STATUS), + errmsg("Could not obtain temp schema list : %s", + PQerrorMessage(conn)))); + return; + } + + tempSchemaCount = PQntuples(res); + for (int i = 0; i < tempSchemaCount; i++) { + nspname = PQgetvalue(res, i, 0); + if (strchr(&nspname[7], '_') == NULL) { + ereport(ERROR, (errcode(ERRCODE_INVALID_STATUS), + errmsg("Error when parse schema name : %s", + PQerrorMessage(conn)))); + return; + } + + AddTempSchemaInfo(tempSchemaList, nspname); + } + + PQclear(res); +} + +static void CleanTempSchemaList(TempSchemaInfo** tempSchemaList) +{ + TempSchemaInfo* tempSchema = *tempSchemaList; + + while (tempSchema != NULL) { + TempSchemaInfo* savedTempSchema = tempSchema; + tempSchema = tempSchema->next; + pfree_ext(savedTempSchema->tempSchemaName); + pfree_ext(savedTempSchema); + } + + *tempSchemaList = NULL; +} + +static void AddActiveBackendInfo(ActiveBackendInfo** activeBackednList, + int64 sessionID, uint32 tempID, uint32 timeLineID) +{ + ActiveBackendInfo* tempActiveBackend = NULL; + + tempActiveBackend = (ActiveBackendInfo*)palloc(sizeof(ActiveBackendInfo)); + + tempActiveBackend->next = NULL; + tempActiveBackend->sessionID = sessionID; + tempActiveBackend->tempID = tempID; + tempActiveBackend->timeLineID = timeLineID; + + if (*activeBackednList == NULL) { + *activeBackednList = tempActiveBackend; + } else { + tempActiveBackend->next = (*activeBackednList)->next; + (*activeBackednList)->next = tempActiveBackend; + } + +} + +static void GetActiveBackendList(PGconn* conn, ActiveBackendInfo** activeBackendList) +{ + int activeBackendCount; + PGresult* res = NULL; + int64 sessionID; + uint32 tempID; + uint32 timeLineID; + int rc; + + /* SQL Statement */ + char STMT_ACTIVE_BACKEND_LIST[2 * NAMEDATALEN + 128] = {0}; + + rc = sprintf_s(STMT_ACTIVE_BACKEND_LIST, + sizeof(STMT_ACTIVE_BACKEND_LIST), + "SELECT SESSIONID, TEMPID, TIMELINEID FROM PG_DATABASE D, " + "PG_STAT_GET_ACTIVITY_FOR_TEMPTABLE() AS S WHERE " + "S.DATID = D.OID AND D.DATNAME = '%s'", + PQdb(conn)); + securec_check_ss_c(rc, "\0", "\0"); + + /* Get active backend list. */ + res = PQexec(conn, STMT_ACTIVE_BACKEND_LIST); + if (res == NULL || PQresultStatus(res) != PGRES_TUPLES_OK) { + PQclear(res); + ereport(ERROR, (errcode(ERRCODE_INVALID_STATUS), + errmsg("Could not obtain active backend list : %s", + PQerrorMessage(conn)))); + return; + } + + activeBackendCount = PQntuples(res); + for (int i = 0; i < activeBackendCount; i++) { + char* result = PQgetvalue(res, i, 0); + sessionID = atoll(result); + result = PQgetvalue(res, i, 1); + tempID = atoi(result); + result = PQgetvalue(res, i, 2); + timeLineID = atoi(result); + + AddActiveBackendInfo(activeBackendList, sessionID, tempID, timeLineID); + } + + PQclear(res); +} + +static void CleanActiveBackendList(ActiveBackendInfo** activeBackendList) +{ + ActiveBackendInfo* activeBackend = *activeBackendList; + while (activeBackend != NULL) { + ActiveBackendInfo* savedActiveBackend = activeBackend; + activeBackend = activeBackend->next; + pfree_ext(savedActiveBackend); + } + + *activeBackendList = NULL; +} + +static void DropTempSchema(PGconn* conn, char* nspname) +{ + PGresult* res = NULL; + int rc; + + char toastnspName[NAMEDATALEN] = {0}; + /* SQL Statement */ + char STMT_DROP_TEMP_SCHEMA[2 * NAMEDATALEN + 64] = {0}; + + rc = snprintf_s(toastnspName, NAMEDATALEN, strlen(nspname) + 7, "pg_toast_temp_%s", nspname + 8); + securec_check_ss_c(rc, "\0", "\0"); + + rc = snprintf_s(STMT_DROP_TEMP_SCHEMA, + sizeof(STMT_DROP_TEMP_SCHEMA), + 2 * strlen(nspname) + 30, + "DROP SCHEMA %s, %s CASCADE;", + nspname, + toastnspName); + securec_check_ss_c(rc, "\0", "\0"); + + res = PQexec(conn, STMT_DROP_TEMP_SCHEMA); + + /* If exec is not success, give an log and go on. */ + if (res == NULL || PQresultStatus(res) != PGRES_COMMAND_OK) { + ereport(ERROR, (errcode(ERRCODE_INVALID_STATUS), + errmsg("Could not drop temp schema %s, %s: %s\n", + nspname, toastnspName, PQresultErrorMessage(res)))); + } + + PQclear(res); +} + +static void DropTempSchemas(PGconn* conn) +{ + /* Store temp schema names selected from DN. */ + TempSchemaInfo* tempSchemaList = NULL; + /* Store current active backend pid to decide which temp schema should be dropped. */ + ActiveBackendInfo* activeBackendList = NULL; + + GetTempSchemaList(conn, &tempSchemaList); + GetActiveBackendList(conn, &activeBackendList); + + TempSchemaInfo* tempSchema = tempSchemaList; + ActiveBackendInfo* activeBackend = NULL; + char tempBuffer[NAMEDATALEN] = {0}; + char tempBuffer2[NAMEDATALEN] = {0}; + errno_t rc; + if (tempSchemaList == NULL || activeBackendList == NULL) + return; + while (tempSchema != NULL) { + /* + * Get sessionID, tempID, timelineID from end to start. + * Temp schema name is pg_temp_%s_%u_%u_%lu. + * These items are DN'name, timelineID, tempID, sessionID. + */ + const char* lastPos = strrchr(tempSchema->tempSchemaName, '_'); + if (lastPos == NULL) + elog(ERROR, "strrchr failed, can't find '%c' in '%s'\n", '_', tempSchema->tempSchemaName); + int64 sessionID = strtoll(lastPos + 1, NULL, 10); + rc = strncpy_s(tempBuffer, sizeof(tempBuffer), tempSchema->tempSchemaName, + lastPos - tempSchema->tempSchemaName); + securec_check_c(rc, "\0", "\0"); + const char* secondLastPos = strrchr(tempBuffer, '_'); + if (secondLastPos == NULL) + elog(ERROR, "strrchr failed, can't find '%c' in '%s'\n", '_', tempBuffer); + uint32 tempID = strtol(secondLastPos + 1, NULL, 10); + rc = strncpy_s(tempBuffer2, sizeof(tempBuffer2), tempBuffer, secondLastPos - tempBuffer); + securec_check_c(rc, "\0", "\0"); + const char* thirdLastPos = strrchr(tempBuffer2, '_'); + if (thirdLastPos == NULL) + elog(ERROR, "strrchr failed, can't find '%c' in '%s'\n", '_', tempBuffer2); + uint32 timeLineID = strtol(thirdLastPos + 1, NULL, 10); + + activeBackend = activeBackendList; + /* + * Thus, We use sessionID, timeLineID and tempID together + * to judge which session a temp table belongs to, instead of sessionID only. + */ + while (activeBackend != NULL) { + if (sessionID == activeBackend->sessionID && tempID == activeBackend->tempID && + timeLineID == activeBackend->timeLineID) + break; + activeBackend = activeBackend->next; + } + if (activeBackend == NULL) { + /* + * Now it is not an active backend, furthermore, we need to drop it; + */ + DropTempSchema(conn, tempSchema->tempSchemaName); + } + tempSchema = tempSchema->next; + } + + CleanTempSchemaList(&tempSchemaList); + CleanActiveBackendList(&activeBackendList); +} + +/* + * @Description: Drop inactive temp schemas of each database. + * @in: void + * @return: the result of cleaning temporary schemas + */ +static bool DropTempNamespace() +{ + DatabaseNames* dbList = NULL; + DatabaseNames* curDatabase = NULL; + + PGconn* conn = NULL; + conn = LoginDatabase("localhost", g_instance.attr.attr_network.PostPortNumber, + NULL, NULL, DEFAULT_DATABASE, PGXC_CLEAN, "auto"); + + /* Get database list. */ + GetDatabaseList(conn, &dbList); + + PQfinish(conn); + conn = NULL; + + if (dbList != NULL) { + for (curDatabase = dbList; curDatabase != NULL; curDatabase = curDatabase->next) { + conn = LoginDatabase("localhost", g_instance.attr.attr_network.PostPortNumber, + NULL, NULL, curDatabase->databaseName, PGXC_CLEAN, "auto"); + + if (conn == NULL) { + ereport(ERROR, (errcode(ERRCODE_INVALID_STATUS), + (errmsg("Could not connect to the database %s.", + curDatabase->databaseName)))); + return false; + } + + DropTempSchemas(conn); + + ereport(DEBUG5, (errmsg("Drop temp namespace for database \"%s\" finished.", + curDatabase->databaseName))); + PQfinish(conn); + conn = NULL; + } + } + + CleanDatabaseList(&dbList); + return true; +} +#endif