forked from huawei/openGauss-server
!1511 发布订阅支持以二进制格式发送数据和发布端主备切换不断开
Merge pull request !1511 from 薛蒙恩/pubsub330
This commit is contained in:
Gitee
commit
557583acd0
|
|
@ -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 = 92607;
|
||||
|
||||
#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));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -498,6 +498,7 @@ typedef struct _SubscriptionInfo {
|
|||
char *subslotname;
|
||||
char *subsynccommit;
|
||||
char *subpublications;
|
||||
char *subbinary;
|
||||
} SubscriptionInfo;
|
||||
|
||||
/* global decls */
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
||||
|
|
@ -69,7 +68,7 @@ Subscription *GetSubscription(Oid subid, bool missing_ok)
|
|||
|
||||
/* Get slotname */
|
||||
datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, Anum_pg_subscription_subslotname, &isnull);
|
||||
if (!isnull) {
|
||||
if (unlikely(isnull)) {
|
||||
sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
|
||||
} else {
|
||||
sub->slotname = NULL;
|
||||
|
|
@ -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.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ bool open_join_children = true;
|
|||
bool will_shutdown = false;
|
||||
|
||||
/* hard-wired binary version number */
|
||||
const uint32 GRAND_VERSION_NUM = 92606;
|
||||
const uint32 GRAND_VERSION_NUM = 92607;
|
||||
|
||||
const uint32 PREDPUSH_SAME_LEVEL_VERSION_NUM = 92522;
|
||||
const uint32 UPSERT_WHERE_VERSION_NUM = 92514;
|
||||
|
|
@ -101,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 = 92607;
|
||||
|
||||
/* Version number of the guc parameter backend_version added in V500R001C20 */
|
||||
const uint32 V5R1C20_BACKEND_VERSION_NUM = 92305;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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,93 @@ 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;
|
||||
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 *) palloc((len + 1) * sizeof(char));
|
||||
pq_copymsgbytes(in, value->data, len);
|
||||
/* not strictly necessary but per StringInfo practice */
|
||||
value->data[len] = '\0';
|
||||
/* make StringInfo fully valid */
|
||||
value->len = len;
|
||||
value->cursor = 0;
|
||||
value->maxlen = len;
|
||||
break;
|
||||
}
|
||||
default:
|
||||
elog(ERROR, "unrecognized data representation type '%c'", kind);
|
||||
break;
|
||||
|
|
@ -451,7 +481,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 +534,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 +603,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, "", "");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
@ -74,11 +78,14 @@ 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;
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 */
|
||||
|
|
|
|||
|
|
@ -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 */
|
||||
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 */
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ typedef struct PGOutputData {
|
|||
|
||||
List *publication_names;
|
||||
List *publications;
|
||||
bool binary;
|
||||
} PGOutputData;
|
||||
|
||||
#endif /* PGOUTPUT_H */
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
|
|
|
|||
Loading…
Reference in New Issue