forked from huawei/openGauss-server
Compare commits
1 Commits
master
...
undorecycl
| Author | SHA1 | Date |
|---|---|---|
林科旭
|
0c4cf14576 |
|
|
@ -1229,12 +1229,6 @@ parse_next_sync_groups(char **pgroup, char *result)
|
|||
static int
|
||||
transform_az_name(char *config_value, char *allAZString, int allAZStringBufLen, const char *data_dir)
|
||||
{
|
||||
if (strcmp(config_value, "''") == 0) {
|
||||
errno_t rc = strncpy_s(allAZString, allAZStringBufLen, config_value, strlen(config_value));
|
||||
securec_check_c(rc, "\0", "\0");
|
||||
return SUCCESS;
|
||||
}
|
||||
|
||||
char *azString = NULL;
|
||||
char *buf = allAZString;
|
||||
int buflen = allAZStringBufLen;
|
||||
|
|
|
|||
|
|
@ -5923,7 +5923,7 @@ int main(int argc, char** argv)
|
|||
&option_index)) != -1)
|
||||
#endif
|
||||
#else
|
||||
while ((c = getopt_long(argc, argv, "b:cD:e:fi:G:l:m:M:N:o:O:p:P:r:R:v:x:sS:t:u:U:wWZ:C:dqL:T:Q:", long_options,
|
||||
while ((c = getopt_long(argc, argv, "b:cD:e:fi:G:l:m:M:N:o:O:p:P:r:R:v:x:sS:t:u:U:wWZ:dqL:T:Q:", long_options,
|
||||
&option_index)) != -1)
|
||||
#endif
|
||||
#endif
|
||||
|
|
|
|||
|
|
@ -233,7 +233,7 @@ char* all_data_nodename_list = NULL;
|
|||
const uint32 USTORE_UPGRADE_VERSION = 92368;
|
||||
const uint32 PACKAGE_ENHANCEMENT = 92444;
|
||||
const uint32 SUBSCRIPTION_VERSION = 92580;
|
||||
const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92606;
|
||||
const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92607;
|
||||
|
||||
#ifdef DUMPSYSLOG
|
||||
char* syslogpath = NULL;
|
||||
|
|
|
|||
|
|
@ -31,9 +31,6 @@
|
|||
it will be backuped up in external dirs */
|
||||
parray *pgdata_nobackup_dir = NULL;
|
||||
|
||||
/* list of logical replication slots */
|
||||
parray *logical_replslot = NULL;
|
||||
|
||||
static int standby_message_timeout_local = 10 ; /* 10 sec = default */
|
||||
static XLogRecPtr stop_backup_lsn = InvalidXLogRecPtr;
|
||||
static XLogRecPtr stop_stream_lsn = InvalidXLogRecPtr;
|
||||
|
|
@ -92,11 +89,10 @@ static void backup_cleanup(bool fatal, void *userdata);
|
|||
|
||||
static void *backup_files(void *arg);
|
||||
|
||||
static void do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool backup_logs,
|
||||
bool backup_replslots);
|
||||
static void do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool backup_logs);
|
||||
|
||||
static void pg_start_backup(const char *label, bool smooth, pgBackup *backup,
|
||||
PGNodeInfo *nodeInfo, PGconn *conn, bool backup_replslots);
|
||||
PGNodeInfo *nodeInfo, PGconn *conn);
|
||||
static void pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn, PGNodeInfo *nodeInfo);
|
||||
static int checkpoint_timeout(PGconn *backup_conn);
|
||||
|
||||
|
|
@ -562,7 +558,7 @@ static void sync_files(parray *database_map, const char *database_path, parray *
|
|||
* Move files from 'pgdata' to a subdirectory in 'backup_path'.
|
||||
*/
|
||||
static void
|
||||
do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool backup_logs, bool backup_replslots)
|
||||
do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool backup_logs)
|
||||
{
|
||||
int i;
|
||||
char database_path[MAXPGPATH];
|
||||
|
|
@ -595,7 +591,7 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool
|
|||
securec_check_c(rc, "\0", "\0");
|
||||
|
||||
/* Call pg_start_backup function in openGauss connect */
|
||||
pg_start_backup(label, smooth_checkpoint, ¤t, nodeInfo, backup_conn, backup_replslots);
|
||||
pg_start_backup(label, smooth_checkpoint, ¤t, nodeInfo, backup_conn);
|
||||
|
||||
/* Obtain current timeline */
|
||||
#if PG_VERSION_NUM >= 90600
|
||||
|
|
@ -628,10 +624,10 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool
|
|||
/* list files with the logical path. omit $PGDATA */
|
||||
if (fio_is_remote(FIO_DB_HOST))
|
||||
fio_list_dir(backup_files_list, instance_config.pgdata,
|
||||
true, true, false, backup_logs, true, 0, backup_replslots);
|
||||
true, true, false, backup_logs, true, 0);
|
||||
else
|
||||
dir_list_file(backup_files_list, instance_config.pgdata,
|
||||
true, true, false, backup_logs, true, 0, FIO_LOCAL_HOST, backup_replslots);
|
||||
true, true, false, backup_logs, true, 0, FIO_LOCAL_HOST);
|
||||
|
||||
/*
|
||||
* Get database_map (name to oid) for use in partial restore feature.
|
||||
|
|
@ -753,11 +749,6 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool
|
|||
}
|
||||
pgdata_nobackup_dir = NULL;
|
||||
|
||||
if (logical_replslot) {
|
||||
free_dir_list(logical_replslot);
|
||||
}
|
||||
logical_replslot = NULL;
|
||||
|
||||
/* Cleanup */
|
||||
if (backup_list)
|
||||
{
|
||||
|
|
@ -858,7 +849,7 @@ static void do_after_backup()
|
|||
*/
|
||||
int
|
||||
do_backup(time_t start_time, pgSetBackupParams *set_backup_params,
|
||||
bool no_validate, bool no_sync, bool backup_logs, bool backup_replslots)
|
||||
bool no_validate, bool no_sync, bool backup_logs)
|
||||
{
|
||||
PGconn *backup_conn = NULL;
|
||||
PGNodeInfo nodeInfo;
|
||||
|
|
@ -934,7 +925,7 @@ do_backup(time_t start_time, pgSetBackupParams *set_backup_params,
|
|||
add_note(¤t, set_backup_params->note);
|
||||
|
||||
/* backup data */
|
||||
do_backup_instance(backup_conn, &nodeInfo, no_sync, backup_logs, backup_replslots);
|
||||
do_backup_instance(backup_conn, &nodeInfo, no_sync, backup_logs);
|
||||
pgut_atexit_pop(backup_cleanup, NULL);
|
||||
|
||||
/* compute size of wal files of this backup stored in the archive */
|
||||
|
|
@ -1043,15 +1034,13 @@ confirm_block_size(PGconn *conn, const char *name, int blcksz)
|
|||
*/
|
||||
static void
|
||||
pg_start_backup(const char *label, bool smooth, pgBackup *backup,
|
||||
PGNodeInfo *nodeInfo, PGconn *conn, bool backup_replslots)
|
||||
PGNodeInfo *nodeInfo, PGconn *conn)
|
||||
{
|
||||
PGresult *res;
|
||||
const char *params[2];
|
||||
uint32 lsn_hi;
|
||||
uint32 lsn_lo;
|
||||
int ret;
|
||||
int i;
|
||||
XLogRecPtr startLsn;
|
||||
|
||||
params[0] = label;
|
||||
|
||||
|
|
@ -1079,33 +1068,7 @@ pg_start_backup(const char *label, bool smooth, pgBackup *backup,
|
|||
XLogDataFromLSN(ret, PQgetvalue(res, 0, 0), &lsn_hi, &lsn_lo);
|
||||
securec_check_for_sscanf_s(ret, 2, "\0", "\0");
|
||||
/* Calculate LSN */
|
||||
startLsn = ((uint64) lsn_hi )<< 32 | lsn_lo;
|
||||
|
||||
if (backup_replslots) {
|
||||
logical_replslot = parray_new();
|
||||
/* query for logical replication slots of subscriptions */
|
||||
res = pgut_execute(conn,
|
||||
"SELECT slot_name, restart_lsn FROM pg_catalog.pg_get_replication_slots()"
|
||||
"WHERE slot_type = 'logical' AND plugin = 'pgoutput'", 0, NULL);
|
||||
if (PQntuples(res) == 0) {
|
||||
elog(LOG, "logical replication slots for subscriptions not found");
|
||||
} else {
|
||||
XLogRecPtr repslotLsn;
|
||||
|
||||
for (i = 0; i < PQntuples(res); i++) {
|
||||
XLogDataFromLSN(ret, PQgetvalue(res, i, 1), &lsn_hi, &lsn_lo);
|
||||
securec_check_for_sscanf_s(ret, 2, "\0", "\0");
|
||||
repslotLsn = ((uint64) lsn_hi )<< 32 | lsn_lo;
|
||||
startLsn = Min(startLsn, repslotLsn);
|
||||
|
||||
char* slotname = pg_strdup(PQgetvalue(res, i, 0));
|
||||
parray_append(logical_replslot, slotname);
|
||||
}
|
||||
elog(WARNING, "logical replication slots for subscriptions will be backed up. "
|
||||
"If don't use them after restoring, please drop them to avoid affecting xlog recycling.");
|
||||
}
|
||||
}
|
||||
backup->start_lsn = startLsn;
|
||||
backup->start_lsn = ((uint64) lsn_hi )<< 32 | lsn_lo;
|
||||
|
||||
PQclear(res);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -42,6 +42,13 @@ const char *pgdata_exclude_dir[] =
|
|||
(const char *)"pg_stat_tmp",
|
||||
(const char *)"pgsql_tmp",
|
||||
|
||||
/*
|
||||
* It is generally not useful to backup the contents of this directory even
|
||||
* if the intention is to restore to another master. See backup.sgml for a
|
||||
* more detailed description.
|
||||
*/
|
||||
(const char *)"pg_replslot",
|
||||
|
||||
/* Contents removed on startup, see dsm_cleanup_for_mmap(). */
|
||||
(const char *)"pg_dynshmem",
|
||||
|
||||
|
|
@ -61,7 +68,7 @@ const char *pgdata_exclude_dir[] =
|
|||
(const char *)"pg_subtrans",
|
||||
|
||||
/* end of list */
|
||||
NULL, /* pg_log and pg_replslot will be set later */
|
||||
NULL, /* pg_log will be set later */
|
||||
NULL
|
||||
};
|
||||
|
||||
|
|
@ -121,20 +128,17 @@ may be removed int the future */
|
|||
|
||||
static int pgCompareString(const void *str1, const void *str2);
|
||||
|
||||
static char dir_check_file(pgFile *file, bool backup_logs, bool backup_replslots);
|
||||
static char dir_check_file(pgFile *file, bool backup_logs);
|
||||
static char check_in_tablespace(pgFile *file, bool in_tablespace);
|
||||
static char check_db_dir(pgFile *file);
|
||||
static char check_digit_file(pgFile *file);
|
||||
static char check_nobackup_dir(pgFile *file);
|
||||
static void dir_list_file_internal(parray *files, pgFile *parent, const char *parent_dir,
|
||||
bool exclude, bool follow_symlink, bool backup_logs,
|
||||
bool skip_hidden, int external_dir_num, fio_location location,
|
||||
bool backup_replslots);
|
||||
bool skip_hidden, int external_dir_num, fio_location location);
|
||||
static void opt_path_map(ConfigOption *opt, const char *arg,
|
||||
TablespaceList *list, const char *type);
|
||||
|
||||
char check_logical_replslot_dir(const char *rel_path);
|
||||
|
||||
/* Tablespace mapping */
|
||||
static TablespaceList tablespace_dirs = {NULL, NULL};
|
||||
/* Extra directories mapping */
|
||||
|
|
@ -534,7 +538,7 @@ db_map_entry_free(void *entry)
|
|||
void
|
||||
dir_list_file(parray *files, const char *root, bool exclude, bool follow_symlink,
|
||||
bool add_root, bool backup_logs, bool skip_hidden, int external_dir_num,
|
||||
fio_location location, bool backup_replslots)
|
||||
fio_location location)
|
||||
{
|
||||
pgFile *file;
|
||||
|
||||
|
|
@ -561,7 +565,7 @@ dir_list_file(parray *files, const char *root, bool exclude, bool follow_symlink
|
|||
parray_append(files, file);
|
||||
|
||||
dir_list_file_internal(files, file, root, exclude, follow_symlink,
|
||||
backup_logs, skip_hidden, external_dir_num, location, backup_replslots);
|
||||
backup_logs, skip_hidden, external_dir_num, location);
|
||||
|
||||
if (!add_root)
|
||||
pgFileFree(file);
|
||||
|
|
@ -585,7 +589,7 @@ dir_list_file(parray *files, const char *root, bool exclude, bool follow_symlink
|
|||
* - datafiles
|
||||
*/
|
||||
static char
|
||||
dir_check_file(pgFile *file, bool backup_logs, bool backup_replslots)
|
||||
dir_check_file(pgFile *file, bool backup_logs)
|
||||
{
|
||||
int i;
|
||||
int sscanf_res;
|
||||
|
|
@ -648,29 +652,6 @@ dir_check_file(pgFile *file, bool backup_logs, bool backup_replslots)
|
|||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* Backup pg_replslot if it is specified.
|
||||
* It is generally not useful to backup the contents of this directory even
|
||||
* if the intention is to restore to another master. See backup.sgml for a
|
||||
* more detailed description.
|
||||
*/
|
||||
if (!backup_replslots) {
|
||||
if (strcmp(file->rel_path, PG_REPLSLOT_DIR) == 0) {
|
||||
/* Skip */
|
||||
elog(VERBOSE, "Excluding directory content: %s", file->rel_path);
|
||||
return CHECK_EXCLUDE_FALSE;
|
||||
}
|
||||
} else {
|
||||
/*
|
||||
* Check file that under pg_replslot and judge whether it
|
||||
* belonged to logical replication slots for subscriptions.
|
||||
*/
|
||||
if (strcmp(file->rel_path, PG_REPLSLOT_DIR) != 0 &&
|
||||
path_is_prefix_of_path(PG_REPLSLOT_DIR, file->rel_path)) {
|
||||
return check_logical_replslot_dir(file->rel_path);
|
||||
}
|
||||
}
|
||||
|
||||
ret = check_nobackup_dir(file);
|
||||
if (ret != -1) { /* -1 means need backup */
|
||||
return ret;
|
||||
|
|
@ -768,35 +749,6 @@ static char check_nobackup_dir(pgFile *file)
|
|||
return ret;
|
||||
}
|
||||
|
||||
char check_logical_replslot_dir(const char *rel_path)
|
||||
{
|
||||
char ret = CHECK_FALSE;
|
||||
int i = 0;
|
||||
char *tmp = pg_strdup(rel_path);
|
||||
char *p;
|
||||
#define DIRECTORY_DELIMITER "/"
|
||||
|
||||
if (logical_replslot) {
|
||||
/* extract slot name from rel_path, such as sub1 from pg_replslot/sub1/snap */
|
||||
p = strtok(tmp, DIRECTORY_DELIMITER);
|
||||
if (p != NULL) {
|
||||
p = strtok(NULL, DIRECTORY_DELIMITER);
|
||||
}
|
||||
|
||||
for (i = 0; p != NULL && i < (int)parray_num(logical_replslot); i++) {
|
||||
char *slotName = (char *)parray_get(logical_replslot, i);
|
||||
if (strcmp(p, slotName) == 0) {
|
||||
pfree(tmp);
|
||||
return CHECK_TRUE;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
ret = CHECK_TRUE;
|
||||
}
|
||||
pfree(tmp);
|
||||
return ret;
|
||||
}
|
||||
|
||||
static char check_db_dir(pgFile *file)
|
||||
{
|
||||
char ret = -1;
|
||||
|
|
@ -937,8 +889,7 @@ bool SkipSomeDirFile(pgFile *file, struct dirent *dent, bool skipHidden)
|
|||
static void
|
||||
dir_list_file_internal(parray *files, pgFile *parent, const char *parent_dir,
|
||||
bool exclude, bool follow_symlink, bool backup_logs,
|
||||
bool skip_hidden, int external_dir_num, fio_location location,
|
||||
bool backup_replslots)
|
||||
bool skip_hidden, int external_dir_num, fio_location location)
|
||||
{
|
||||
DIR *dir;
|
||||
struct dirent *dent;
|
||||
|
|
@ -986,7 +937,7 @@ dir_list_file_internal(parray *files, pgFile *parent, const char *parent_dir,
|
|||
|
||||
if (exclude)
|
||||
{
|
||||
check_res = dir_check_file(file, backup_logs, backup_replslots);
|
||||
check_res = dir_check_file(file, backup_logs);
|
||||
if (check_res == CHECK_FALSE)
|
||||
{
|
||||
/* Skip */
|
||||
|
|
@ -1012,7 +963,7 @@ dir_list_file_internal(parray *files, pgFile *parent, const char *parent_dir,
|
|||
*/
|
||||
if (S_ISDIR(file->mode))
|
||||
dir_list_file_internal(files, file, child, exclude, follow_symlink,
|
||||
backup_logs, skip_hidden, external_dir_num, location, backup_replslots);
|
||||
backup_logs, skip_hidden, external_dir_num, location);
|
||||
}
|
||||
|
||||
if (errno && errno != ENOENT)
|
||||
|
|
|
|||
|
|
@ -51,7 +51,6 @@ typedef struct
|
|||
bool exclusive_backup;
|
||||
bool skip_hidden;
|
||||
int external_dir_num;
|
||||
bool backup_replslots;
|
||||
} fio_list_dir_request;
|
||||
|
||||
typedef struct
|
||||
|
|
@ -1795,7 +1794,7 @@ cleanup:
|
|||
/* Compile the array of files located on remote machine in directory root */
|
||||
void fio_list_dir(parray *files, const char *root, bool exclude,
|
||||
bool follow_symlink, bool add_root, bool backup_logs,
|
||||
bool skip_hidden, int external_dir_num, bool backup_replslots)
|
||||
bool skip_hidden, int external_dir_num)
|
||||
{
|
||||
fio_header hdr;
|
||||
fio_list_dir_request req;
|
||||
|
|
@ -1812,7 +1811,6 @@ void fio_list_dir(parray *files, const char *root, bool exclude,
|
|||
req.exclusive_backup = exclusive_backup;
|
||||
req.skip_hidden = skip_hidden;
|
||||
req.external_dir_num = external_dir_num;
|
||||
req.backup_replslots = backup_replslots;
|
||||
|
||||
hdr.cop = FIO_LIST_DIR;
|
||||
hdr.size = sizeof(req);
|
||||
|
|
@ -1872,14 +1870,7 @@ void fio_list_dir(parray *files, const char *root, bool exclude,
|
|||
securec_check_ss_c(nRet, "\0", "\0");
|
||||
}
|
||||
|
||||
/*
|
||||
* Check file that under pg_replslot and judge whether it
|
||||
* belonged to logical replication slots for subscriptions.
|
||||
*/
|
||||
if (backup_replslots && strcmp(buf, PG_REPLSLOT_DIR) != 0 &&
|
||||
path_is_prefix_of_path(PG_REPLSLOT_DIR, buf) && check_logical_replslot_dir(file->rel_path) != 1) {
|
||||
continue;
|
||||
}
|
||||
|
||||
|
||||
parray_append(files, file);
|
||||
}
|
||||
|
|
@ -1923,7 +1914,7 @@ static void fio_list_dir_impl(int out, char* buf)
|
|||
|
||||
dir_list_file(file_files, req->path, req->exclude, req->follow_symlink,
|
||||
req->add_root, req->backup_logs, req->skip_hidden,
|
||||
req->external_dir_num, FIO_LOCAL_HOST, req->backup_replslots);
|
||||
req->external_dir_num, FIO_LOCAL_HOST);
|
||||
|
||||
/* send information about files to the main process */
|
||||
for (i = 0; i < (int)parray_num(file_files); i++)
|
||||
|
|
|
|||
|
|
@ -163,7 +163,5 @@ extern z_off_t fio_gzseek(gzFile f, z_off_t offset, int whence);
|
|||
extern const char* fio_gzerror(gzFile file, int *errnum);
|
||||
#endif
|
||||
|
||||
extern char check_logical_replslot_dir(const char *rel_path);
|
||||
|
||||
#endif
|
||||
|
||||
|
|
|
|||
|
|
@ -154,7 +154,6 @@ void help_pg_probackup(void)
|
|||
printf(_(" [--remote-port=port] [--ssh-options=ssh_options]\n"));
|
||||
printf(_(" [--remote-libpath=libpath]\n"));
|
||||
printf(_(" [--ttl=interval] [--expire-time=time]\n"));
|
||||
printf(_(" [--backup-pg-replslot]\n"));
|
||||
printf(_(" [--help]\n"));
|
||||
|
||||
printf(_("\n %s restore -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
|
||||
|
|
@ -421,7 +420,6 @@ static void help_backup(void)
|
|||
printf(_(" [--remote-port=port] [--ssh-options=ssh_options]\n"));
|
||||
printf(_(" [--remote-libpath=libpath]\n"));
|
||||
printf(_(" [--ttl=interval] [--expire-time=time]\n\n"));
|
||||
printf(_(" [--backup-pg-replslot]\n"));
|
||||
|
||||
printf(_(" -B, --backup-path=backup-path location of the backup storage area\n"));
|
||||
printf(_(" --instance=instance_name name of the instance\n"));
|
||||
|
|
@ -443,7 +441,6 @@ static void help_backup(void)
|
|||
printf(_(" --note=text add note to backup\n"));
|
||||
printf(_(" (example: --note='backup before app update to v13.1')\n"));
|
||||
printf(_(" --archive-timeout=timeout wait timeout for WAL segment archiving (default: 5min)\n"));
|
||||
printf(_(" --backup-pg-replslot] backup of '%s' directory\n"), PG_REPLSLOT_DIR);
|
||||
|
||||
printf(_("\n Logging options:\n"));
|
||||
printf(_(" --log-level-console=log-level-console\n"));
|
||||
|
|
|
|||
|
|
@ -77,7 +77,6 @@ int rw_timeout = 0;
|
|||
|
||||
/* backup options */
|
||||
bool backup_logs = false;
|
||||
bool backup_replslots = false;
|
||||
bool smooth_checkpoint;
|
||||
char *remote_agent;
|
||||
static char *backup_note = NULL;
|
||||
|
|
@ -187,7 +186,6 @@ static ConfigOption cmd_options[] =
|
|||
{ 'b', 145, "wal", &delete_wal, SOURCE_CMD_STRICT },
|
||||
{ 'b', 146, "expired", &delete_expired, SOURCE_CMD_STRICT },
|
||||
{ 's', 172, "status", &delete_status, SOURCE_CMD_STRICT },
|
||||
{ 'b', 186, "backup-pg-replslot", &backup_replslots, SOURCE_CMD_STRICT},
|
||||
|
||||
{ 'b', 147, "force", &force, SOURCE_CMD_STRICT },
|
||||
{ 'b', 148, "compress", &compress_shortcut, SOURCE_CMD_STRICT },
|
||||
|
|
@ -552,7 +550,7 @@ static int do_actual_operate()
|
|||
elog(ERROR, "required parameter not specified: BACKUP_MODE "
|
||||
"(-b, --backup-mode)");
|
||||
|
||||
return do_backup(start_time, set_backup_params, no_validate, no_sync, backup_logs, backup_replslots);
|
||||
return do_backup(start_time, set_backup_params, no_validate, no_sync, backup_logs);
|
||||
}
|
||||
case RESTORE_CMD:
|
||||
return do_restore_or_validate(current.backup_id,
|
||||
|
|
|
|||
|
|
@ -69,7 +69,6 @@ extern const char *PROGRAM_FULL_PATH;
|
|||
#define HEADER_MAP "page_header_map"
|
||||
#define HEADER_MAP_TMP "page_header_map_tmp"
|
||||
#define PG_RELATIVE_TBLSPC_DIR "pg_location"
|
||||
#define PG_REPLSLOT_DIR "pg_replslot"
|
||||
|
||||
/* Timeout defaults */
|
||||
#define ARCHIVE_TIMEOUT_DEFAULT 300
|
||||
|
|
|
|||
|
|
@ -54,9 +54,6 @@ extern bool smooth_checkpoint;
|
|||
it will be backuped up in external dirs */
|
||||
extern parray *pgdata_nobackup_dir;
|
||||
|
||||
/* list of logical replication slots */
|
||||
extern parray *logical_replslot;
|
||||
|
||||
/* remote probackup options */
|
||||
extern char* remote_agent;
|
||||
|
||||
|
|
@ -92,7 +89,7 @@ extern const char *pgdata_exclude_dir[];
|
|||
|
||||
/* in backup.c */
|
||||
extern int do_backup(time_t start_time, pgSetBackupParams *set_backup_params,
|
||||
bool no_validate, bool no_sync, bool backup_logs, bool backup_replslots);
|
||||
bool no_validate, bool no_sync, bool backup_logs);
|
||||
extern BackupMode parse_backup_mode(const char *value);
|
||||
extern const char *deparse_backup_mode(BackupMode mode);
|
||||
extern void process_block_change(ForkNumber forknum, const RelFileNode rnode,
|
||||
|
|
@ -242,8 +239,7 @@ extern const char* deparse_compress_alg(int alg);
|
|||
/* in dir.c */
|
||||
extern void dir_list_file(parray *files, const char *root, bool exclude,
|
||||
bool follow_symlink, bool add_root, bool backup_logs,
|
||||
bool skip_hidden, int external_dir_num, fio_location location,
|
||||
bool backup_replslots = false);
|
||||
bool skip_hidden, int external_dir_num, fio_location location);
|
||||
|
||||
extern void create_data_directories(parray *dest_files,
|
||||
const char *data_dir,
|
||||
|
|
@ -436,8 +432,7 @@ extern int fio_send_file(const char *from_fullpath, const char *to_fullpath, FIL
|
|||
pgFile *file, char **errormsg);
|
||||
|
||||
extern void fio_list_dir(parray *files, const char *root, bool exclude, bool follow_symlink,
|
||||
bool add_root, bool backup_logs, bool skip_hidden, int external_dir_num,
|
||||
bool backup_replslots = false);
|
||||
bool add_root, bool backup_logs, bool skip_hidden, int external_dir_num);
|
||||
|
||||
extern bool pgut_rmtree(const char *path, bool rmtopdir, bool strict);
|
||||
|
||||
|
|
|
|||
|
|
@ -8641,7 +8641,7 @@
|
|||
),
|
||||
AddFuncGroup(
|
||||
"pg_stat_get_wal_senders", 1,
|
||||
AddBuiltinFunc(_0(3099), _1("pg_stat_get_wal_senders"), _2(0), _3(false), _4(true), _5(pg_stat_get_wal_senders), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(10), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('s'), _19(0), _20(0), _21(21, 20, 23, 25, 25, 25, 25, 1184, 1184, 25, 25, 25, 25, 25, 25, 25, 25, 25, 25, 23, 25, 25), _22(21, 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o'), _23(21, "pid", "sender_pid", "local_role", "peer_role", "peer_state", "state", "catchup_start", "catchup_end", "sender_sent_location", "sender_write_location", "sender_flush_location", "sender_replay_location", "receiver_received_location", "receiver_write_location", "receiver_flush_location", "receiver_replay_location", "sync_percent", "sync_state", "sync_priority", "sync_most_available", "channel"), _24(NULL), _25("pg_stat_get_wal_senders"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33("statistics: information about currently active replication"), _34('f'), _35(NULL), _36(0), _37(false), _38(NULL), _39(NULL), _40(0))
|
||||
AddBuiltinFunc(_0(3099), _1("pg_stat_get_wal_senders"), _2(0), _3(false), _4(true), _5(pg_stat_get_wal_senders), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(10), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('s'), _19(0), _20(0), _21(22, 20, 23, 25, 25, 25, 25, 1184, 1184, 25, 25, 25, 25, 25, 25, 25, 25, 25, 25, 23, 23, 25, 25), _22(22, 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o'), _23(22, "pid", "sender_pid", "local_role", "peer_role", "peer_state", "state", "catchup_start", "catchup_end", "sender_sent_location", "sender_write_location", "sender_flush_location", "sender_replay_location", "receiver_received_location", "receiver_write_location", "receiver_flush_location", "receiver_replay_location", "sync_percent", "sync_state", "sync_group", "sync_priority", "sync_most_available", "channel"), _24(NULL), _25("pg_stat_get_wal_senders"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33("statistics: information about currently active replication"), _34('f'), _35(NULL), _36(0), _37(false), _38(NULL), _39(NULL), _40(0))
|
||||
),
|
||||
AddFuncGroup(
|
||||
"pg_stat_get_wlm_ec_operator_info", 1,
|
||||
|
|
|
|||
|
|
@ -68,7 +68,7 @@ Subscription *GetSubscription(Oid subid, bool missing_ok)
|
|||
|
||||
/* Get slotname */
|
||||
datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, Anum_pg_subscription_subslotname, &isnull);
|
||||
if (!isnull) {
|
||||
if (unlikely(isnull)) {
|
||||
sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
|
||||
} else {
|
||||
sub->slotname = NULL;
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ bool open_join_children = true;
|
|||
bool will_shutdown = false;
|
||||
|
||||
/* hard-wired binary version number */
|
||||
const uint32 GRAND_VERSION_NUM = 92606;
|
||||
const uint32 GRAND_VERSION_NUM = 92607;
|
||||
|
||||
const uint32 PREDPUSH_SAME_LEVEL_VERSION_NUM = 92522;
|
||||
const uint32 UPSERT_WHERE_VERSION_NUM = 92514;
|
||||
|
|
@ -101,7 +101,7 @@ const uint32 PRIVS_DIRECTORY_VERSION_NUM = 92460;
|
|||
const uint32 COMMENT_RECORD_PARAM_VERSION_NUM = 92484;
|
||||
const uint32 SCAN_BATCH_MODE_VERSION_NUM = 92568;
|
||||
const uint32 PUBLICATION_VERSION_NUM = 92580;
|
||||
const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92606;
|
||||
const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92607;
|
||||
|
||||
/* Version number of the guc parameter backend_version added in V500R001C20 */
|
||||
const uint32 V5R1C20_BACKEND_VERSION_NUM = 92305;
|
||||
|
|
|
|||
|
|
@ -26,11 +26,9 @@ import logging
|
|||
try:
|
||||
from .dao.gsql_execute import GSqlExecute
|
||||
from .dao.execute_factory import ExecuteFactory
|
||||
from .mcts import MCTS
|
||||
except ImportError:
|
||||
from dao.gsql_execute import GSqlExecute
|
||||
from dao.execute_factory import ExecuteFactory
|
||||
from mcts import MCTS
|
||||
|
||||
ENABLE_MULTI_NODE = False
|
||||
SAMPLE_NUM = 5
|
||||
|
|
@ -194,12 +192,9 @@ class IndexAdvisor:
|
|||
self.workload_used_index))
|
||||
if DRIVER:
|
||||
self.db.close_conn()
|
||||
if MAX_INDEX_STORAGE:
|
||||
opt_config = MCTS(self.workload_info[0], atomic_config_total, candidate_indexes,
|
||||
MAX_INDEX_STORAGE, MAX_INDEX_NUM)
|
||||
else:
|
||||
opt_config = greedy_determine_opt_config(self.workload_info[0], atomic_config_total,
|
||||
candidate_indexes, self.index_cost_total[0])
|
||||
|
||||
opt_config = greedy_determine_opt_config(self.workload_info[0], atomic_config_total,
|
||||
candidate_indexes, self.index_cost_total[0])
|
||||
self.retain_lower_cost_index(candidate_indexes)
|
||||
if len(opt_config) == 0:
|
||||
print("No optimal indexes generated!")
|
||||
|
|
@ -948,7 +943,7 @@ def check_parameter(args):
|
|||
raise argparse.ArgumentTypeError("%s is an invalid positive int value" %
|
||||
args.max_index_num)
|
||||
if args.max_index_storage is not None and args.max_index_storage <= 0:
|
||||
raise argparse.ArgumentTypeError("%s is an invalid positive float value" %
|
||||
raise argparse.ArgumentTypeError("%s is an invalid positive int value" %
|
||||
args.max_index_storage)
|
||||
JSON_TYPE = args.json
|
||||
MAX_INDEX_NUM = args.max_index_num
|
||||
|
|
@ -976,7 +971,7 @@ def main(argv):
|
|||
arg_parser.add_argument(
|
||||
"--max_index_num", help="Maximum number of suggested indexes", type=int)
|
||||
arg_parser.add_argument("--max_index_storage",
|
||||
help="Maximum storage of suggested indexes/MB", type=float)
|
||||
help="Maximum storage of suggested indexes/MB", type=int)
|
||||
arg_parser.add_argument("--multi_iter_mode", action='store_true',
|
||||
help="Whether to use multi-iteration algorithm", default=False)
|
||||
arg_parser.add_argument("--multi_node", action='store_true',
|
||||
|
|
|
|||
|
|
@ -1,397 +0,0 @@
|
|||
import sys
|
||||
import math
|
||||
import random
|
||||
import copy
|
||||
|
||||
STORAGE_THRESHOLD = 0
|
||||
AVAILABLE_CHOICES = None
|
||||
ATOMIC_CHOICES = None
|
||||
WORKLOAD_INFO = None
|
||||
MAX_INDEX_NUM = 0
|
||||
|
||||
|
||||
def is_same_index(index, compared_index):
|
||||
return index.table == compared_index.table and \
|
||||
index.columns == compared_index.columns and \
|
||||
index.index_type == compared_index.index_type
|
||||
|
||||
|
||||
def atomic_config_is_valid(atomic_config, config):
|
||||
# if candidate indexes contains all atomic index of current config1, then record it
|
||||
for atomic_index in atomic_config:
|
||||
is_exist = False
|
||||
for index in config:
|
||||
if is_same_index(index, atomic_index):
|
||||
index.storage = atomic_index.storage
|
||||
is_exist = True
|
||||
break
|
||||
if not is_exist:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def find_subsets_num(choice):
|
||||
atomic_subsets_num = []
|
||||
for pos, atomic in enumerate(ATOMIC_CHOICES):
|
||||
if not atomic or len(atomic) > len(choice):
|
||||
continue
|
||||
# find valid atomic index
|
||||
if atomic_config_is_valid(atomic, choice):
|
||||
atomic_subsets_num.append(pos)
|
||||
# find the same atomic index as the candidate index
|
||||
if len(atomic) == 1 and (is_same_index(choice[-1], atomic[0])):
|
||||
choice[-1].atomic_pos = pos
|
||||
return atomic_subsets_num
|
||||
|
||||
|
||||
def find_best_benefit(choice):
|
||||
atomic_subsets_num = find_subsets_num(choice)
|
||||
total_benefit = 0
|
||||
for ind, obj in enumerate(WORKLOAD_INFO):
|
||||
# calculate the best benefit for the current sql
|
||||
max_benefit = 0
|
||||
for pos in atomic_subsets_num:
|
||||
if (obj.cost_list[0] - obj.cost_list[pos]) > max_benefit:
|
||||
max_benefit = obj.cost_list[0] - obj.cost_list[pos]
|
||||
total_benefit += max_benefit
|
||||
return total_benefit
|
||||
|
||||
|
||||
def get_diff(available_choices, choices):
|
||||
except_choices = copy.copy(available_choices)
|
||||
for i in available_choices:
|
||||
for j in choices:
|
||||
if is_same_index(i, j):
|
||||
except_choices.remove(i)
|
||||
return except_choices
|
||||
|
||||
|
||||
class State(object):
|
||||
"""
|
||||
The game state of the Monte Carlo tree search,
|
||||
the state data recorded under a certain Node node,
|
||||
including the current game score, the current number of game rounds,
|
||||
and the execution record from the beginning to the current.
|
||||
|
||||
It is necessary to realize whether the current state has reached the end of the game state,
|
||||
and support the operation of randomly fetching from the Action collection.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.current_storage = 0.0
|
||||
self.current_benefit = 0.0
|
||||
# record the sum of choices up to the current state
|
||||
self.accumulation_choices = []
|
||||
# record available choices of current state
|
||||
self.available_choices = []
|
||||
self.displayable_choices = []
|
||||
|
||||
def get_available_choices(self):
|
||||
return self.available_choices
|
||||
|
||||
def set_available_choices(self, choices):
|
||||
self.available_choices = choices
|
||||
|
||||
def get_current_storage(self):
|
||||
return self.current_storage
|
||||
|
||||
def set_current_storage(self, value):
|
||||
self.current_storage = value
|
||||
|
||||
def get_current_benefit(self):
|
||||
return self.current_benefit
|
||||
|
||||
def set_current_benefit(self, value):
|
||||
self.current_benefit = value
|
||||
|
||||
def get_accumulation_choices(self):
|
||||
return self.accumulation_choices
|
||||
|
||||
def set_accumulation_choices(self, choices):
|
||||
self.accumulation_choices = choices
|
||||
|
||||
def is_terminal(self):
|
||||
# the current node is a leaf node
|
||||
return len(self.accumulation_choices) == MAX_INDEX_NUM
|
||||
|
||||
def compute_benefit(self):
|
||||
return self.current_benefit
|
||||
|
||||
def get_next_state_with_random_choice(self):
|
||||
# ensure that the choices taken are not repeated
|
||||
if not self.available_choices:
|
||||
return None
|
||||
random_choice = random.choice([choice for choice in self.available_choices])
|
||||
self.available_choices.remove(random_choice)
|
||||
choice = copy.copy(self.accumulation_choices)
|
||||
choice.append(random_choice)
|
||||
benefit = find_best_benefit(choice)
|
||||
# if current choice not satisfy restrictions, then continue get next choice
|
||||
if benefit <= self.current_benefit or \
|
||||
self.current_storage + random_choice.storage > STORAGE_THRESHOLD:
|
||||
return self.get_next_state_with_random_choice()
|
||||
|
||||
next_state = State()
|
||||
# initialize the properties of the new state
|
||||
next_state.set_accumulation_choices(choice)
|
||||
next_state.set_current_benefit(benefit)
|
||||
next_state.set_current_storage(self.current_storage + random_choice.storage)
|
||||
next_state.set_available_choices(get_diff(AVAILABLE_CHOICES, choice))
|
||||
return next_state
|
||||
|
||||
def __repr__(self):
|
||||
self.displayable_choices = ['{}: {}'.format(choice.table, choice.columns)
|
||||
for choice in self.accumulation_choices]
|
||||
return "reward: {}, storage :{}, choices: {}".format(
|
||||
self.current_benefit, self.current_storage, self.displayable_choices)
|
||||
|
||||
|
||||
class Node(object):
|
||||
"""
|
||||
The Node of the Monte Carlo tree search tree contains the parent node and
|
||||
current point information,
|
||||
which is used to calculate the traversal times and quality value of the UCB,
|
||||
and the State of the Node selected by the game.
|
||||
"""
|
||||
def __init__(self):
|
||||
self.visit_number = 0
|
||||
self.quality = 0.0
|
||||
|
||||
self.parent = None
|
||||
self.children = []
|
||||
self.state = None
|
||||
|
||||
def get_parent(self):
|
||||
return self.parent
|
||||
|
||||
def set_parent(self, parent):
|
||||
self.parent = parent
|
||||
|
||||
def get_children(self):
|
||||
return self.children
|
||||
|
||||
def expand_child(self, node):
|
||||
node.set_parent(self)
|
||||
self.children.append(node)
|
||||
|
||||
def set_state(self, state):
|
||||
self.state = state
|
||||
|
||||
def get_state(self):
|
||||
return self.state
|
||||
|
||||
def get_visit_number(self):
|
||||
return self.visit_number
|
||||
|
||||
def set_visit_number(self, number):
|
||||
self.visit_number = number
|
||||
|
||||
def update_visit_number(self):
|
||||
self.visit_number += 1
|
||||
|
||||
def get_quality_value(self):
|
||||
return self.quality
|
||||
|
||||
def set_quality_value(self, value):
|
||||
self.quality = value
|
||||
|
||||
def update_quality_value(self, reward):
|
||||
self.quality += reward
|
||||
|
||||
def is_all_expand(self):
|
||||
return len(self.children) == \
|
||||
len(AVAILABLE_CHOICES) - len(self.get_state().get_accumulation_choices())
|
||||
|
||||
def __repr__(self):
|
||||
return "Node: {}, Q/N: {}/{}, State: {}".format(
|
||||
hash(self), self.quality, self.visit_number, self.state)
|
||||
|
||||
|
||||
def tree_policy(node):
|
||||
"""
|
||||
In the Selection and Expansion stages of Monte Carlo tree search,
|
||||
the node that needs to be searched (such as the root node) is passed in,
|
||||
and the best node that needs to be expanded is returned
|
||||
according to the exploration/exploitation algorithm.
|
||||
Note that if the node is a leaf node, it will be returned directly.
|
||||
|
||||
The basic strategy is to first find the child nodes that have not been selected at present,
|
||||
and select them randomly if there are more than one. If both are selected,
|
||||
find the one with the largest UCB value that has weighed exploration/exploitation,
|
||||
and randomly select if the UCB values are equal.
|
||||
"""
|
||||
|
||||
# check if the current node is leaf node
|
||||
while node and not node.get_state().is_terminal():
|
||||
|
||||
if node.is_all_expand():
|
||||
node = best_child(node, True)
|
||||
else:
|
||||
# return the new sub node
|
||||
sub_node = expand(node)
|
||||
# when there is no node that satisfies the condition in the remaining nodes,
|
||||
# this state is empty
|
||||
if sub_node.get_state():
|
||||
return sub_node
|
||||
|
||||
# return the leaf node
|
||||
return node
|
||||
|
||||
|
||||
def default_policy(node):
|
||||
"""
|
||||
In the Simulation stage of Monte Carlo tree search, input a node that needs to be expanded,
|
||||
create a new node after random operation, and return the reward of the new node.
|
||||
Note that the input node should not be a child node,
|
||||
and there are unexecuted Actions that can be expendable.
|
||||
|
||||
The basic strategy is to choose the Action at random.
|
||||
"""
|
||||
|
||||
# get the state of the game
|
||||
current_state = copy.deepcopy(node.get_state())
|
||||
|
||||
# run until the game over
|
||||
while not current_state.is_terminal():
|
||||
# pick one random action to play and get next state
|
||||
next_state = current_state.get_next_state_with_random_choice()
|
||||
if not next_state:
|
||||
break
|
||||
current_state = next_state
|
||||
|
||||
final_state_reward = current_state.compute_benefit()
|
||||
return final_state_reward
|
||||
|
||||
|
||||
def expand(node):
|
||||
"""
|
||||
Enter a node, expand a new node on the node, use the random method to execute the Action,
|
||||
and return the new node. Note that it is necessary to ensure that the newly
|
||||
added nodes are different from other node Action
|
||||
"""
|
||||
|
||||
new_state = node.get_state().get_next_state_with_random_choice()
|
||||
sub_node = Node()
|
||||
sub_node.set_state(new_state)
|
||||
node.expand_child(sub_node)
|
||||
|
||||
return sub_node
|
||||
|
||||
|
||||
def best_child(node, is_exploration):
|
||||
"""
|
||||
Using the UCB algorithm,
|
||||
select the child node with the highest score after weighing the exploration and exploitation.
|
||||
Note that if it is the prediction stage,
|
||||
the current Q-value score with the highest score is directly selected.
|
||||
"""
|
||||
|
||||
best_score = -sys.maxsize
|
||||
best_sub_node = None
|
||||
|
||||
# travel all sub nodes to find the best one
|
||||
for sub_node in node.get_children():
|
||||
# The children nodes of the node contains the children node whose state is empty,
|
||||
# this kind of node comes from the node that does not meet the conditions.
|
||||
if not sub_node.get_state():
|
||||
continue
|
||||
# ignore exploration for inference
|
||||
if is_exploration:
|
||||
C = 1 / math.sqrt(2.0)
|
||||
else:
|
||||
C = 0.0
|
||||
|
||||
# UCB = quality / times + C * sqrt(2 * ln(total_times) / times)
|
||||
left = sub_node.get_quality_value() / sub_node.get_visit_number()
|
||||
right = 2.0 * math.log(node.get_visit_number()) / sub_node.get_visit_number()
|
||||
score = left + C * math.sqrt(right)
|
||||
# get the maximum score, while filtering nodes that do not meet the space constraints and
|
||||
# nodes that have no revenue
|
||||
if score > best_score \
|
||||
and sub_node.get_state().get_current_storage() <= STORAGE_THRESHOLD \
|
||||
and sub_node.get_state().get_current_benefit() > 0:
|
||||
best_sub_node = sub_node
|
||||
best_score = score
|
||||
|
||||
return best_sub_node
|
||||
|
||||
|
||||
def backpropagate(node, reward):
|
||||
"""
|
||||
In the Backpropagation stage of Monte Carlo tree search,
|
||||
input the node that needs to be expended and the reward of the newly executed Action,
|
||||
feed it back to the expend node and all upstream nodes,
|
||||
and update the corresponding data.
|
||||
"""
|
||||
|
||||
# update util the root node
|
||||
while node is not None:
|
||||
# update the visit number
|
||||
node.update_visit_number()
|
||||
|
||||
# update the quality value
|
||||
node.update_quality_value(reward)
|
||||
|
||||
# change the node to the parent node
|
||||
node = node.parent
|
||||
|
||||
|
||||
def monte_carlo_tree_search(node):
|
||||
"""
|
||||
Implement the Monte Carlo tree search algorithm, pass in a root node,
|
||||
expand new nodes and update data according to the
|
||||
tree structure that has been explored before in a limited time,
|
||||
and then return as long as the child node with the highest exploitation.
|
||||
|
||||
When making predictions,
|
||||
you only need to select the node with the largest exploitation according to the Q value,
|
||||
and find the next optimal node.
|
||||
"""
|
||||
|
||||
computation_budget = len(AVAILABLE_CHOICES) * 3
|
||||
|
||||
# run as much as possible under the computation budget
|
||||
for i in range(computation_budget):
|
||||
# 1. find the best node to expand
|
||||
expand_node = tree_policy(node)
|
||||
if not expand_node:
|
||||
# when it is None, it means that all nodes are added but no nodes meet the space limit
|
||||
break
|
||||
# 2. random get next action and get reward
|
||||
reward = default_policy(expand_node)
|
||||
|
||||
# 3. update all passing nodes with reward
|
||||
backpropagate(expand_node, reward)
|
||||
|
||||
# get the best next node
|
||||
best_next_node = best_child(node, False)
|
||||
|
||||
return best_next_node
|
||||
|
||||
|
||||
def MCTS(workload_info, atomic_choices, available_choices, storage_threshold, max_index_num):
|
||||
global ATOMIC_CHOICES, STORAGE_THRESHOLD, WORKLOAD_INFO, AVAILABLE_CHOICES, MAX_INDEX_NUM
|
||||
WORKLOAD_INFO = workload_info
|
||||
AVAILABLE_CHOICES = available_choices
|
||||
ATOMIC_CHOICES = atomic_choices
|
||||
STORAGE_THRESHOLD = storage_threshold
|
||||
MAX_INDEX_NUM = max_index_num if max_index_num else len(available_choices)
|
||||
|
||||
# create the initialized state and initialized node
|
||||
init_state = State()
|
||||
choices = copy.copy(available_choices)
|
||||
init_state.set_available_choices(choices)
|
||||
init_node = Node()
|
||||
init_node.set_state(init_state)
|
||||
current_node = init_node
|
||||
|
||||
opt_config = []
|
||||
# set the rounds to play
|
||||
for i in range(len(AVAILABLE_CHOICES)):
|
||||
if current_node:
|
||||
current_node = monte_carlo_tree_search(current_node)
|
||||
if current_node:
|
||||
opt_config = current_node.state.accumulation_choices
|
||||
else:
|
||||
break
|
||||
return opt_config
|
||||
|
|
@ -27,7 +27,6 @@ from collections.abc import Iterable
|
|||
from collections import defaultdict
|
||||
|
||||
import index_advisor_workload as iaw
|
||||
import mcts
|
||||
|
||||
|
||||
def hash_any(obj):
|
||||
|
|
@ -228,32 +227,6 @@ select * from student_range_part1 where credit=1;
|
|||
|
||||
class IndexAdvisorTester(unittest.TestCase):
|
||||
|
||||
def test_mcts(self):
|
||||
storage_threshold = 12
|
||||
index1 = iaw.IndexItem('public.a', 'col1', index_type='global')
|
||||
index2 = iaw.IndexItem('public.b', 'col1', index_type='global')
|
||||
index3 = iaw.IndexItem('public.c', 'col1', index_type='global')
|
||||
index4 = iaw.IndexItem('public.d', 'col1', index_type='global')
|
||||
|
||||
atomic_index1 = iaw.IndexItem('public.a', 'col1', index_type='global')
|
||||
atomic_index2 = iaw.IndexItem('public.b', 'col1', index_type='global')
|
||||
atomic_index3 = iaw.IndexItem('public.c', 'col1', index_type='global')
|
||||
atomic_index4 = iaw.IndexItem('public.d', 'col1', index_type='global')
|
||||
|
||||
atomic_index1.storage = 10
|
||||
atomic_index2.storage = 4
|
||||
atomic_index3.storage = 7
|
||||
available_choices = [index1, index2, index3, index4]
|
||||
atomic_choices = [[], [atomic_index2], [atomic_index1], [atomic_index3],
|
||||
[atomic_index2, atomic_index3], [atomic_index4]]
|
||||
query = iaw.QueryItem('select * from gia_01', 1)
|
||||
query.cost_list = [10, 7, 5, 9, 4, 11]
|
||||
workload_info = [query]
|
||||
|
||||
results = mcts.MCTS(workload_info, atomic_choices, available_choices, storage_threshold, 2)
|
||||
self.assertLessEqual([index1.atomic_pos, index2.atomic_pos, index3.atomic_pos], [2, 1, 3])
|
||||
self.assertSetEqual({results[0].table, results[1].table}, {'public.b', 'public.c'})
|
||||
|
||||
def test_get_indexable_columns(self):
|
||||
tables = 'table1 table2 table2 table3 table3 table3'.split()
|
||||
columns = 'col1,col2 col2 col3 col1,col2 col2,col3 col2,col5'.split()
|
||||
|
|
|
|||
|
|
@ -827,6 +827,9 @@ void InitBSqlPluginHookIfNeeded()
|
|||
{
|
||||
const char* b_sql_plugin = "b_sql_plugin";
|
||||
CFunInfo tmpCF;
|
||||
if (!CheckIfExtensionExists(b_sql_plugin)) {
|
||||
return;
|
||||
}
|
||||
|
||||
tmpCF = load_external_function(b_sql_plugin, INIT_PLUGIN_OBJECT, false, false);
|
||||
if (tmpCF.user_fn != NULL) {
|
||||
|
|
@ -862,11 +865,9 @@ List* pg_parse_query(const char* query_string, List** query_string_locationlist)
|
|||
|
||||
List* (*parser_hook)(const char*, List**) = raw_parser;
|
||||
#ifndef ENABLE_MULTIPLE_NODES
|
||||
if (u_sess->attr.attr_sql.b_sql_plugin) {
|
||||
int id = GetCustomParserId();
|
||||
if (id >= 0 && g_instance.raw_parser_hook[id] != NULL) {
|
||||
parser_hook = (List* (*)(const char*, List**))g_instance.raw_parser_hook[id];
|
||||
}
|
||||
int id = GetCustomParserId();
|
||||
if (id >= 0 && g_instance.raw_parser_hook[id] != NULL) {
|
||||
parser_hook = (List* (*)(const char*, List**))g_instance.raw_parser_hook[id];
|
||||
}
|
||||
#endif
|
||||
raw_parsetree_list = parser_hook(query_string, query_string_locationlist);
|
||||
|
|
@ -6114,9 +6115,6 @@ void ProcessInterrupts(void)
|
|||
|
||||
/* The logical replication launcher can be stopped at any time. */
|
||||
proc_exit(0);
|
||||
} else if (IsLogicalWorker()) {
|
||||
ereport(FATAL, (errcode(ERRCODE_ADMIN_SHUTDOWN),
|
||||
errmsg("terminating logical replication worker due to administrator command")));
|
||||
#endif
|
||||
} else if (IsTxnSnapCapturerProcess()) {
|
||||
ereport(FATAL,
|
||||
|
|
@ -7574,7 +7572,7 @@ int PostgresMain(int argc, char* argv[], const char* dbname, const char* usernam
|
|||
init_set_params_htab();
|
||||
|
||||
#ifndef ENABLE_MULTIPLE_NODES
|
||||
if (u_sess->proc_cxt.MyDatabaseId != InvalidOid && DB_IS_CMPT(B_FORMAT) && u_sess->attr.attr_sql.b_sql_plugin) {
|
||||
if (u_sess->proc_cxt.MyDatabaseId != InvalidOid && DB_IS_CMPT(B_FORMAT)) {
|
||||
InitBSqlPluginHookIfNeeded();
|
||||
}
|
||||
#endif
|
||||
|
|
|
|||
|
|
@ -848,7 +848,7 @@ static bool InitSession(knl_session_context* session)
|
|||
t_thrd.proc_cxt.PostInit->InitSession();
|
||||
|
||||
#ifndef ENABLE_MULTIPLE_NODES
|
||||
if (u_sess->proc_cxt.MyDatabaseId != InvalidOid && DB_IS_CMPT(B_FORMAT) && u_sess->attr.attr_sql.b_sql_plugin) {
|
||||
if (u_sess->proc_cxt.MyDatabaseId != InvalidOid && DB_IS_CMPT(B_FORMAT)) {
|
||||
InitBSqlPluginHookIfNeeded();
|
||||
}
|
||||
#endif
|
||||
|
|
|
|||
|
|
@ -1278,10 +1278,13 @@ static const char *MXStatusToString(MultiXactStatus status)
|
|||
|
||||
static char *mxid_to_string(MultiXactId multi, int nmembers, MultiXactMember *members)
|
||||
{
|
||||
char *str = NULL;
|
||||
static char *str = NULL;
|
||||
StringInfoData buf;
|
||||
int i;
|
||||
|
||||
if (str != NULL)
|
||||
pfree(str);
|
||||
|
||||
initStringInfo(&buf);
|
||||
|
||||
appendStringInfo(&buf, XID_FMT " %d[" XID_FMT " (%s)", multi, nmembers,
|
||||
|
|
@ -1292,7 +1295,7 @@ static char *mxid_to_string(MultiXactId multi, int nmembers, MultiXactMember *me
|
|||
}
|
||||
|
||||
appendStringInfoChar(&buf, ']');
|
||||
str = MemoryContextStrdup(SESS_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE), buf.data);
|
||||
str = MemoryContextStrdup(TopMemoryContext, buf.data);
|
||||
pfree(buf.data);
|
||||
return str;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -448,7 +448,9 @@ static void RecycleWaitIfNotUsed()
|
|||
#endif
|
||||
) {
|
||||
uint64 nonRecycled = 0;
|
||||
while (true) {
|
||||
while (g_instance.attr.attr_storage.undo_zone_count == 0) {
|
||||
/* Clear any already-pending wakeups */
|
||||
ResetLatch(&t_thrd.proc->procLatch);
|
||||
if (t_thrd.undorecycler_cxt.shutdown_requested) {
|
||||
ereport(LOG, (errmodule(MOD_UNDO), errmsg(UNDOFORMAT("UndoRecycler: shutting down"))));
|
||||
ResourceOwnerRelease(t_thrd.utils_cxt.CurrentResourceOwner, RESOURCE_RELEASE_BEFORE_LOCKS, false, true);
|
||||
|
|
|
|||
|
|
@ -406,7 +406,7 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple
|
|||
pfree(outputbytes);
|
||||
}
|
||||
} else {
|
||||
char* outputstr = NULL;
|
||||
char *outputstr;
|
||||
pq_sendbyte(out, LOGICALREP_COLUMN_TEXT);
|
||||
if (!typclass->typbyval && typclass->typlen == -1) {
|
||||
/* definitely detoasted Datum */
|
||||
|
|
@ -464,8 +464,10 @@ static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
|
|||
len = pq_getmsgint(in, sizeof(uint32)); /* read length */
|
||||
|
||||
/* and data */
|
||||
value->data = (char *)palloc0((len + 1) * sizeof(char));
|
||||
value->data = (char *) palloc((len + 1) * sizeof(char));
|
||||
pq_copymsgbytes(in, value->data, len);
|
||||
/* not strictly necessary but per StringInfo practice */
|
||||
value->data[len] = '\0';
|
||||
/* make StringInfo fully valid */
|
||||
value->len = len;
|
||||
value->cursor = 0;
|
||||
|
|
|
|||
|
|
@ -94,7 +94,7 @@
|
|||
#include "utils/relcache.h"
|
||||
|
||||
static const int NAPTIME_PER_CYCLE = 10; /* max sleep time between cycles (10ms) */
|
||||
static const double HALF = 0.5;
|
||||
static const float HALF = 0.5;
|
||||
|
||||
typedef struct FlushPosition {
|
||||
dlist_node node;
|
||||
|
|
@ -121,6 +121,18 @@ static void LogicalrepWorkerSighub(SIGNAL_ARGS)
|
|||
t_thrd.applyworker_cxt.got_SIGHUP = true;
|
||||
}
|
||||
|
||||
/* SIGTERM: time to die */
|
||||
static void LogicalrepWorkerSigterm(SIGNAL_ARGS)
|
||||
{
|
||||
int saveErrno = errno;
|
||||
|
||||
t_thrd.applyworker_cxt.got_SIGTERM = true;
|
||||
if (t_thrd.proc)
|
||||
SetLatch(&t_thrd.proc->procLatch);
|
||||
|
||||
errno = saveErrno;
|
||||
}
|
||||
|
||||
/*
|
||||
* Make sure that we started local transaction.
|
||||
*
|
||||
|
|
@ -1056,15 +1068,13 @@ static void ApplyLoop(void)
|
|||
/* mark as idle, before starting to loop */
|
||||
pgstat_report_activity(STATE_IDLE, NULL);
|
||||
|
||||
for (;;) {
|
||||
while (!t_thrd.applyworker_cxt.got_SIGTERM) {
|
||||
MemoryContextSwitchTo(t_thrd.applyworker_cxt.messageContext);
|
||||
|
||||
int len;
|
||||
char *buf = NULL;
|
||||
unsigned char type;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/* Wait a while for data to arrive */
|
||||
if ((WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_receive(NAPTIME_PER_CYCLE, &type, &buf, &len)) {
|
||||
StringInfoData s;
|
||||
|
|
@ -1361,7 +1371,7 @@ void ApplyWorkerMain()
|
|||
*/
|
||||
gspqsignal(SIGHUP, LogicalrepWorkerSighub);
|
||||
gspqsignal(SIGINT, StatementCancelHandler);
|
||||
gspqsignal(SIGTERM, die);
|
||||
gspqsignal(SIGTERM, LogicalrepWorkerSigterm);
|
||||
|
||||
gspqsignal(SIGQUIT, quickdie);
|
||||
gspqsignal(SIGALRM, handle_sig_alarm);
|
||||
|
|
@ -1717,11 +1727,3 @@ static void UpdateConninfo(char* standbysInfo)
|
|||
|
||||
ereport(LOG, (errmsg("Update conninfo successfully, new conninfo %s.", standbysInfo)));
|
||||
}
|
||||
|
||||
/*
|
||||
* Is current process a logical replication worker?
|
||||
*/
|
||||
bool IsLogicalWorker(void)
|
||||
{
|
||||
return t_thrd.applyworker_cxt.curWorker != NULL;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -78,7 +78,7 @@ void _PG_output_plugin_init(OutputPluginCallbacks *cb)
|
|||
cb->shutdown_cb = pgoutput_shutdown;
|
||||
}
|
||||
|
||||
static void parse_output_parameters(List *options, PGOutputData *data)
|
||||
static void parse_output_parameters(List* options, PGOutputData* data)
|
||||
{
|
||||
ListCell *lc;
|
||||
bool protocol_version_given = false;
|
||||
|
|
|
|||
|
|
@ -5814,7 +5814,7 @@ Datum gs_paxos_stat_replication(PG_FUNCTION_ARGS)
|
|||
*/
|
||||
Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
|
||||
{
|
||||
#define PG_STAT_GET_WAL_SENDERS_COLS 21
|
||||
#define PG_STAT_GET_WAL_SENDERS_COLS 22
|
||||
|
||||
TupleDesc tupdesc;
|
||||
int *sync_priority = NULL;
|
||||
|
|
@ -6051,9 +6051,11 @@ Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
|
|||
/* sync_state and sync_prority */
|
||||
if (!SyncRepRequested()) {
|
||||
values[j++] = CStringGetTextDatum("Async");
|
||||
nulls[j++] = true;
|
||||
values[j++] = Int32GetDatum(0);
|
||||
} else {
|
||||
values[j++] = CStringGetTextDatum("Sync");
|
||||
nulls[j++] = true;
|
||||
values[j++] = Int32GetDatum(sync_priority[i]);
|
||||
}
|
||||
} else {
|
||||
|
|
@ -6075,12 +6077,15 @@ Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
|
|||
*/
|
||||
if (priority == 0) {
|
||||
values[j++] = CStringGetTextDatum("Async");
|
||||
nulls[j++] = true;
|
||||
} else if (list_member_int((List*)list_nth(sync_standbys, group), i)) {
|
||||
values[j++] = GetWalsndSyncRepConfig(walsnd)->syncrep_method == SYNC_REP_PRIORITY
|
||||
? CStringGetTextDatum("Sync")
|
||||
: CStringGetTextDatum("Quorum");
|
||||
values[j++] = Int32GetDatum(group);
|
||||
} else {
|
||||
values[j++] = CStringGetTextDatum("Potential");
|
||||
values[j++] = Int32GetDatum(group);
|
||||
}
|
||||
values[j++] = Int32GetDatum(priority);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,37 @@
|
|||
-- for any x group
|
||||
DO $$
|
||||
DECLARE
|
||||
ans boolean;
|
||||
BEGIN
|
||||
select case when working_version_num()=92301 then true else false end as ans into ans;
|
||||
if ans = false then
|
||||
DROP FUNCTION IF EXISTS pg_catalog.pg_stat_get_wal_senders() CASCADE;
|
||||
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 3099;
|
||||
CREATE FUNCTION pg_catalog.pg_stat_get_wal_senders(
|
||||
OUT pid bigint,
|
||||
OUT sender_pid integer,
|
||||
OUT local_role text,
|
||||
OUT peer_role text,
|
||||
OUT peer_state text,
|
||||
OUT state text,
|
||||
OUT catchup_start timestamp with time zone,
|
||||
OUT catchup_end timestamp with time zone,
|
||||
OUT sender_sent_location text,
|
||||
OUT sender_write_location text,
|
||||
OUT sender_flush_location text,
|
||||
OUT sender_replay_location text,
|
||||
OUT receiver_received_location text,
|
||||
OUT receiver_write_location text,
|
||||
OUT receiver_flush_location text,
|
||||
OUT receiver_replay_location text,
|
||||
OUT sync_percent text,
|
||||
OUT sync_state text,
|
||||
OUT sync_priority integer,
|
||||
OUT sync_most_available text,
|
||||
OUT channel text
|
||||
) RETURNS SETOF record
|
||||
STABLE NOT FENCED NOT SHIPPABLE ROWS 10
|
||||
LANGUAGE internal AS 'pg_stat_get_wal_senders';
|
||||
end if;
|
||||
END$$;
|
||||
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 0;
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
-- for any x group
|
||||
DO $$
|
||||
DECLARE
|
||||
ans boolean;
|
||||
BEGIN
|
||||
select case when working_version_num()=92301 then true else false end as ans into ans;
|
||||
if ans = false then
|
||||
DROP FUNCTION IF EXISTS pg_catalog.pg_stat_get_wal_senders() CASCADE;
|
||||
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 3099;
|
||||
CREATE FUNCTION pg_catalog.pg_stat_get_wal_senders(
|
||||
OUT pid bigint,
|
||||
OUT sender_pid integer,
|
||||
OUT local_role text,
|
||||
OUT peer_role text,
|
||||
OUT peer_state text,
|
||||
OUT state text,
|
||||
OUT catchup_start timestamp with time zone,
|
||||
OUT catchup_end timestamp with time zone,
|
||||
OUT sender_sent_location text,
|
||||
OUT sender_write_location text,
|
||||
OUT sender_flush_location text,
|
||||
OUT sender_replay_location text,
|
||||
OUT receiver_received_location text,
|
||||
OUT receiver_write_location text,
|
||||
OUT receiver_flush_location text,
|
||||
OUT receiver_replay_location text,
|
||||
OUT sync_percent text,
|
||||
OUT sync_state text,
|
||||
OUT sync_priority integer,
|
||||
OUT sync_most_available text,
|
||||
OUT channel text
|
||||
) RETURNS SETOF record
|
||||
STABLE NOT FENCED NOT SHIPPABLE ROWS 10
|
||||
LANGUAGE internal AS 'pg_stat_get_wal_senders';
|
||||
end if;
|
||||
END$$;
|
||||
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 0;
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
-- for any x group
|
||||
DO $$
|
||||
DECLARE
|
||||
ans boolean;
|
||||
BEGIN
|
||||
select case when working_version_num()=92301 then true else false end as ans into ans;
|
||||
if ans = false then
|
||||
DROP FUNCTION IF EXISTS pg_catalog.pg_stat_get_wal_senders() CASCADE;
|
||||
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 3099;
|
||||
CREATE FUNCTION pg_catalog.pg_stat_get_wal_senders(
|
||||
OUT pid bigint,
|
||||
OUT sender_pid integer,
|
||||
OUT local_role text,
|
||||
OUT peer_role text,
|
||||
OUT peer_state text,
|
||||
OUT state text,
|
||||
OUT catchup_start timestamp with time zone,
|
||||
OUT catchup_end timestamp with time zone,
|
||||
OUT sender_sent_location text,
|
||||
OUT sender_write_location text,
|
||||
OUT sender_flush_location text,
|
||||
OUT sender_replay_location text,
|
||||
OUT receiver_received_location text,
|
||||
OUT receiver_write_location text,
|
||||
OUT receiver_flush_location text,
|
||||
OUT receiver_replay_location text,
|
||||
OUT sync_percent text,
|
||||
OUT sync_state text,
|
||||
OUT sync_group integer,
|
||||
OUT sync_priority integer,
|
||||
OUT sync_most_available text,
|
||||
OUT channel text
|
||||
) RETURNS SETOF record
|
||||
STABLE NOT FENCED NOT SHIPPABLE ROWS 10
|
||||
LANGUAGE internal AS 'pg_stat_get_wal_senders';
|
||||
end if;
|
||||
END$$;
|
||||
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 0;
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
-- for any x group
|
||||
DO $$
|
||||
DECLARE
|
||||
ans boolean;
|
||||
BEGIN
|
||||
select case when working_version_num()=92301 then true else false end as ans into ans;
|
||||
if ans = false then
|
||||
DROP FUNCTION IF EXISTS pg_catalog.pg_stat_get_wal_senders() CASCADE;
|
||||
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 3099;
|
||||
CREATE FUNCTION pg_catalog.pg_stat_get_wal_senders(
|
||||
OUT pid bigint,
|
||||
OUT sender_pid integer,
|
||||
OUT local_role text,
|
||||
OUT peer_role text,
|
||||
OUT peer_state text,
|
||||
OUT state text,
|
||||
OUT catchup_start timestamp with time zone,
|
||||
OUT catchup_end timestamp with time zone,
|
||||
OUT sender_sent_location text,
|
||||
OUT sender_write_location text,
|
||||
OUT sender_flush_location text,
|
||||
OUT sender_replay_location text,
|
||||
OUT receiver_received_location text,
|
||||
OUT receiver_write_location text,
|
||||
OUT receiver_flush_location text,
|
||||
OUT receiver_replay_location text,
|
||||
OUT sync_percent text,
|
||||
OUT sync_state text,
|
||||
OUT sync_group integer,
|
||||
OUT sync_priority integer,
|
||||
OUT sync_most_available text,
|
||||
OUT channel text
|
||||
) RETURNS SETOF record
|
||||
STABLE NOT FENCED NOT SHIPPABLE ROWS 10
|
||||
LANGUAGE internal AS 'pg_stat_get_wal_senders';
|
||||
end if;
|
||||
END$$;
|
||||
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 0;
|
||||
|
|
@ -13,6 +13,5 @@
|
|||
#define LOGICALWORKER_H
|
||||
|
||||
extern void ApplyWorkerMain();
|
||||
extern bool IsLogicalWorker(void);
|
||||
|
||||
#endif /* LOGICALWORKER_H */
|
||||
|
|
|
|||
Loading…
Reference in New Issue