diff --git a/src/bin/pg_dump/pg_dump.cpp b/src/bin/pg_dump/pg_dump.cpp
index 7e0a942a8..e734947b5 100644
--- a/src/bin/pg_dump/pg_dump.cpp
+++ b/src/bin/pg_dump/pg_dump.cpp
@@ -233,6 +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;
 
 #ifdef DUMPSYSLOG
 char* syslogpath = NULL;
@@ -4444,7 +4445,9 @@ void getSubscriptions(Archive *fout)
     int i_subslotname;
     int i_subsynccommit;
     int i_subpublications;
-    int i, ntups;
+    int i_subbinary;
+    int i;
+    int ntups;
 
     if (no_subscriptions || GetVersionNum(fout) < SUBSCRIPTION_VERSION) {
         return;
@@ -4469,14 +4472,20 @@ void getSubscriptions(Archive *fout)
     resetPQExpBuffer(query);
 
     /* Get the subscriptions in current database. */
-    appendPQExpBuffer(query,
-        "SELECT s.tableoid, s.oid, s.subname,"
-        "(%s s.subowner) AS rolname, "
-        " s.subconninfo, s.subslotname, s.subsynccommit, s.subpublications "
-        "FROM pg_catalog.pg_subscription s "
+    appendPQExpBuffer(query, "SELECT s.tableoid, s.oid, s.subname,"
+        "(%s s.subowner) AS rolname, s.subconninfo, s.subslotname, "
+        "s.subsynccommit, s.subpublications, \n", username_subquery);
+
+    if (GetVersionNum(fout) >= SUBSCRIPTION_BINARY_VERSION_NUM) {
+        appendPQExpBuffer(query, " s.subbinary\n");
+    } else {
+        appendPQExpBuffer(query, " false AS subbinary\n");
+    }
+
+    appendPQExpBuffer(query, "FROM pg_catalog.pg_subscription s "
         "WHERE s.subdbid = (SELECT oid FROM pg_catalog.pg_database"
-        "                   WHERE datname = current_database())",
-        username_subquery);
+        "                   WHERE datname = current_database())");
+
     res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
 
     ntups = PQntuples(res);
@@ -4494,6 +4503,7 @@ void getSubscriptions(Archive *fout)
     i_subslotname = PQfnumber(res, "subslotname");
     i_subsynccommit = PQfnumber(res, "subsynccommit");
     i_subpublications = PQfnumber(res, "subpublications");
+    i_subbinary = PQfnumber(res, "subbinary");
 
     subinfo = (SubscriptionInfo *)pg_malloc(ntups * sizeof(SubscriptionInfo));
 
@@ -4512,6 +4522,7 @@ void getSubscriptions(Archive *fout)
         }
         subinfo[i].subsynccommit = gs_strdup(PQgetvalue(res, i, i_subsynccommit));
         subinfo[i].subpublications = gs_strdup(PQgetvalue(res, i, i_subpublications));
+        subinfo[i].subbinary = gs_strdup(PQgetvalue(res, i, i_subbinary));
 
         if (strlen(subinfo[i].rolname) == 0) {
             write_msg(NULL, "WARNING: owner of subscription \"%s\" appears to be invalid\n", subinfo[i].dobj.name);
@@ -4578,6 +4589,10 @@ static void dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
         appendPQExpBufferStr(query, "NONE");
     }
 
+    if (strcmp(subinfo->subbinary, "t") == 0) {
+        appendPQExpBuffer(query, ", binary = true");
+    }
+
     if (strcmp(subinfo->subsynccommit, "off") != 0) {
         appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
     }
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 4c7fdbc93..a705192cc 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -498,6 +498,7 @@ typedef struct _SubscriptionInfo {
     char *subslotname;
     char *subsynccommit;
     char *subpublications;
+    char *subbinary;
 } SubscriptionInfo;
 
 /* global decls */
diff --git a/src/common/backend/catalog/pg_subscription.cpp b/src/common/backend/catalog/pg_subscription.cpp
index 3d1d6833e..ce7eaf981 100644
--- a/src/common/backend/catalog/pg_subscription.cpp
+++ b/src/common/backend/catalog/pg_subscription.cpp
@@ -28,7 +28,6 @@
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
 #include "utils/syscache.h"
-#include "replication/worker_internal.h"
 
 static List *textarray_to_stringlist(ArrayType *textarray);
 
@@ -91,6 +90,13 @@ Subscription *GetSubscription(Oid subid, bool missing_ok)
     }
     sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum));
 
+    datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, Anum_pg_subscription_subbinary, &isnull);
+    if (unlikely(isnull)) {
+        ereport(ERROR, (errcode(ERRCODE_UNEXPECTED_NULL_VALUE),
+            errmsg("null binary for subscription %u", subid)));
+    }
+    sub->binary = DatumGetBool(datum);
+
     ReleaseSysCache(tup);
 
     return sub;
@@ -183,7 +189,7 @@ char *get_subscription_name(Oid subid, bool missing_ok)
 }
 
 /* Clear the list content, only deal with DefElem and string content */
-static void ClearListContent(List *list)
+void ClearListContent(List *list)
 {
     ListCell *cell = NULL;
     foreach(cell, list) {
@@ -203,25 +209,6 @@ static void ClearListContent(List *list)
     }
 }
 
-/*
- * Decrypt conninfo for subscription.
- * IMPORTANT: caller should clear and free the memory after using it immediately
- */
-char *DecryptConninfo(char *encryptConninfo)
-{
-    const char* sensitiveOptionsArray[] = {"password"};
-    const int sensitiveArrayLength = lengthof(sensitiveOptionsArray);
-    List *defList = ConninfoToDefList(encryptConninfo);
-    DecryptOptions(defList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE);
-    char *decryptConninfo = DefListToString(defList);
-
-    /* defList has plain content, clear it before free */
-    ClearListContent(defList);
-    list_free_ext(defList);
-    /* IMPORTANT: caller should clear and free the memory after using it immediately */
-    return decryptConninfo;
-}
-
 /*
  * Convert text array to list of strings.
  *
diff --git a/src/common/backend/utils/init/globals.cpp b/src/common/backend/utils/init/globals.cpp
index a8329d9ea..a147cdfc8 100644
--- a/src/common/backend/utils/init/globals.cpp
+++ b/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 = 92605;
+const uint32 GRAND_VERSION_NUM = 92606;
 
 const uint32 PREDPUSH_SAME_LEVEL_VERSION_NUM = 92522;
 const uint32 UPSERT_WHERE_VERSION_NUM = 92514;
@@ -101,6 +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;
 
 /* Version number of the guc parameter backend_version added in V500R001C20 */
 const uint32 V5R1C20_BACKEND_VERSION_NUM = 92305;
diff --git a/src/gausskernel/optimizer/commands/subscriptioncmds.cpp b/src/gausskernel/optimizer/commands/subscriptioncmds.cpp
index b40214c01..656ffbe8a 100644
--- a/src/gausskernel/optimizer/commands/subscriptioncmds.cpp
+++ b/src/gausskernel/optimizer/commands/subscriptioncmds.cpp
@@ -44,7 +44,7 @@
 #include "utils/array.h"
 #include "utils/acl.h"
 
-static void ConnectPublisher(char *conninfo, char* slotname);
+static bool ConnectPublisher(char* conninfo, char* slotname);
 static void CreateSlotInPublisher(char *slotname);
 static void ValidateReplicationSlot(char *slotname, List *publications);
 
@@ -56,7 +56,7 @@ static void ValidateReplicationSlot(char *slotname, List *publications);
  * accommodate that.
  */
 static void parse_subscription_options(const List *options, char **conninfo, List **publications, bool *enabled_given,
-    bool *enabled, bool *slot_name_given, char **slot_name, char **synchronous_commit)
+    bool *enabled, bool *slot_name_given, char **slot_name, char **synchronous_commit, bool *binary_given, bool *binary)
 {
     ListCell *lc;
 
@@ -76,6 +76,10 @@ static void parse_subscription_options(const List *options, char **conninfo, Lis
     if (synchronous_commit) {
         *synchronous_commit = NULL;
     }
+    if (binary) {
+        *binary_given = false;
+        *binary = false;
+    }
 
     /* Parse options */
     foreach (lc, options) {
@@ -124,6 +128,15 @@ static void parse_subscription_options(const List *options, char **conninfo, Lis
             /* Test if the given value is valid for synchronous_commit GUC. */
             (void)set_config_option("synchronous_commit", *synchronous_commit, PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET,
                 false, 0, false);
+        } else if (strcmp(defel->defname, "binary") == 0 && binary) {
+            if (*binary_given) {
+                ereport(ERROR,
+                        (errcode(ERRCODE_SYNTAX_ERROR),
+                         errmsg("conflicting or redundant options")));
+            }
+
+            *binary_given = true;
+            *binary = defGetBoolean(defel);
         } else {
             ereport(ERROR,
                 (errcode(ERRCODE_SYNTAX_ERROR), errmsg("unrecognized subscription parameter: %s", defel->defname)));
@@ -197,26 +210,82 @@ static Datum publicationListToArray(List *publist)
 }
 
 /*
- * connect publisher and create slot.
- * the input conninfo should be encrypt, we will decrypt password inside
+ * Parse the original connection string which is encrypted, poll all hosts and ports,
+ * and try to connect to the publisher.
+ * When checkRemoteMode is true, the remotemode must be normal or primary.
+ * Return true to indicate successful connection.
  */
-static void ConnectPublisher(char *conninfo, char *slotname)
+bool AttemptConnectPublisher(const char *conninfoOriginal, char* slotname, bool checkRemoteMode)
+{
+    size_t conninfoLen = strlen(conninfoOriginal) + 1;
+
+    char* conninfo = NULL;
+    StringInfoData conninfoWithoutHostport;
+    initStringInfo(&conninfoWithoutHostport);
+    HostPort* hostPortList[MAX_REPLNODE_NUM] = {NULL};
+    ParseConninfo(conninfoOriginal, &conninfoWithoutHostport, hostPortList);
+    if (hostPortList[0] == NULL) {
+        ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg(
+            "invalid connection string syntax, missing host and port")));
+    }
+    bool connectSuccess = false;
+    conninfo = (char*)palloc(conninfoLen * sizeof(char));
+    for (int i = 0; i < MAX_REPLNODE_NUM; ++i) {
+        if (hostPortList[i] == NULL) {
+            break;
+        }
+        int ret = snprintf_s(conninfo, conninfoLen, conninfoLen - 1,
+            "%s host=%s port=%s", conninfoWithoutHostport.data,
+            hostPortList[i]->host, hostPortList[i]->port);
+        securec_check_ss(ret, "\0", "\0");
+
+        connectSuccess = ConnectPublisher(conninfo, slotname);
+        if (!connectSuccess) {
+            /* try next host */
+            continue;
+        }
+        if (!checkRemoteMode) {
+            break;
+        }
+        ServerMode publisherServerMde = IdentifyRemoteMode();
+        if (publisherServerMde == NORMAL_MODE || publisherServerMde == PRIMARY_MODE) {
+            break;
+        }
+        /* it's a standby, try next host */
+        (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_disconnect();
+        connectSuccess = false;
+    }
+    pfree_ext(conninfo);
+
+    /* clean up */
+    FreeStringInfo(&conninfoWithoutHostport);
+    for (int i = 0; i < MAX_REPLNODE_NUM; ++i) {
+        if (hostPortList[i] == NULL) {
+            break;
+        }
+        pfree_ext(hostPortList[i]->host);
+        pfree_ext(hostPortList[i]->port);
+        pfree_ext(hostPortList[i]);
+    }
+    return connectSuccess;
+}
+
+/*
+ * connect to publisher with conninfo
+ */
+static bool ConnectPublisher(char* conninfo, char* slotname)
 {
     /* Try to connect to the publisher. */
     volatile WalRcvData *walrcv = t_thrd.walreceiverfuncs_cxt.WalRcv;
     SpinLockAcquire(&walrcv->mutex);
     walrcv->conn_target = REPCONNTARGET_PUBLICATION;
     SpinLockRelease(&walrcv->mutex);
-
-    char *decryptConninfo = DecryptConninfo(conninfo);
+    char* decryptConninfo = EncryptOrDecryptConninfo(conninfo, 'D');
     bool connectSuccess = (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_connect(decryptConninfo, NULL, slotname, -1);
     int rc = memset_s(decryptConninfo, strlen(decryptConninfo), 0, strlen(decryptConninfo));
     securec_check(rc, "", "");
     pfree_ext(decryptConninfo);
-
-    if (!connectSuccess) {
-        ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("could not connect to the publisher")));
-    }
+    return connectSuccess;
 }
 
 /*
@@ -293,9 +362,10 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
     bool enabled_given = false;
     bool enabled = true;
     char *synchronous_commit;
-    char *conninfo;
     char *slotname;
     bool slotname_given;
+    bool binary;
+    bool binary_given;
     char originname[NAMEDATALEN];
     List *publications;
     int rc;
@@ -305,7 +375,7 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
      * Connection and publication should not be specified here.
      */
     parse_subscription_options(stmt->options, NULL, NULL, &enabled_given, &enabled, &slotname_given, &slotname,
-        &synchronous_commit);
+        &synchronous_commit, &binary_given, &binary);
 
     /*
      * Since creating a replication slot is not transactional, rolling back
@@ -333,11 +403,10 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
         synchronous_commit = "off";
     }
 
-    conninfo = stmt->conninfo;
     publications = stmt->publication;
 
     /* Check the connection info string. */
-    libpqrcv_check_conninfo(conninfo);
+    libpqrcv_check_conninfo(stmt->conninfo);
 
     /* Everything ok, form a new tuple. */
     rc = memset_s(values, sizeof(values), 0, sizeof(values));
@@ -349,18 +418,12 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
     values[Anum_pg_subscription_subname - 1] = DirectFunctionCall1(namein, CStringGetDatum(stmt->subname));
     values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(owner);
     values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled);
+    values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(binary);
 
     /* encrypt conninfo */
-    List *conninfoList = ConninfoToDefList(stmt->conninfo);
-    /* Sensitive options for subscription, will be encrypted when saved to catalog. */
-    const char* sensitiveOptionsArray[] = {"password"};
-    const int sensitiveArrayLength = lengthof(sensitiveOptionsArray);
-    EncryptGenericOptions(conninfoList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE);
-    char *encryptConninfo = DefListToString(conninfoList);
-
+    char *encryptConninfo = EncryptOrDecryptConninfo(stmt->conninfo, 'E');
     values[Anum_pg_subscription_subconninfo - 1] = CStringGetTextDatum(encryptConninfo);
 
-    pfree_ext(conninfoList);
     if (enabled) {
         if (!slotname_given) {
             slotname = stmt->subname;
@@ -396,11 +459,14 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
      */
     if (enabled) {
         Assert(slotname);
-        ConnectPublisher(encryptConninfo, slotname);
+
+        if (!AttemptConnectPublisher(encryptConninfo, slotname, true)) {
+            ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("Failed to connect to publisher.")));
+        }
+
         CreateSlotInPublisher(slotname);
         (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_disconnect();
     }
-
     pfree_ext(encryptConninfo);
     heap_close(rel, RowExclusiveLock);
     rc = memset_s(stmt->conninfo, strlen(stmt->conninfo), 0, strlen(stmt->conninfo));
@@ -439,6 +505,8 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt)
     Oid subid;
     bool enabled_given = false;
     bool enabled;
+    bool binary_given;
+    bool binary;
     char *synchronous_commit;
     char *conninfo;
     char *slot_name;
@@ -473,7 +541,7 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt)
 
     /* Parse options. */
     parse_subscription_options(stmt->options, &conninfo, &publications, &enabled_given, &enabled, &slotname_given,
-        &slot_name, &synchronous_commit);
+        &slot_name, &synchronous_commit, &binary_given, &binary);
 
     /* Form a new tuple. */
     rc = memset_s(nulls, sizeof(nulls), false, sizeof(nulls));
@@ -490,23 +558,15 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt)
     if (conninfo) {
         /* Check the connection info string. */
         libpqrcv_check_conninfo(conninfo);
-
-        /* encrypt conninfo */
-        List *conninfoList = ConninfoToDefList(conninfo);
-        /* Sensitive options for subscription, will be encrypted when saved to catalog. */
-        const char* sensitiveOptionsArray[] = {"password"};
-        const int sensitiveArrayLength = lengthof(sensitiveOptionsArray);
-        EncryptGenericOptions(conninfoList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE);
-        encryptConninfo = DefListToString(conninfoList);
-        needFreeConninfo = true;
-
+        encryptConninfo = EncryptOrDecryptConninfo(conninfo, 'E');
+        rc = memset_s(conninfo, strlen(conninfo), 0, strlen(conninfo));
+        securec_check(rc, "\0", "\0");
         values[Anum_pg_subscription_subconninfo - 1] = CStringGetTextDatum(encryptConninfo);
         replaces[Anum_pg_subscription_subconninfo - 1] = true;
+        needFreeConninfo = true;
 
-        pfree_ext(conninfoList);
-
+        /* need to check whether new conninfo can be used to connect to new publisher */
         if (sub->enabled || (enabled_given && enabled)) {
-            /* we need to check whether new conninfo can be used to connect to new publisher */
             checkConn = true;
         }
     }
@@ -548,6 +608,10 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt)
         values[Anum_pg_subscription_subsynccommit - 1] = CStringGetTextDatum(synchronous_commit);
         replaces[Anum_pg_subscription_subsynccommit - 1] = true;
     }
+    if (binary_given) {
+        values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(binary);
+        replaces[Anum_pg_subscription_subbinary - 1] = true;
+    }
     if (publications != NIL) {
         values[Anum_pg_subscription_subpublications - 1] = publicationListToArray(publications);
         replaces[Anum_pg_subscription_subpublications - 1] = true;
@@ -570,16 +634,18 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt)
     if (sub->enabled && !enabled) {
         ereport(ERROR, (errmsg("If you want to deactivate this subscription, use DROP SUBSCRIPTION.")));
     }
-    /* enable subscription */
-    if (!sub->enabled && enabled) {
-        /* if slot hasn't been created, then create it */
-        if (!sub->slotname || !*(sub->slotname)) {
+    /* enabling subscription, but slot hasn't been created,
+     * then mark createSlot to true.
+     */
+    if (!sub->enabled && enabled && (!sub->slotname || !*(sub->slotname))) {
             createSlot = true;
-        }
     }
 
     if (checkConn || createSlot || validateSlot) {
-        ConnectPublisher(encryptConninfo, finalSlotName);
+        if (!AttemptConnectPublisher(encryptConninfo, finalSlotName, true)) {
+            ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg(
+                checkConn ? "The new conninfo cannot connect to new publisher." : "Failed to connect to publisher.")));
+        }
 
         if (createSlot) {
             CreateSlotInPublisher(finalSlotName);
@@ -597,12 +663,6 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt)
     if (needFreeConninfo) {
         pfree_ext(encryptConninfo);
     }
-
-    if (conninfo) {
-        rc = memset_s(conninfo, strlen(conninfo), 0, strlen(conninfo));
-        securec_check(rc, "", "");
-    }
-
     return myself;
 }
 
@@ -753,7 +813,11 @@ void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
     initStringInfo(&cmd);
     appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s", quote_identifier(slotname));
 
-    ConnectPublisher(conninfo, slotname);
+    if (!AttemptConnectPublisher(conninfo, slotname, true)) {
+        ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg(
+            "could not connect to publisher.")));
+    }
+
     PG_TRY();
     {
         int sqlstate = 0;
@@ -779,6 +843,7 @@ void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 
     (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_disconnect();
 
+    pfree_ext(conninfo);
     pfree(cmd.data);
     heap_close(rel, NoLock);
 }
@@ -908,3 +973,149 @@ void RenameSubscription(List *oldname, const char *newname)
 
     return;
 }
+
+/*
+ * Parse the host or port string into a string array,
+ * where host and port are separated by ",".
+ * input: conn --- host or port string separated by ","
+ * output: connArray --- host or port string array
+ * return: the length of connArray
+ * for example:
+ * (1):
+ * conn = 1.1.1.1,2.2.2.2,...,9.9.9.9
+ * connArray = {
+ *              1,.1.1.1,
+ *              2.2.2.2,
+ *              ...,
+ *              9.9.9.9
+ *              }
+ * return 9
+ * (2):
+ * conn = 1,2,...,9
+ * connArray = {1,2,...,9}
+ * return 9
+ */
+static int HostsPortsToArray(const char* conn, char** connArray)
+{
+    if (conn == NULL) {
+        return 0;
+    }
+    char* cp = NULL;
+    char* cur = NULL;
+    char *buf = pstrdup(conn);
+
+    cp = buf;
+    int i = 0;
+    while (*cp) {
+        cur = cp;
+        while (*cp && *cp != ',') {
+            ++cp;
+        }
+        if (*cp == ',') {
+            *cp = '\0';
+            ++cp;
+        }
+        if (i >= MAX_REPLNODE_NUM) {
+            ereport(ERROR, (errmsg("Currently, a maximum of %d servers are "
+                "supported.", MAX_REPLNODE_NUM)));
+        }
+        connArray[i++] = pstrdup(cur);
+
+        if (*cp == 0) {
+            break;
+        }
+    }
+    pfree(buf);
+    return i;
+}
+
+/*
+ * parse host and port
+ */
+static void ParseHostPort(char* hoststr, char* portstr, HostPort** hostPortList)
+{
+    char* hosts[MAX_REPLNODE_NUM] = {NULL};
+    char* ports[MAX_REPLNODE_NUM] = {NULL};
+    int hostNum = HostsPortsToArray(hoststr, hosts);
+    int portNum = HostsPortsToArray(portstr, ports);
+    if (hostNum != portNum) {
+        ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("The number of host and port are inconsistent.")));
+    }
+
+    for (int i = 0; i < hostNum; ++i) {
+        hostPortList[i] = (HostPort*)palloc(sizeof(HostPort));
+        hostPortList[i]->host = hosts[i];
+        hostPortList[i]->port = ports[i];
+    }
+}
+
+/*
+ * Parse conninfo
+ * conninfo format:
+ *      'dbname=abc user=username password=xxxx host=ip1,ip2,...,ip9 port=p1,p2,...,p9'
+ * after parsing:
+ * conninfoWithoutHostPort:
+ *      'dbname=abc user=username password=xxxx'
+ * hostPortList:
+ *      {
+ *          {host=ip1, port=p1},
+ *          {host=ip2, port=p2},
+ *          ...
+ *          {host=ip9, port=p9}
+ *      }
+ */
+void ParseConninfo(const char* conninfo, StringInfoData* conninfoWithoutHostPort, HostPort** hostPortList)
+{
+    List* conninfoList = ConninfoToDefList(conninfo);
+    ListCell* l = NULL;
+
+    char* hostStr = NULL;
+    char* portStr = NULL;
+    foreach (l, conninfoList) {
+        DefElem* defel = (DefElem*)lfirst(l);
+        if (pg_strcasecmp(defel->defname, "host") == 0) {
+            hostStr = defGetString(defel);
+        } else if (pg_strcasecmp(defel->defname, "port") == 0) {
+            portStr = defGetString(defel);
+        } else {
+            appendStringInfo(conninfoWithoutHostPort, "%s=%s ", defel->defname, defGetString(defel));
+        }
+    }
+    if (hostPortList != NULL) {
+        ParseHostPort(hostStr, portStr, hostPortList);
+    }
+}
+
+/*
+ * encrypt conninfo when action = 'E'
+ * decrypt conninfo when action = 'D'
+ * conninfoNew: encrypted or decrypted conninfo
+ */
+char* EncryptOrDecryptConninfo(const char* conninfo, const char action)
+{
+    /* parse conninfo to list */
+    List *conninfoList = ConninfoToDefList(conninfo);
+    /* Sensitive options for subscription */
+    const char* sensitiveOptionsArray[] = {"password"};
+    const int sensitiveArrayLength = lengthof(sensitiveOptionsArray);
+    switch (action) {
+        /* Encrypt */
+        case 'E':
+            EncryptGenericOptions(conninfoList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE);
+            break;
+
+        /* Decrypt */
+        case 'D':
+            DecryptOptions(conninfoList, sensitiveOptionsArray, sensitiveArrayLength, SUBSCRIPTION_MODE);
+            break;
+
+        default:
+            break;
+    }
+
+    char* conninfoNew = DefListToString(conninfoList);
+    ClearListContent(conninfoList);
+    list_free_ext(conninfoList);
+
+    return conninfoNew;
+}
diff --git a/src/gausskernel/storage/replication/libpqwalreceiver.cpp b/src/gausskernel/storage/replication/libpqwalreceiver.cpp
index 41bab8a79..30a338a1a 100755
--- a/src/gausskernel/storage/replication/libpqwalreceiver.cpp
+++ b/src/gausskernel/storage/replication/libpqwalreceiver.cpp
@@ -255,8 +255,14 @@ void StartRemoteStreaming(const LibpqrcvConnectParam *options)
             stringlist_to_identifierstr(t_thrd.libwalreceiver_cxt.streamConn, options->publicationNames);
         appendStringInfo(&cmd, ", publication_names %s",
             PQescapeLiteral(t_thrd.libwalreceiver_cxt.streamConn, pubnames_str, strlen(pubnames_str)));
-        appendStringInfoChar(&cmd, ')');
         pfree(pubnames_str);
+
+        if (options->binary && PQserverVersion(t_thrd.libwalreceiver_cxt.streamConn) >= 90204) {
+            appendStringInfoString(&cmd, ", binary 'true'");
+            ereport(DEBUG5, (errmsg("append binary true")));
+        }
+
+        appendStringInfoChar(&cmd, ')');
     }
 
     PGresult *res = libpqrcv_PQexec(cmd.data);
@@ -420,7 +426,7 @@ void IdentifyRemoteSystem(bool checkRemote)
 }
 
 /* identify remote mode, should do this after connect success. */
-static ServerMode IdentifyRemoteMode()
+ServerMode IdentifyRemoteMode()
 {
     Assert(t_thrd.libwalreceiver_cxt.streamConn != NULL);
     volatile WalRcvData *walrcv = t_thrd.walreceiverfuncs_cxt.WalRcv;
@@ -443,7 +449,9 @@ static ServerMode IdentifyRemoteMode()
                                   num_fields)));
     }
     remoteMode = (ServerMode)pg_strtoint32(PQgetvalue(res, 0, 0));
-    if (!t_thrd.walreceiver_cxt.AmWalReceiverForFailover && (!IS_PRIMARY_NORMAL(remoteMode)) &&
+    if (walrcv->conn_target != REPCONNTARGET_PUBLICATION &&
+        !t_thrd.walreceiver_cxt.AmWalReceiverForFailover &&
+        (!IS_PRIMARY_NORMAL(remoteMode)) &&
         /* remoteMode of cascade standby is a standby */
         !t_thrd.xlog_cxt.is_cascade_standby && !IS_SHARED_STORAGE_MODE) {
         PQclear(res);
diff --git a/src/gausskernel/storage/replication/logical/proto.cpp b/src/gausskernel/storage/replication/logical/proto.cpp
index 5387fe2a7..2c537dd4c 100644
--- a/src/gausskernel/storage/replication/logical/proto.cpp
+++ b/src/gausskernel/storage/replication/logical/proto.cpp
@@ -18,7 +18,6 @@
 #include "catalog/pg_type.h"
 #include "libpq/pqformat.h"
 #include "replication/logicalproto.h"
-#include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
 
@@ -28,7 +27,7 @@
 static const int LOGICALREP_IS_REPLICA_IDENTITY = 1;
 
 static void logicalrep_write_attrs(StringInfo out, Relation rel);
-static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple);
+static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binary);
 
 static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel);
 static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple);
@@ -115,7 +114,7 @@ void logicalrep_write_origin(StringInfo out, const char *origin, XLogRecPtr orig
 /*
  * Write INSERT to the output stream.
  */
-void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple)
+void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple, bool binary)
 {
     pq_sendbyte(out, 'I'); /* action INSERT */
 
@@ -123,7 +122,7 @@ void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple)
     pq_sendint32(out, RelationGetRelid(rel));
 
     pq_sendbyte(out, 'N'); /* new tuple follows */
-    logicalrep_write_tuple(out, rel, newtuple);
+    logicalrep_write_tuple(out, rel, newtuple, binary);
 }
 
 /*
@@ -151,7 +150,7 @@ LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtu
 /*
  * Write UPDATE to the output stream.
  */
-void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple)
+void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple, bool binary)
 {
     pq_sendbyte(out, 'U'); /* action UPDATE */
 
@@ -166,11 +165,11 @@ void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, H
             pq_sendbyte(out, 'O'); /* old tuple follows */
         else
             pq_sendbyte(out, 'K'); /* old key follows */
-        logicalrep_write_tuple(out, rel, oldtuple);
+        logicalrep_write_tuple(out, rel, oldtuple, binary);
     }
 
     pq_sendbyte(out, 'N'); /* new tuple follows */
-    logicalrep_write_tuple(out, rel, newtuple);
+    logicalrep_write_tuple(out, rel, newtuple, binary);
 }
 
 /*
@@ -213,7 +212,7 @@ LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, Logica
 /*
  * Write DELETE to the output stream.
  */
-void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple)
+void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple, bool binary)
 {
     char relreplident = RelationGetRelReplident(rel);
     Assert(relreplident == REPLICA_IDENTITY_DEFAULT ||
@@ -229,7 +228,7 @@ void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple)
     else
         pq_sendbyte(out, 'K'); /* old key follows */
 
-    logicalrep_write_tuple(out, rel, oldtuple);
+    logicalrep_write_tuple(out, rel, oldtuple, binary);
 }
 
 /*
@@ -344,7 +343,7 @@ void logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp)
 /*
  * Write a tuple to the outputstream, in the most efficient format possible.
  */
-static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple)
+static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binary)
 {
     TupleDesc desc;
     Datum values[MaxTupleAttributeNumber];
@@ -371,7 +370,6 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple
         HeapTuple typtup;
         Form_pg_type typclass;
         Form_pg_attribute att = desc->attrs[i];
-        char *outputstr;
 
         /* skip dropped columns */
         if (att->attisdropped || GetGeneratedCol(desc, i)) {
@@ -379,7 +377,7 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple
         }
 
         if (isnull[i]) {
-            pq_sendbyte(out, 'n'); /* null column */
+            pq_sendbyte(out, LOGICALREP_COLUMN_NULL); /* null column */
             continue;
         }
 
@@ -388,61 +386,91 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple
             elog(ERROR, "cache lookup failed for type %u", att->atttypid);
         typclass = (Form_pg_type)GETSTRUCT(typtup);
 
-        pq_sendbyte(out, 't'); /* 'text' data follows */
-        if (!typclass->typbyval && typclass->typlen == -1) {
-            /* definitely detoasted Datum */
-            Datum val = PointerGetDatum(PG_DETOAST_DATUM(values[i]));
-            outputstr = OidOutputFunctionCall(typclass->typoutput, val);
+        /*
+        * Send in binary if requested and type has suitable send function.
+        */
+        if (binary && OidIsValid(typclass->typsend)) {
+            bytea* outputbytes = NULL;
+            pq_sendbyte(out, LOGICALREP_COLUMN_BINARY);
+            if (!typclass->typbyval && typclass->typlen == -1) {
+                /* definitely detoasted Datum */
+                Datum val = PointerGetDatum(PG_DETOAST_DATUM(values[i]));
+                outputbytes = OidSendFunctionCall(typclass->typsend, val);
+            } else {
+                outputbytes = OidSendFunctionCall(typclass->typsend, values[i]);
+            }
+            int len = VARSIZE(outputbytes) - VARHDRSZ;
+            pq_sendint(out, len, 4);    /* length */
+            pq_sendbytes(out, VARDATA(outputbytes), len);    /* data */
+            if (outputbytes != NULL) {
+                pfree(outputbytes);
+            }
         } else {
-            outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]);
+            char* outputstr = NULL;
+            pq_sendbyte(out, LOGICALREP_COLUMN_TEXT);
+            if (!typclass->typbyval && typclass->typlen == -1) {
+                /* definitely detoasted Datum */
+                Datum val = PointerGetDatum(PG_DETOAST_DATUM(values[i]));
+                outputstr = OidOutputFunctionCall(typclass->typoutput, val);
+            } else {
+                outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]);
+            }
+            pq_sendcountedtext(out, outputstr, strlen(outputstr), false);
+            if (outputstr != NULL) {
+                pfree(outputstr);
+            }
         }
-        pq_sendcountedtext(out, outputstr, strlen(outputstr), false);
-        pfree(outputstr);
         ReleaseSysCache(typtup);
     }
 }
 
 /*
- * Read tuple in remote format from stream.
- *
- * The returned tuple points into the input stringinfo.
+ * Read tuple in logical replication format from stream.
  */
 static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
 {
-    uint16 i;
-    uint16 natts;
-    int rc;
-
     /* Get number of attributes. */
-    natts = pq_getmsgint(in, sizeof(uint16));
+    uint16 natts = pq_getmsgint(in, sizeof(uint16));
 
-    rc = memset_s(tuple->changed, sizeof(tuple->changed), 0, sizeof(tuple->changed));
-    securec_check(rc, "", "");
+    /* Allocate space for per-column values; zero out unused StringInfoDatas */
+    tuple->colvalues = (StringInfoData *) palloc0(natts * sizeof(StringInfoData));
+    tuple->colstatus = (char *) palloc(natts * sizeof(char));
     tuple->ncols = natts;
 
     /* Read the data */
-    for (i = 0; i < natts; i++) {
-        char kind;
-
-        kind = pq_getmsgbyte(in);
+    for (uint16 i = 0; i < natts; i++) {
+        char kind = pq_getmsgbyte(in);
+        tuple->colstatus[i] = kind;
+        uint32 len;
+        StringInfo value = &tuple->colvalues[i];
 
         switch (kind) {
-            case 'n': /* null */
-                tuple->values[i] = NULL;
-                tuple->changed[i] = true;
+            case LOGICALREP_COLUMN_NULL: /* null */
+                /* nothing more to do */
                 break;
-            case 't': { /* text formatted value */
-                uint32 len;
-                tuple->changed[i] = true;
-
-                len = pq_getmsgint(in, sizeof(uint32)); /* read length */
+            case LOGICALREP_COLUMN_TEXT:
+                len = pq_getmsgint(in, sizeof(uint32));    /* read length */
 
                 /* and data */
-                tuple->values[i] = (char *)palloc(len + 1);
-                pq_copymsgbytes(in, tuple->values[i], len);
-                tuple->values[i][len] = '\0';
+                value->data = (char *) palloc((len + 1) * sizeof(char));
+                pq_copymsgbytes(in, value->data, len);
+                value->data[len] = '\0';
+                /* make StringInfo fully valid */
+                value->len = len;
+                value->cursor = 0;
+                value->maxlen = len;
+                break;
+            case LOGICALREP_COLUMN_BINARY:
+                len = pq_getmsgint(in, sizeof(uint32));    /* read length */
+
+                /* and data */
+                value->data = (char *)palloc0((len + 1) * sizeof(char));
+                pq_copymsgbytes(in, value->data, len);
+                /* make StringInfo fully valid */
+                value->len = len;
+                value->cursor = 0;
+                value->maxlen = len;
                 break;
-            }
             default:
                 elog(ERROR, "unrecognized data representation type '%c'", kind);
                 break;
@@ -451,7 +479,7 @@ static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
 }
 
 /*
- * Write relation attributes to the stream.
+ * Write relation attributes metadata to the stream.
  */
 static void logicalrep_write_attrs(StringInfo out, Relation rel)
 {
@@ -504,7 +532,7 @@ static void logicalrep_write_attrs(StringInfo out, Relation rel)
 }
 
 /*
- * Read relation attribute names from the stream.
+ * Read relation attribute metadata from the stream.
  */
 static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel)
 {
@@ -573,3 +601,25 @@ static const char *logicalrep_read_namespace(StringInfo in)
 
     return nspname;
 }
+
+/*
+ * Write conninfo to the output stream.
+ */
+void logicalrep_write_conninfo(StringInfo out, char* conninfo)
+{
+    pq_sendbyte(out, 'S'); /* action */
+
+    pq_writestring(out, conninfo); /* conninfo follows */
+}
+
+/*
+ * Read conninfo from stream.
+ */
+void logicalrep_read_conninfo(StringInfo in, char** conninfo)
+{
+    const char* conninfoTemp = pq_getmsgstring(in);
+    size_t conninfoLen = strlen(conninfoTemp) + 1;
+    *conninfo = (char*)palloc(conninfoLen);
+    int rc = strcpy_s(*conninfo, conninfoLen, conninfoTemp);
+    securec_check(rc, "", "");
+}
diff --git a/src/gausskernel/storage/replication/logical/worker.cpp b/src/gausskernel/storage/replication/logical/worker.cpp
index 0079eda34..f91d90ef3 100644
--- a/src/gausskernel/storage/replication/logical/worker.cpp
+++ b/src/gausskernel/storage/replication/logical/worker.cpp
@@ -41,6 +41,7 @@
 #include "catalog/pg_partition_fn.h"
 
 #include "commands/trigger.h"
+#include "commands/subscriptioncmds.h"
 
 #include "executor/executor.h"
 #include "executor/node/nodeModifyTable.h"
@@ -111,6 +112,8 @@ static void store_flush_position(XLogRecPtr remote_lsn);
 static void reread_subscription(void);
 static void ApplyWorkerProcessMsg(char type, StringInfo s, XLogRecPtr *lastRcv);
 static void apply_dispatch(StringInfo s);
+static void apply_handle_conninfo(StringInfo s);
+static void UpdateConninfo(char* standbysInfo);
 
 /* SIGHUP: set flag to re-read config file at next convenient time */
 static void LogicalrepWorkerSighub(SIGNAL_ARGS)
@@ -265,11 +268,11 @@ static void slot_store_error_callback(void *arg)
 }
 
 /*
- * Store data in C string form into slot.
- * This is similar to BuildTupleFromCStrings but TupleTableSlot fits our
- * use better.
+ * Store tuple data into slot.
+ *
+ * Incoming data can be either text or binary format.
  */
-static void slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, char **values)
+static void slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, LogicalRepTupleData *tupleData)
 {
     int natts = slot->tts_tupleDescriptor->natts;
     int i;
@@ -286,19 +289,52 @@ static void slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel
     errcallback.previous = t_thrd.log_cxt.error_context_stack;
     t_thrd.log_cxt.error_context_stack = &errcallback;
 
-    /* Call the "in" function for each non-dropped attribute */
+    /* Call the "in" function for each non-dropped, non-null attribute */
     for (i = 0; i < natts; i++) {
         Form_pg_attribute att = slot->tts_tupleDescriptor->attrs[i];
         int remoteattnum = rel->attrmap[i];
-        Oid typinput;
-        Oid typioparam;
 
-        if (!att->attisdropped && remoteattnum >= 0 && values[remoteattnum] != NULL) {
+        if (!att->attisdropped && remoteattnum >= 0) {
+            StringInfo colvalue = &tupleData->colvalues[remoteattnum];
             errarg.remote_attnum = remoteattnum;
 
-            getTypeInputInfo(att->atttypid, &typinput, &typioparam);
-            slot->tts_values[i] = OidInputFunctionCall(typinput, values[remoteattnum], typioparam, att->atttypmod);
-            slot->tts_isnull[i] = false;
+            if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_TEXT) {
+                Oid typinput;
+                Oid typioparam;
+
+                getTypeInputInfo(att->atttypid, &typinput, &typioparam);
+                slot->tts_values[i] = OidInputFunctionCall(typinput, colvalue->data, typioparam, att->atttypmod);
+                slot->tts_isnull[i] = false;
+            } else if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_BINARY) {
+                Oid typreceive;
+                Oid typioparam;
+
+                /*
+                 * In some code paths we may be asked to re-parse the same
+                 * tuple data.  Reset the StringInfo's cursor so that works.
+                 */
+                colvalue->cursor = 0;
+
+                getTypeBinaryInputInfo(att->atttypid, &typreceive, &typioparam);
+                slot->tts_values[i] = OidReceiveFunctionCall(typreceive, colvalue, typioparam, att->atttypmod);
+
+                /* Trouble if it didn't eat the whole buffer */
+                if (colvalue->cursor != colvalue->len) {
+                    ereport(ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
+                        errmsg("incorrect binary data format in logical replication column %d", 
+                        remoteattnum + 1)));
+                }
+                slot->tts_isnull[i] = false;
+            } else {
+                /*
+                 * NULL value from remote.  (We don't expect to see
+                 * LOGICALREP_COLUMN_UNCHANGED here, but if we do, treat it as
+                 * NULL.)
+                 */
+                slot->tts_values[i] = (Datum) 0;
+                slot->tts_isnull[i] = true;
+            }
+            /* Reset attnum for error callback */
             errarg.remote_attnum = -1;
         } else {
             /*
@@ -318,18 +354,19 @@ static void slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel
 }
 
 /*
- * Replace selected columns with user data provided as C strings.
+ * Replace updated columns with data from the LogicalRepTupleData struct.
  * This is somewhat similar to heap_modify_tuple but also calls the type
  * input functions on the user data.
- * "slot" is filled with a copy of the tuple in "srcslot", with
- * columns selected by the "replaces" array replaced with data values
- * from "values".
+ *
+ * "slot" is filled with a copy of the tuple in "srcslot", replacing
+ * columns provided in "tupleData" and leaving others as-is.
+ *
  * Caution: unreplaced pass-by-ref columns in "slot" will point into the
  * storage for "srcslot".  This is OK for current usage, but someday we may
  * need to materialize "slot" at the end to make it independent of "srcslot".
  */
-static void slot_modify_cstrings(TupleTableSlot *slot, TupleTableSlot *srcslot, LogicalRepRelMapEntry *rel,
-    char **values, const bool *replaces)
+static void slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot, LogicalRepRelMapEntry *rel,
+    LogicalRepTupleData *tupleData)
 {
     int natts = slot->tts_tupleDescriptor->natts;
     int i;
@@ -364,23 +401,47 @@ static void slot_modify_cstrings(TupleTableSlot *slot, TupleTableSlot *srcslot,
         Form_pg_attribute att = slot->tts_tupleDescriptor->attrs[i];
         int remoteattnum = rel->attrmap[i];
 
-        if (remoteattnum < 0 || !replaces[remoteattnum]) {
+        if (remoteattnum < 0) {
             continue;
         }
 
-        if (values[remoteattnum] != NULL) {
-            Oid typinput;
-            Oid typioparam;
-
+        if (tupleData->colstatus[remoteattnum] != LOGICALREP_COLUMN_UNCHANGED) {
+            StringInfo colvalue = &tupleData->colvalues[remoteattnum];
             errarg.remote_attnum = remoteattnum;
 
-            getTypeInputInfo(att->atttypid, &typinput, &typioparam);
-            slot->tts_values[i] = OidInputFunctionCall(typinput, values[remoteattnum], typioparam, att->atttypmod);
-            slot->tts_isnull[i] = false;
+            if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_TEXT) {
+                Oid typinput;
+                Oid typioparam;
+
+                getTypeInputInfo(att->atttypid, &typinput, &typioparam);
+                slot->tts_values[i] = OidInputFunctionCall(typinput, colvalue->data, typioparam, att->atttypmod);
+                slot->tts_isnull[i] = false;
+            } else if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_BINARY) {
+                Oid typreceive;
+                Oid typioparam;
+
+                /*
+                 * In some code paths we may be asked to re-parse the same
+                 * tuple data.  Reset the StringInfo's cursor so that works.
+                 */
+                colvalue->cursor = 0;
+
+                getTypeBinaryInputInfo(att->atttypid, &typreceive, &typioparam);
+                slot->tts_values[i] = OidReceiveFunctionCall(typreceive, colvalue, typioparam, att->atttypmod);
+
+                /* Trouble if it didn't eat the whole buffer */
+                if (colvalue->cursor != colvalue->len) {
+                    ereport(ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
+                        errmsg("incorrect binary data format in logical replication column %d", remoteattnum + 1)));
+                }
+                slot->tts_isnull[i] = false;
+            } else {
+                /* must be LOGICALREP_COLUMN_NULL */
+                slot->tts_values[i] = (Datum) 0;
+                slot->tts_isnull[i] = true;
+            }
+            
             errarg.remote_attnum = -1;
-        } else {
-            slot->tts_values[i] = (Datum)0;
-            slot->tts_isnull[i] = true;
         }
     }
 
@@ -516,7 +577,7 @@ static void apply_handle_insert(StringInfo s)
     PushActiveSnapshot(GetTransactionSnapshot());
     /* Process and store remote tuple in the slot */
     oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
-    slot_store_cstrings(remoteslot, rel, newtup.values);
+    slot_store_data(remoteslot, rel, &newtup);
     slot_fill_defaults(rel, estate, remoteslot);
     MemoryContextSwitchTo(oldctx);
 
@@ -646,7 +707,7 @@ static void apply_handle_update(StringInfo s)
         int remoteattnum = rel->attrmap[i];
         if (!att->attisdropped && remoteattnum >= 0) {
             Assert(remoteattnum < newtup.ncols);
-            if (newtup.changed[i]) {
+            if (newtup.colstatus[i] != LOGICALREP_COLUMN_UNCHANGED) {
                 target_rte->updatedCols = bms_add_member(target_rte->updatedCols,
                                                      i + 1 - FirstLowInvalidHeapAttributeNumber);
             }
@@ -660,7 +721,7 @@ static void apply_handle_update(StringInfo s)
 
     /* Build the search tuple. */
     oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
-    slot_store_cstrings(remoteslot, rel, has_oldtup ? oldtup.values : newtup.values);
+    slot_store_data(remoteslot, rel, has_oldtup ? &oldtup : &newtup);
     MemoryContextSwitchTo(oldctx);
 
     /*
@@ -683,7 +744,7 @@ static void apply_handle_update(StringInfo s)
     if (found) {
         /* Process and store remote tuple in the slot */
         oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
-        slot_modify_cstrings(remoteslot, localslot, rel, newtup.values, newtup.changed);
+        slot_modify_data(remoteslot, localslot, rel, &newtup);
         MemoryContextSwitchTo(oldctx);
 
         EvalPlanQualSetSlot(&epqstate, remoteslot);
@@ -748,7 +809,7 @@ static void apply_handle_delete(StringInfo s)
 
     /* Find the tuple using the replica identity index. */
     oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
-    slot_store_cstrings(remoteslot, rel, oldtup.values);
+    slot_store_data(remoteslot, rel, &oldtup);
     MemoryContextSwitchTo(oldctx);
 
     /*
@@ -824,6 +885,9 @@ static void apply_dispatch(StringInfo s)
         case 'O':
             apply_handle_origin(s);
             break;
+        case 'S':
+            apply_handle_conninfo(s);
+            break;
         default:
             ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("invalid logical replication message type \"%c\"", action)));
@@ -1224,6 +1288,17 @@ static void reread_subscription(void)
         proc_exit(0);
     }
 
+    /*
+    * Exit if any parameter that affects the remote connection was changed.
+    * The launcher will start a new worker.
+    */
+    if (strcmp(newsub->name, t_thrd.applyworker_cxt.mySubscription->name) != 0 ||
+        newsub->binary != t_thrd.applyworker_cxt.mySubscription->binary) {
+            ereport(LOG, (errmsg("logical replication apply worker for subscription \"%s\" "
+                "will restart because of a parameter change", t_thrd.applyworker_cxt.mySubscription->name)));
+            proc_exit(0);
+    }
+
     /* !slotname should never happen when enabled is true. */
     Assert(newsub->slotname);
     
@@ -1422,14 +1497,9 @@ void ApplyWorkerMain()
 
     CommitTransactionCommand();
 
-    char *decryptConnInfo = DecryptConninfo(t_thrd.applyworker_cxt.mySubscription->conninfo);
-    bool connectSuccess = (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_connect(decryptConnInfo, NULL,
-        t_thrd.applyworker_cxt.mySubscription->name, -1);
-    rc = memset_s(decryptConnInfo, strlen(decryptConnInfo), 0, strlen(decryptConnInfo));
-    securec_check(rc, "", "");
-    pfree_ext(decryptConnInfo);
-    if (!connectSuccess) {
-        ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("could not connect to the publisher")));
+    if (!AttemptConnectPublisher(t_thrd.applyworker_cxt.mySubscription->conninfo,
+        t_thrd.applyworker_cxt.mySubscription->name, true)) {
+        ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("Failed to connect to publisher.")));
     }
 
     /*
@@ -1447,6 +1517,7 @@ void ApplyWorkerMain()
     options.slotname = t_thrd.applyworker_cxt.mySubscription->slotname;
     options.protoVersion = LOGICALREP_PROTO_VERSION_NUM;
     options.publicationNames = t_thrd.applyworker_cxt.mySubscription->publications;
+    options.binary = t_thrd.applyworker_cxt.mySubscription->binary;
 
     /* Start streaming from the slot. */
     (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_startstreaming(&options);
@@ -1595,3 +1666,64 @@ char* DefListToString(const List *defList)
     return buf.data;
 }
 
+/*
+ * Handle conninfo update message.
+ */
+static void apply_handle_conninfo(StringInfo s)
+{
+    char* standbysInfo = NULL;
+    logicalrep_read_conninfo(s, &standbysInfo);
+    UpdateConninfo(standbysInfo);
+    pfree_ext(standbysInfo);
+}
+
+static void UpdateConninfo(char* standbysInfo)
+{
+    Relation rel;
+    bool nulls[Natts_pg_subscription];
+    bool replaces[Natts_pg_subscription];
+    Datum values[Natts_pg_subscription];
+    HeapTuple tup;
+    Subscription* sub = t_thrd.applyworker_cxt.mySubscription;
+    Oid subid = sub->oid;
+
+    StartTransactionCommand();
+    rel = heap_open(SubscriptionRelationId, RowExclusiveLock);
+    /* Fetch the existing tuple. */
+    tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, u_sess->proc_cxt.MyDatabaseId,
+        CStringGetDatum(t_thrd.applyworker_cxt.mySubscription->name));
+    if (!HeapTupleIsValid(tup)) {
+        ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT), errmsg("subscription \"%s\" does not exist",
+            t_thrd.applyworker_cxt.mySubscription->name)));
+    }
+    subid = HeapTupleGetOid(tup);
+
+    /* Form a new tuple. */
+    int rc = memset_s(nulls, sizeof(nulls), false, sizeof(nulls));
+    securec_check(rc, "", "");
+    rc = memset_s(values, sizeof(values), 0, sizeof(values));
+    securec_check(rc, "", "");
+    rc = memset_s(replaces, sizeof(replaces), false, sizeof(replaces));
+    securec_check(rc, "", "");
+
+    /* get conninfoWithoutHostport */
+    StringInfoData conninfoWithoutHostport;
+    initStringInfo(&conninfoWithoutHostport);
+    ParseConninfo(sub->conninfo, &conninfoWithoutHostport, (HostPort**)NULL);
+
+    /* join conninfoWithoutHostport together with standbysinfo */
+    appendStringInfo(&conninfoWithoutHostport, " %s", standbysInfo);
+    /* Replace connection information */
+    values[Anum_pg_subscription_subconninfo - 1] = CStringGetTextDatum(conninfoWithoutHostport.data);
+    replaces[Anum_pg_subscription_subconninfo - 1] = true;
+    tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls, replaces);
+
+    /* Update the catalog. */
+    simple_heap_update(rel, &tup->t_self, tup);
+    CatalogUpdateIndexes(rel, tup);
+
+    heap_close(rel, RowExclusiveLock);
+    CommitTransactionCommand();
+
+    ereport(LOG, (errmsg("Update conninfo successfully, new conninfo %s.", standbysInfo)));
+}
diff --git a/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp b/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp
index 425fd6d28..f9f070473 100644
--- a/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp
+++ b/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp
@@ -15,6 +15,8 @@
 
 #include "catalog/pg_publication.h"
 
+#include "commands/defrem.h"
+
 #include "replication/logical.h"
 #include "replication/logicalproto.h"
 #include "replication/origin.h"
@@ -44,6 +46,8 @@ static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, RepOriginId orig
 
 static List *LoadPublications(List *pubnames);
 static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue);
+static bool ReplconninfoChanged();
+static void GetConninfo(StringInfoData* standbysInfo);
 
 /* Entry in the map used to remember which relation schemas we sent. */
 typedef struct RelationSyncEntry {
@@ -79,6 +83,9 @@ static void parse_output_parameters(List *options, PGOutputData *data)
     ListCell *lc;
     bool protocol_version_given = false;
     bool publication_names_given = false;
+    bool binary_option_given = false;
+
+    data->binary = false;
 
     foreach (lc, options) {
         DefElem *defel = (DefElem *)lfirst(lc);
@@ -108,6 +115,12 @@ static void parse_output_parameters(List *options, PGOutputData *data)
 
             if (!SplitIdentifierString(strVal(defel->arg), ',', &(data->publication_names)))
                 ereport(ERROR, (errcode(ERRCODE_INVALID_NAME), errmsg("invalid publication_names syntax")));
+        } else if (strcmp(defel->defname, "binary") == 0) {
+            if (binary_option_given)
+                ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options")));
+            binary_option_given = true;
+
+            data->binary = defGetBoolean(defel);
         } else
             elog(ERROR, "unrecognized pgoutput option: %s", defel->defname);
     }
@@ -206,6 +219,22 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *t
     OutputPluginPrepareWrite(ctx, true);
     logicalrep_write_commit(ctx->out, txn, commit_lsn);
     OutputPluginWrite(ctx, true);
+
+    /*
+     * Send the newest connecttion information to the subscriber,
+     * when the connection information about the standby changes.
+     */
+    if (ReplconninfoChanged()) {
+        StringInfoData standbysInfo;
+        initStringInfo(&standbysInfo);
+
+        GetConninfo(&standbysInfo);
+        OutputPluginPrepareWrite(ctx, true);
+        logicalrep_write_conninfo(ctx->out, standbysInfo.data);
+        OutputPluginWrite(ctx, true);
+
+        FreeStringInfo(&standbysInfo);
+    }
 }
 
 /*
@@ -300,19 +329,19 @@ static void pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
     switch (change->action) {
         case REORDER_BUFFER_CHANGE_INSERT:
             OutputPluginPrepareWrite(ctx, true);
-            logicalrep_write_insert(ctx->out, relation, &change->data.tp.newtuple->tuple);
+            logicalrep_write_insert(ctx->out, relation, &change->data.tp.newtuple->tuple, data->binary);
             OutputPluginWrite(ctx, true);
             break;
         case REORDER_BUFFER_CHANGE_UINSERT:
             OutputPluginPrepareWrite(ctx, true);
-            logicalrep_write_insert(ctx->out, relation, (HeapTuple)(&change->data.utp.newtuple->tuple));
+            logicalrep_write_insert(ctx->out, relation, (HeapTuple)(&change->data.utp.newtuple->tuple), data->binary);
             OutputPluginWrite(ctx, true);
             break;
         case REORDER_BUFFER_CHANGE_UPDATE: {
             HeapTuple oldtuple = change->data.tp.oldtuple ? &change->data.tp.oldtuple->tuple : NULL;
 
             OutputPluginPrepareWrite(ctx, true);
-            logicalrep_write_update(ctx->out, relation, oldtuple, &change->data.tp.newtuple->tuple);
+            logicalrep_write_update(ctx->out, relation, oldtuple, &change->data.tp.newtuple->tuple, data->binary);
             OutputPluginWrite(ctx, true);
             break;
         }
@@ -320,14 +349,14 @@ static void pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
             HeapTuple oldtuple = change->data.utp.oldtuple ? ((HeapTuple)(&change->data.utp.oldtuple->tuple)) : NULL;
 
             OutputPluginPrepareWrite(ctx, true);
-            logicalrep_write_update(ctx->out, relation, oldtuple, (HeapTuple)(&change->data.utp.newtuple->tuple));
+            logicalrep_write_update(ctx->out, relation, oldtuple, (HeapTuple)(&change->data.utp.newtuple->tuple), data->binary);
             OutputPluginWrite(ctx, true);
             break;
         }
         case REORDER_BUFFER_CHANGE_DELETE:
             if (change->data.tp.oldtuple) {
                 OutputPluginPrepareWrite(ctx, true);
-                logicalrep_write_delete(ctx->out, relation, &change->data.tp.oldtuple->tuple);
+                logicalrep_write_delete(ctx->out, relation, &change->data.tp.oldtuple->tuple, data->binary);
                 OutputPluginWrite(ctx, true);
             } else
                 elog(DEBUG1, "didn't send DELETE change because of missing oldtuple");
@@ -335,7 +364,7 @@ static void pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
         case REORDER_BUFFER_CHANGE_UDELETE:
             if (change->data.utp.oldtuple) {
                 OutputPluginPrepareWrite(ctx, true);
-                logicalrep_write_delete(ctx->out, relation, (HeapTuple)(&change->data.utp.oldtuple->tuple));
+                logicalrep_write_delete(ctx->out, relation, (HeapTuple)(&change->data.utp.oldtuple->tuple), data->binary);
                 OutputPluginWrite(ctx, true);
             } else
                 elog(DEBUG1, "didn't send DELETE change because of missing oldtuple");
@@ -605,3 +634,43 @@ static void rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashval
         entry->pubactions.pubdelete = false;
     }
 }
+
+static void GetConninfo(StringInfoData* standbysInfo)
+{
+    bool primaryJoined = false;
+    StringInfoData hosts;
+    StringInfoData ports;
+    initStringInfo(&hosts);
+    initStringInfo(&ports);
+    for (int i = 1; i < MAX_REPLNODE_NUM + 1; ++i) {
+        t_thrd.postmaster_cxt.ReplConnChangeType[i] = 0;
+        if (t_thrd.postmaster_cxt.ReplConnArray[i] == NULL) {
+            continue;
+        }
+        if (!primaryJoined) {
+            appendStringInfo(&hosts, "%s,%s",
+                t_thrd.postmaster_cxt.ReplConnArray[i]->localhost,
+                t_thrd.postmaster_cxt.ReplConnArray[i]->remotehost);
+            appendStringInfo(&ports, "%d,%d",
+                t_thrd.postmaster_cxt.ReplConnArray[i]->localport,
+                t_thrd.postmaster_cxt.ReplConnArray[i]->remoteport);
+            primaryJoined = true;
+        } else {
+            appendStringInfo(&hosts, ",%s",
+                t_thrd.postmaster_cxt.ReplConnArray[i]->remotehost);
+            appendStringInfo(&ports, ",%d",
+                t_thrd.postmaster_cxt.ReplConnArray[i]->remoteport);
+        }
+    }
+    appendStringInfo(standbysInfo, "host=%s port=%s", hosts.data, ports.data);
+}
+
+static inline bool ReplconninfoChanged()
+{
+    for (int i = 1; i < MAX_REPLNODE_NUM; ++i) {
+        if (t_thrd.postmaster_cxt.ReplConnChangeType[i]) {
+            return true;
+        }
+    }
+    return false;
+}
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 12d1c3a56..6765cd38c 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -50,13 +50,15 @@ CATALOG(pg_subscription,6126) BKI_SHARED_RELATION BKI_ROWTYPE_OID(6128) BKI_SCHE
     NameData subslotname;    /* Slot name on publisher */
     text subsynccommit;      /* Synchronous commit setting for worker */
     text subpublications[1]; /* List of publications subscribed to */
+    bool subbinary;          /* True if the subscription wants the
+                              * publisher to send data in binary */
 #endif
 }
 FormData_pg_subscription;
 
 typedef FormData_pg_subscription *Form_pg_subscription;
 
-#define Natts_pg_subscription 8
+#define Natts_pg_subscription 9
 #define Anum_pg_subscription_subdbid 1
 #define Anum_pg_subscription_subname 2
 #define Anum_pg_subscription_subowner 3
@@ -65,6 +67,7 @@ typedef FormData_pg_subscription *Form_pg_subscription;
 #define Anum_pg_subscription_subslotname 6
 #define Anum_pg_subscription_subsynccommit 7
 #define Anum_pg_subscription_subpublications 8
+#define Anum_pg_subscription_subbinary 9
 
 
 typedef struct Subscription {
@@ -77,6 +80,7 @@ typedef struct Subscription {
     char *slotname;     /* Name of the replication slot */
     char *synccommit;   /* Synchronous commit setting for worker */
     List *publications; /* List of publication names to subscribe to */
+    bool binary;        /* Indicates if the subscription wants data in binary format */
 } Subscription;
 
 
@@ -86,6 +90,7 @@ extern Oid get_subscription_oid(const char *subname, bool missing_ok);
 extern char *get_subscription_name(Oid subid, bool missing_ok);
 
 extern int CountDBSubscriptions(Oid dbid);
-extern char *DecryptConninfo(char *encryptConninfo);
+extern void ClearListContent(List *list);
+
 
 #endif /* PG_SUBSCRIPTION_H */
diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h
index 9ee8cf177..4f1498225 100644
--- a/src/include/commands/subscriptioncmds.h
+++ b/src/include/commands/subscriptioncmds.h
@@ -17,6 +17,11 @@
 
 #include "nodes/parsenodes.h"
 
+typedef struct HostPort {
+    char* host;
+    char* port;
+} HostPort;
+
 extern ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel);
 extern ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt);
 extern void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel);
@@ -24,6 +29,12 @@ extern void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel);
 extern ObjectAddress AlterSubscriptionOwner(const char *name, Oid newOwnerId);
 extern void AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId);
 extern void RenameSubscription(List* oldname, const char* newname);
+extern void AddStandbysInfo(char* standbysInfo);
+extern void DropStandbysInfo(char* standbysInfo);
+
+extern void ParseConninfo(const char* conninfo, StringInfoData* conninfoWithoutHostPort, HostPort** hostPortList);
+extern char* EncryptOrDecryptConninfo(const char* conninfo, const char action);
+extern bool AttemptConnectPublisher(const char *conninfoOriginal, char* slotname, bool checkRemoteMode);
 
 #endif							/* SUBSCRIPTIONCMDS_H */
 
diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h
index b3f8660b4..80e2ab6db 100644
--- a/src/include/miscadmin.h
+++ b/src/include/miscadmin.h
@@ -90,6 +90,7 @@ extern const uint32 SUPPORT_DATA_REPAIR;
 extern const uint32 SCAN_BATCH_MODE_VERSION_NUM;
 extern const uint32 RELMAP_4K_VERSION_NUM;
 extern const uint32 PUBLICATION_VERSION_NUM;
+extern const uint32 SUBSCRIPTION_BINARY_VERSION_NUM;
 extern const uint32 ANALYZER_HOOK_VERSION_NUM;
 extern const uint32 SUPPORT_HASH_XLOG_VERSION_NUM;
 extern const uint32 PITR_INIT_VERSION_NUM;
diff --git a/src/include/replication/libpqwalreceiver.h b/src/include/replication/libpqwalreceiver.h
index ba0ce4c25..ecd6a1a92 100755
--- a/src/include/replication/libpqwalreceiver.h
+++ b/src/include/replication/libpqwalreceiver.h
@@ -35,6 +35,7 @@ typedef struct LibpqrcvConnectParam {
     bool logical;
     uint32 protoVersion;    /* Logical protocol version */
     List *publicationNames; /* String list of publications */
+    bool binary;            /* Ask publisher to use binary */
 }LibpqrcvConnectParam;
 
 extern int32 pg_atoi(char* s, int size, int c);
@@ -53,5 +54,6 @@ extern bool libpqrcv_command(const char *cmd, char **err, int *sqlstate);
 extern void IdentifyRemoteSystem(bool checkRemote);
 extern void CreateRemoteReplicationSlot(XLogRecPtr startpoint, const char* slotname, bool isLogical);
 extern void StartRemoteStreaming(const LibpqrcvConnectParam *options);
+extern ServerMode IdentifyRemoteMode();
 
 #endif
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 42fb59f13..4e2876b86 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -35,12 +35,21 @@
  * Keep in mind that the columns correspond to the *remote* table.
  */
 typedef struct LogicalRepTupleData {
-    char *values[MaxTupleAttributeNumber]; /* value in out function format or NULL if values is NULL */
-    bool changed[MaxTupleAttributeNumber]; /* marker for changed/unchanged values */
+    /* Array of StringInfos, one per column; some may be unused */
+    StringInfoData *colvalues;
+    /* Array of markers for null/unchanged/text/binary, one per column */
+    char       *colstatus;
     /* Length of above arrays */
     int ncols;
 } LogicalRepTupleData;
 
+/* Possible values for LogicalRepTupleData.colstatus[colnum] */
+/* These values are also used in the on-the-wire protocol */
+#define LOGICALREP_COLUMN_NULL        'n'
+#define LOGICALREP_COLUMN_UNCHANGED    'u'
+#define LOGICALREP_COLUMN_TEXT        't'
+#define LOGICALREP_COLUMN_BINARY    'b' /* added in PG14 */
+
 typedef uint32 LogicalRepRelId;
 
 /* Relation information */
@@ -81,16 +90,18 @@ extern void logicalrep_read_begin(StringInfo in, LogicalRepBeginData *begin_data
 extern void logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
 extern void logicalrep_read_commit(StringInfo in, LogicalRepCommitData *commit_data);
 extern void logicalrep_write_origin(StringInfo out, const char *origin, XLogRecPtr origin_lsn);
-extern void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple);
+extern void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple, bool binary);
 extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup);
-extern void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple);
+extern void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple, bool binary);
 extern LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup,
     LogicalRepTupleData *newtup);
-extern void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple);
+extern void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple, bool binary);
 extern LogicalRepRelId logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup);
 extern void logicalrep_write_rel(StringInfo out, Relation rel);
 extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
 extern void logicalrep_write_typ(StringInfo out, Oid typoid);
 extern void logicalrep_read_typ(StringInfo out, LogicalRepTyp *ltyp);
+extern void logicalrep_write_conninfo(StringInfo out, char* conninfo);
+extern void logicalrep_read_conninfo(StringInfo in, char** conninfo);
 
 #endif /* LOGICALREP_PROTO_H */
diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h
index fb63f49b3..0a2fd3570 100644
--- a/src/include/replication/pgoutput.h
+++ b/src/include/replication/pgoutput.h
@@ -24,6 +24,7 @@ typedef struct PGOutputData {
 
     List *publication_names;
     List *publications;
+    bool binary;
 } PGOutputData;
 
 #endif /* PGOUTPUT_H */
diff --git a/src/test/regress/input/subscription.source b/src/test/regress/input/subscription.source
index ff384b618..2d5a4c4e9 100644
--- a/src/test/regress/input/subscription.source
+++ b/src/test/regress/input/subscription.source
@@ -24,6 +24,7 @@ CREATE SUBSCRIPTION testsub CONNECTION 'foo';
 CREATE SUBSCRIPTION testsub PUBLICATION foo;
 -- fail - could not connect to the publisher
 create subscription testsub2 connection 'host=abc' publication pub;
+create subscription testsub2 connection 'host=abc port=12345' publication pub;
 set client_min_messages to error;
 -- fail - syntax error, invalid connection string syntax: missing "="
 CREATE SUBSCRIPTION testsub CONNECTION 'testconn' PUBLICATION testpub;
@@ -32,12 +33,15 @@ CREATE SUBSCRIPTION testsub CONNECTION 'dbname=doesnotexist' PUBLICATION testpub
 CREATE SUBSCRIPTION testsub CONNECTION 'dbname=doesnotexist' PUBLICATION testpub WITH (ENABLED=false, slot_name='testsub', synchronous_commit=off);
 -- create SUBSCRIPTION with conninfo in two single quote, used to check mask string bug
 CREATE SUBSCRIPTION testsub_maskconninfo CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234''' PUBLICATION testpub WITH (ENABLED=false, slot_name='testsub', synchronous_commit=off);
-
+-- fail - The number of host and port are inconsistent
+create subscription sub1 connection 'dbname=postgres user=pubusr password=Huawei@123 host=192.168.0.38,192.168.0.38,192.168.0.38 port=14001,14501' publication pub1;
+-- fail - a maximum of 9 servers are supported
+create subscription sub1 connection 'dbname=postgres user=pubusr password=Huawei@123 host=192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38 port=14001,14501' publication pub1;
 -- alter connection
 ALTER SUBSCRIPTION testsub CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234''';
 ALTER SUBSCRIPTION testsub CONNECTION 'dbname=does_not_exist';
 reset client_min_messages;
-select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications from pg_subscription where subname='testsub';
+select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications, subbinary from pg_subscription where subname='testsub';
 --- alter subscription
 ------ set publication
 ALTER SUBSCRIPTION testsub SET PUBLICATION testpub2, testpub3;
@@ -57,6 +61,9 @@ select subname, subenabled, subsynccommit from pg_subscription  where subname='t
 ALTER SUBSCRIPTION testsub SET (slot_name='testsub');
 -- alter owner
 ALTER SUBSCRIPTION testsub owner to regress_subscription_user2;
+-- alter subbinary to true
+ALTER SUBSCRIPTION testsub SET (binary=true);
+select subname, subbinary from pg_subscription where subname='testsub';
 --rename
 ALTER SUBSCRIPTION testsub rename to testsub_rename;
 --- inside a transaction block
diff --git a/src/test/regress/output/subscription.source b/src/test/regress/output/subscription.source
index d2317de8f..7bf363408 100644
--- a/src/test/regress/output/subscription.source
+++ b/src/test/regress/output/subscription.source
@@ -60,8 +60,10 @@ LINE 1: CREATE SUBSCRIPTION testsub PUBLICATION foo;
                                     ^
 -- fail - could not connect to the publisher
 create subscription testsub2 connection 'host=abc' publication pub;
+ERROR:  The number of host and port are inconsistent.
+create subscription testsub2 connection 'host=abc port=12345' publication pub;
 WARNING:  apply worker could not connect to the remote server
-ERROR:  could not connect to the publisher
+ERROR:  Failed to connect to publisher.
 set client_min_messages to error;
 -- fail - syntax error, invalid connection string syntax: missing "="
 CREATE SUBSCRIPTION testsub CONNECTION 'testconn' PUBLICATION testpub;
@@ -72,14 +74,20 @@ ERROR:  unrecognized subscription parameter: create_slot
 CREATE SUBSCRIPTION testsub CONNECTION 'dbname=doesnotexist' PUBLICATION testpub WITH (ENABLED=false, slot_name='testsub', synchronous_commit=off);
 -- create SUBSCRIPTION with conninfo in two single quote, used to check mask string bug
 CREATE SUBSCRIPTION testsub_maskconninfo CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234''' PUBLICATION testpub WITH (ENABLED=false, slot_name='testsub', synchronous_commit=off);
+-- fail - The number of host and port are inconsistent
+create subscription sub1 connection 'dbname=postgres user=pubusr password=Huawei@123 host=192.168.0.38,192.168.0.38,192.168.0.38 port=14001,14501' publication pub1;
+ERROR:  The number of host and port are inconsistent.
+-- fail - a maximum of 9 servers are supported
+create subscription sub1 connection 'dbname=postgres user=pubusr password=Huawei@123 host=192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38,192.168.0.38 port=14001,14501' publication pub1;
+ERROR:  Currently, a maximum of 9 servers are supported.
 -- alter connection
 ALTER SUBSCRIPTION testsub CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234''';
 ALTER SUBSCRIPTION testsub CONNECTION 'dbname=does_not_exist';
 reset client_min_messages;
-select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications from pg_subscription where subname='testsub';
- subname |           owner           | subenabled |      subconninfo       | subpublications 
----------+---------------------------+------------+------------------------+-----------------
- testsub | regress_subscription_user | f          | dbname=does_not_exist  | {testpub}
+select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications, subbinary from pg_subscription where subname='testsub';
+ subname |           owner           | subenabled |      subconninfo       | subpublications | subbinary 
+---------+---------------------------+------------+------------------------+-----------------+-----------
+ testsub | regress_subscription_user | f          | dbname=does_not_exist  | {testpub}       | f
 (1 row)
 
 --- alter subscription
@@ -122,6 +130,14 @@ ALTER SUBSCRIPTION testsub SET (slot_name='testsub');
 ERROR:  Currently enabled=false, cannot change slot_name to a non-null value.
 -- alter owner
 ALTER SUBSCRIPTION testsub owner to regress_subscription_user2;
+-- alter subbinary to true
+ALTER SUBSCRIPTION testsub SET (binary=true);
+select subname, subbinary from pg_subscription where subname='testsub';
+ subname | subbinary 
+---------+-----------
+ testsub | t
+(1 row)
+
 --rename
 ALTER SUBSCRIPTION testsub rename to testsub_rename;
 --- inside a transaction block
@@ -134,8 +150,7 @@ COMMIT;
 -- -- active SUBSCRIPTION
 BEGIN;
 ALTER SUBSCRIPTION testsub_rename ENABLE;
-WARNING:  apply worker could not connect to the remote server
-ERROR:  could not connect to the publisher
+ERROR:  invalid connection string syntax, missing host and port
 select subname, subenabled from pg_subscription  where subname='testsub_rename';
 ERROR:  current transaction is aborted, commands ignored until end of transaction block, firstChar[Q]
 ALTER SUBSCRIPTION testsub_rename SET (ENABLED=false);
@@ -281,10 +296,11 @@ SELECT object_name,detail_info FROM pg_query_audit('2022-01-13 9:30:00', '2031-1
  testsub_maskconninfo | ALTER SUBSCRIPTION testsub_maskconninfo SET (conninfo='*************************************************************************************************************************;
  testsub              | ALTER SUBSCRIPTION testsub SET (synchronous_commit=on);
  testsub              | ALTER SUBSCRIPTION testsub owner to regress_subscription_user2;
+ testsub              | ALTER SUBSCRIPTION testsub SET (binary=true);
  testsub              | ALTER SUBSCRIPTION testsub rename to testsub_rename;
  testsub_rename       | DROP SUBSCRIPTION IF EXISTS testsub_rename;
  testsub_maskconninfo | DROP SUBSCRIPTION IF EXISTS testsub_maskconninfo;
-(14 rows)
+(15 rows)
 
 --clear audit log
 SELECT pg_delete_audit('1012-11-10', '3012-11-11');