Nemoo
/
openGauss-server
forked from nuoya/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

clean

This commit is contained in:
Xiao__Ma 2021-07-29 19:20:49 +08:00
parent 2f1e988b6f
commit 2f0096bbf4
7 changed files with 541 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/bin/Makefile
View File

@ -52,7 +52,6 @@ SUBDIRS = \
pg_controldata \
pg_resetxlog \
gs_guc \
pgxc_clean \
gsqlerr \
pg_basebackup \
pg_probackup

1
src/bin/gs_guc/cluster_guc.conf
View File

@ -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.|

45
src/bin/pgxc_clean/pgxc_clean.cpp
View File

@ -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)
{

20
src/common/backend/utils/init/miscinit.cpp
View File

@ -32,6 +32,7 @@
#include <sys/utsname.h>
#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

11
src/common/backend/utils/init/postinit.cpp
View File

@ -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()

15
src/common/backend/utils/misc/guc.cpp
View File

@ -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;

504
src/gausskernel/process/postmaster/twophasecleaner.cpp
View File

@ -31,7 +31,9 @@
#include <sys/stat.h>
#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