!1675 修复发布订阅死锁问题

Merge pull request !1675 from 薛蒙恩/pubsub_deadlock

src/bin/gs_guc/pg_guc.cpp

@@ -1229,6 +1229,12 @@ 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;

src/bin/pg_ctl/pg_ctl.cpp

@@ -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: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:C:dqL:T:Q:", long_options,
             &option_index)) != -1)
 #endif
 #endif

src/bin/pg_dump/pg_dump.cpp

@@ -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 = 92607;
+const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92606;
 
 #ifdef DUMPSYSLOG
 char* syslogpath = NULL;

src/bin/pg_probackup/backup.cpp

@@ -31,6 +31,9 @@
    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;
@@ -89,10 +92,11 @@ 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);
+static void do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool backup_logs,
+    bool backup_replslots);
 
 static void pg_start_backup(const char *label, bool smooth, pgBackup *backup,
-                                PGNodeInfo *nodeInfo, PGconn *conn);
+                                PGNodeInfo *nodeInfo, PGconn *conn, bool backup_replslots);
 static void pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn, PGNodeInfo *nodeInfo);
 static int checkpoint_timeout(PGconn *backup_conn);
 
@@ -558,7 +562,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)
+do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool backup_logs, bool backup_replslots)
 {
     int i;
     char    database_path[MAXPGPATH];
@@ -591,7 +595,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, &current, nodeInfo, backup_conn);
+    pg_start_backup(label, smooth_checkpoint, &current, nodeInfo, backup_conn, backup_replslots);
 
     /* Obtain current timeline */
 #if PG_VERSION_NUM >= 90600
@@ -624,10 +628,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);
+                            true, true, false, backup_logs, true, 0, backup_replslots);
     else
         dir_list_file(backup_files_list, instance_config.pgdata,
-                             true, true, false, backup_logs, true, 0, FIO_LOCAL_HOST);
+                             true, true, false, backup_logs, true, 0, FIO_LOCAL_HOST, backup_replslots);
 
     /*
      * Get database_map (name to oid) for use in partial restore feature.
@@ -749,6 +753,11 @@ 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)
     {
@@ -849,7 +858,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 no_validate, bool no_sync, bool backup_logs, bool backup_replslots)
 {
     PGconn  *backup_conn = NULL;
     PGNodeInfo  nodeInfo;
@@ -925,7 +934,7 @@ do_backup(time_t start_time, pgSetBackupParams *set_backup_params,
         add_note(&current, set_backup_params->note);
 
     /* backup data */
-    do_backup_instance(backup_conn, &nodeInfo, no_sync, backup_logs);
+    do_backup_instance(backup_conn, &nodeInfo, no_sync, backup_logs, backup_replslots);
     pgut_atexit_pop(backup_cleanup, NULL);
 
     /* compute size of wal files of this backup stored in the archive */
@@ -1034,13 +1043,15 @@ 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)
+                PGNodeInfo *nodeInfo, PGconn *conn, bool backup_replslots)
 {
     PGresult   *res;
     const char *params[2];
     uint32  lsn_hi;
     uint32  lsn_lo;
     int     ret;
+    int i;
+    XLogRecPtr startLsn;
 
     params[0] = label;
 
@@ -1068,7 +1079,33 @@ 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 */
-    backup->start_lsn = ((uint64) lsn_hi )<< 32 | lsn_lo;
+    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;
 
     PQclear(res);
 }

src/bin/pg_probackup/dir.cpp

@@ -42,13 +42,6 @@ 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",
 
@@ -68,7 +61,7 @@ const char *pgdata_exclude_dir[] =
     (const char *)"pg_subtrans",
 
     /* end of list */
-    NULL,   /* pg_log will be set later */
+    NULL,   /* pg_log and pg_replslot will be set later */
     NULL
 };
 
@@ -128,17 +121,20 @@ 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);
+static char dir_check_file(pgFile *file, bool backup_logs, bool backup_replslots);
 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 skip_hidden, int external_dir_num, fio_location location,
+                                                    bool backup_replslots);
 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 */
@@ -538,7 +534,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)
+                        fio_location location, bool backup_replslots)
 {
     pgFile  *file;
 
@@ -565,7 +561,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_logs, skip_hidden, external_dir_num, location, backup_replslots);
 
     if (!add_root)
         pgFileFree(file);
@@ -589,7 +585,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)
+dir_check_file(pgFile *file, bool backup_logs, bool backup_replslots)
 {
     int     i;
     int     sscanf_res;
@@ -652,6 +648,29 @@ dir_check_file(pgFile *file, bool backup_logs)
             }
         }
 
+        /*
+         * 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;
@@ -749,6 +768,35 @@ 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;
@@ -889,7 +937,8 @@ 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 skip_hidden, int external_dir_num, fio_location location,
+                                        bool backup_replslots)
 {
     DIR     *dir;
     struct dirent *dent;
@@ -937,7 +986,7 @@ dir_list_file_internal(parray *files, pgFile *parent, const char *parent_dir,
 
         if (exclude)
         {
-            check_res = dir_check_file(file, backup_logs);
+            check_res = dir_check_file(file, backup_logs, backup_replslots);
             if (check_res == CHECK_FALSE)
             {
                 /* Skip */
@@ -963,7 +1012,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_logs, skip_hidden, external_dir_num, location, backup_replslots);
     }
 
     if (errno && errno != ENOENT)

src/bin/pg_probackup/file.cpp

@@ -51,6 +51,7 @@ typedef struct
     bool exclusive_backup;
     bool skip_hidden;
     int  external_dir_num;
+    bool backup_replslots;
 } fio_list_dir_request;
 
 typedef struct
@@ -1794,7 +1795,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 skip_hidden, int external_dir_num, bool backup_replslots)
 {
     fio_header hdr;
     fio_list_dir_request req;
@@ -1811,6 +1812,7 @@ 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);
@@ -1870,7 +1872,14 @@ 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);
         }
@@ -1914,7 +1923,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->external_dir_num, FIO_LOCAL_HOST, req->backup_replslots);
 
     /* send information about files to the main process */
     for (i = 0; i < (int)parray_num(file_files); i++)

src/bin/pg_probackup/file.h

@@ -163,5 +163,7 @@ 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
 

src/bin/pg_probackup/help.cpp

@@ -154,6 +154,7 @@ 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);
@@ -420,6 +421,7 @@ 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"));
@@ -441,6 +443,7 @@ 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"));

src/bin/pg_probackup/pg_probackup.cpp

@@ -77,6 +77,7 @@ 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;
@@ -186,6 +187,7 @@ 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 },
@@ -550,7 +552,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);
+                return do_backup(start_time, set_backup_params, no_validate, no_sync, backup_logs, backup_replslots);
             }
         case RESTORE_CMD:
             return do_restore_or_validate(current.backup_id,

src/bin/pg_probackup/pg_probackupa.h

@@ -69,6 +69,7 @@ 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

src/bin/pg_probackup/pg_probackupc.h

@@ -54,6 +54,9 @@ 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;
 
@@ -89,7 +92,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 no_validate, bool no_sync, bool backup_logs, bool backup_replslots);
 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,
@@ -239,7 +242,8 @@ 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 skip_hidden, int external_dir_num, fio_location location,
+                          bool backup_replslots = false);
 
 extern void create_data_directories(parray *dest_files,
                                         const char *data_dir,
@@ -432,7 +436,8 @@ 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 add_root, bool backup_logs, bool skip_hidden, int external_dir_num,
+                         bool backup_replslots = false);
 
 extern bool pgut_rmtree(const char *path, bool rmtopdir, bool strict);
 

src/common/backend/catalog/builtin_funcs.ini

@@ -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(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))
+        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))
     ),
     AddFuncGroup(
         "pg_stat_get_wlm_ec_operator_info", 1, 

src/common/backend/catalog/pg_subscription.cpp

@@ -68,7 +68,7 @@ Subscription *GetSubscription(Oid subid, bool missing_ok)
 
     /* Get slotname */
     datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, Anum_pg_subscription_subslotname, &isnull);
-    if (unlikely(isnull)) {
+    if (!isnull) {
         sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
     } else {
         sub->slotname = NULL;

src/common/backend/utils/init/globals.cpp

@@ -59,7 +59,7 @@ bool open_join_children = true;
 bool will_shutdown = false;
 
 /* hard-wired binary version number */
-const uint32 GRAND_VERSION_NUM = 92607;
+const uint32 GRAND_VERSION_NUM = 92606;
 
 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 = 92607;
+const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92606;
 
 /* Version number of the guc parameter backend_version added in V500R001C20 */
 const uint32 V5R1C20_BACKEND_VERSION_NUM = 92305;

src/gausskernel/dbmind/tools/components/index_advisor/index_advisor_workload.py

@@ -26,9 +26,11 @@ 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
@@ -192,9 +194,12 @@ class IndexAdvisor:
                                                     self.workload_used_index))
         if DRIVER:
             self.db.close_conn()
-
-        opt_config = greedy_determine_opt_config(self.workload_info[0], atomic_config_total,
-                                                 candidate_indexes, self.index_cost_total[0])
+        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])
         self.retain_lower_cost_index(candidate_indexes)
         if len(opt_config) == 0:
             print("No optimal indexes generated!")
@@ -943,7 +948,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 int value" %
+        raise argparse.ArgumentTypeError("%s is an invalid positive float value" %
                                          args.max_index_storage)
     JSON_TYPE = args.json
     MAX_INDEX_NUM = args.max_index_num
@@ -971,7 +976,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=int)
+                            help="Maximum storage of suggested indexes/MB", type=float)
     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',

src/gausskernel/dbmind/tools/components/index_advisor/mcts.py

@@ -0,0 +1,397 @@
+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

src/gausskernel/dbmind/tools/tests/test_index_advisor_workload.py

@@ -27,6 +27,7 @@ from collections.abc import Iterable
 from collections import defaultdict
 
 import index_advisor_workload as iaw
+import mcts
 
 
 def hash_any(obj):
@@ -227,6 +228,32 @@ 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()

src/gausskernel/process/tcop/postgres.cpp

@@ -827,9 +827,6 @@ 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) {
@@ -865,9 +862,11 @@ List* pg_parse_query(const char* query_string, List** query_string_locationlist)
 
     List* (*parser_hook)(const char*, List**) = raw_parser;
 #ifndef ENABLE_MULTIPLE_NODES
-    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];
+    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];
+        }
     }
 #endif
     raw_parsetree_list = parser_hook(query_string, query_string_locationlist);
@@ -6115,6 +6114,9 @@ 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,
@@ -7572,7 +7574,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)) {
+    if (u_sess->proc_cxt.MyDatabaseId != InvalidOid && DB_IS_CMPT(B_FORMAT) && u_sess->attr.attr_sql.b_sql_plugin) {
         InitBSqlPluginHookIfNeeded();
     }
 #endif

src/gausskernel/process/threadpool/threadpool_worker.cpp

@@ -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)) {
+    if (u_sess->proc_cxt.MyDatabaseId != InvalidOid && DB_IS_CMPT(B_FORMAT) && u_sess->attr.attr_sql.b_sql_plugin) {
         InitBSqlPluginHookIfNeeded();
     }
 #endif

src/gausskernel/storage/access/transam/multixact.cpp

@@ -1278,13 +1278,10 @@ static const char *MXStatusToString(MultiXactStatus status)
 
 static char *mxid_to_string(MultiXactId multi, int nmembers, MultiXactMember *members)
 {
-    static char *str = NULL;
+    char *str = NULL;
     StringInfoData buf;
     int i;
 
-    if (str != NULL)
-        pfree(str);
-
     initStringInfo(&buf);
 
     appendStringInfo(&buf, XID_FMT " %d[" XID_FMT " (%s)", multi, nmembers,
@@ -1295,7 +1292,7 @@ static char *mxid_to_string(MultiXactId multi, int nmembers, MultiXactMember *me
     }
 
     appendStringInfoChar(&buf, ']');
-    str = MemoryContextStrdup(TopMemoryContext, buf.data);
+    str = MemoryContextStrdup(SESS_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE), buf.data);
     pfree(buf.data);
     return str;
 }

src/gausskernel/storage/replication/logical/proto.cpp

@@ -406,7 +406,7 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple
                 pfree(outputbytes);
             }
         } else {
-            char *outputstr;
+            char* outputstr = NULL;
             pq_sendbyte(out, LOGICALREP_COLUMN_TEXT);
             if (!typclass->typbyval && typclass->typlen == -1) {
                 /* definitely detoasted Datum */
@@ -464,10 +464,8 @@ static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
                 len = pq_getmsgint(in, sizeof(uint32));    /* read length */
 
                 /* and data */
-                value->data = (char *) palloc((len + 1) * sizeof(char));
+                value->data = (char *)palloc0((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;

src/gausskernel/storage/replication/logical/worker.cpp

@@ -94,7 +94,7 @@
 #include "utils/relcache.h"
 
 static const int NAPTIME_PER_CYCLE = 10; /* max sleep time between cycles (10ms) */
-static const float HALF = 0.5;
+static const double HALF = 0.5;
 
 typedef struct FlushPosition {
     dlist_node node;
@@ -121,18 +121,6 @@ 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.
  *
@@ -1068,13 +1056,15 @@ static void ApplyLoop(void)
     /* mark as idle, before starting to loop */
     pgstat_report_activity(STATE_IDLE, NULL);
 
-    while (!t_thrd.applyworker_cxt.got_SIGTERM) {
+    for (;;) {
         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;
@@ -1371,7 +1361,7 @@ void ApplyWorkerMain()
      */
     gspqsignal(SIGHUP, LogicalrepWorkerSighub);
     gspqsignal(SIGINT, StatementCancelHandler);
-    gspqsignal(SIGTERM, LogicalrepWorkerSigterm);
+    gspqsignal(SIGTERM, die);
 
     gspqsignal(SIGQUIT, quickdie);
     gspqsignal(SIGALRM, handle_sig_alarm);
@@ -1727,3 +1717,11 @@ 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;
+}

src/gausskernel/storage/replication/pgoutput/pgoutput.cpp

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

src/gausskernel/storage/replication/walsender.cpp

@@ -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 22
+#define PG_STAT_GET_WAL_SENDERS_COLS 21
 
     TupleDesc tupdesc;
     int *sync_priority = NULL;
@@ -6051,11 +6051,9 @@ 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 {
@@ -6077,15 +6075,12 @@ 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);
             }

src/include/catalog/upgrade_sql/rollback_catalog_maindb/rollback_catalog_maindb_92_606.sql

@@ -1,37 +0,0 @@
--- 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;

src/include/catalog/upgrade_sql/rollback_catalog_otherdb/rollback_catalog_otherdb_92_606.sql

@@ -1,37 +0,0 @@
--- 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;

src/include/catalog/upgrade_sql/upgrade_catalog_maindb/upgrade-post_catalog_maindb_92_606.sql

@@ -1,38 +0,0 @@
--- 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;

src/include/catalog/upgrade_sql/upgrade_catalog_otherdb/upgrade-post_catalog_otherdb_92_606.sql

@@ -1,38 +0,0 @@
--- 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;

src/include/replication/logicalworker.h

@@ -13,5 +13,6 @@
 #define LOGICALWORKER_H
 
 extern void ApplyWorkerMain();
+extern bool IsLogicalWorker(void);
 
 #endif /* LOGICALWORKER_H */