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

support any x

This commit is contained in:
gentle_hu 2022-01-17 14:27:33 +08:00
parent f5e1bf0ef7
commit e5a91d0c00
16 changed files with 747 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/common/backend/catalog/builtin_funcs.ini
View File

@ -7557,7 +7557,7 @@
),
AddFuncGroup(
"pg_stat_get_wal_senders", 1,
AddBuiltinFunc(_0(3099), _1("pg_stat_get_wal_senders"), _2(0), _3(false), _4(true), _5(pg_stat_get_wal_senders), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(10), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('s'), _19(0), _20(0), _21(21, 20, 23, 25, 25, 25, 25, 1184, 1184, 25, 25, 25, 25, 25, 25, 25, 25, 25, 25, 23, 25, 25), _22(21, 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o'), _23(21, "pid", "sender_pid", "local_role", "peer_role", "peer_state", "state", "catchup_start", "catchup_end", "sender_sent_location", "sender_write_location", "sender_flush_location", "sender_replay_location", "receiver_received_location", "receiver_write_location", "receiver_flush_location", "receiver_replay_location", "sync_percent", "sync_state", "sync_priority", "sync_most_available", "channel"), _24(NULL), _25("pg_stat_get_wal_senders"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33(NULL), _34('f'))
AddBuiltinFunc(_0(3099), _1("pg_stat_get_wal_senders"), _2(0), _3(false), _4(true), _5(pg_stat_get_wal_senders), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(10), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('s'), _19(0), _20(0), _21(22, 20, 23, 25, 25, 25, 25, 1184, 1184, 25, 25, 25, 25, 25, 25, 25, 25, 25, 25, 23, 23, 25, 25), _22(22, 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o'), _23(22, "pid", "sender_pid", "local_role", "peer_role", "peer_state", "state", "catchup_start", "catchup_end", "sender_sent_location", "sender_write_location", "sender_flush_location", "sender_replay_location", "receiver_received_location", "receiver_write_location", "receiver_flush_location", "receiver_replay_location", "sync_percent", "sync_state", "sync_group", "sync_priority", "sync_most_available", "channel"), _24(NULL), _25("pg_stat_get_wal_senders"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33(NULL), _34('f'))
),
AddFuncGroup(
"pg_stat_get_wlm_ec_operator_info", 1,

2
src/common/backend/utils/init/globals.cpp
View File

@ -59,7 +59,7 @@ bool open_join_children = true;
bool will_shutdown = false;
/* hard-wired binary version number */
const uint32 GRAND_VERSION_NUM = 92300;
const uint32 GRAND_VERSION_NUM = 92301;
const uint32 MATVIEW_VERSION_NUM = 92213;
const uint32 PARTIALPUSH_VERSION_NUM = 92087;

3
src/gausskernel/process/threadpool/knl_thread.cpp
View File

@ -759,7 +759,7 @@ static void knl_t_replscanner_init(knl_t_replscanner_context* replscanner_cxt)
static void knl_t_syncgram_init(knl_t_syncrepgram_context* syncrepgram_cxt)
{
syncrepgram_cxt->syncrep_parse_result = NULL;
syncrepgram_cxt->syncrep_parse_result = NIL;
}
static void knl_t_syncrepscanner_init(knl_t_syncrepscanner_context* syncrepscanner_cxt)
@ -772,6 +772,7 @@ static void knl_t_syncrepscanner_init(knl_t_syncrepscanner_context* syncrepscann
static void knl_t_syncrep_init(knl_t_syncrep_context* syncrep_cxt)
{
syncrep_cxt->SyncRepConfig = NULL;
syncrep_cxt->SyncRepConfigGroups = 0;
syncrep_cxt->announce_next_takeover = true;
}

460
src/gausskernel/storage/replication/syncrep.cpp
View File

@ -80,7 +80,7 @@ static int SyncRepWakeQueue(bool all, int mode);
static void SyncRepWaitCompletionQueue();
static void SyncRepNotifyComplete();
static int SyncRepGetStandbyPriority(void);
static void SyncRepGetStandbyGroupAndPriority(int* gid, int* prio);
#ifndef ENABLE_MULTIPLE_NODES
static bool SyncRepGetSyncLeftTime(XLogRecPtr XactCommitLSN, TimestampTz* leftTime);
#endif
@ -89,15 +89,33 @@ static void SyncRepGetOldestSyncRecPtr(XLogRecPtr* receivePtr, XLogRecPtr* write
static void SyncRepGetNthLatestSyncRecPtr(XLogRecPtr* receivePtr, XLogRecPtr* writePtr, XLogRecPtr* flushPtr,
XLogRecPtr* replayPtr, List* sync_standbys, uint8 nth);
#ifdef USE_ASSERT_CHECKING
static bool SyncRepQueueIsOrderedByLSN(int mode);
#endif
static List *SyncRepGetSyncStandbysPriority(bool *am_sync, List** catchup_standbys = NULL);
static List *SyncRepGetSyncStandbysQuorum(bool *am_sync, List** catchup_standbys = NULL);
static List *SyncRepGetSyncStandbysPriority(bool *am_sync, int groupid, List** catchup_standbys = NULL);
static List *SyncRepGetSyncStandbysQuorum(bool *am_sync, int groupid, List** catchup_standbys = NULL);
static inline void free_sync_standbys_list(List* sync_standbys);
static bool allow_most_available_sync(const List* sync_standbys);
static int cmp_lsn(const void *a, const void *b);
typedef struct TransContext {
/* for global */
int group_nums;
List *existers;
/* for each group*/
SyncRepConfigData* conf;
bool is_star;
} TransContext;
static TransContext* create_transform_context();
static void bind_transform_context(TransContext *tcxt, SyncRepConfigData *conf);
static void free_transform_context(TransContext *tcxt);
static bool analyze_star_and_num(TransContext *tcxt);
static bool analyze_duplicate_names(TransContext *tcxt);
static bool analyze_cluster_names(TransContext *tcxt);
static List* transform_synchronous_standby_names(TransContext *tcxt);
#define CATCHUP_XLOG_DIFF(ptr1, ptr2, amount) \
XLogRecPtrIsInvalid(ptr1) ? false : (XLByteDifference(ptr2, ptr1) < amount)
@ -109,6 +127,19 @@ static int cmp_lsn(const void *a, const void *b);
} \
} while (0)
/*
* sync_standbys_list is a two-dimensional array means sync standby nodes in groups
* The structure is as follows: List[IntList[int, int,...], IntList[int, int, ...], ...]
*/
static inline void free_sync_standbys_list(List* sync_standbys)
{
ListCell *lc = NULL;
foreach(lc, sync_standbys) {
list_free((List*)lfirst(lc));
}
list_free(sync_standbys);
}
/*
* Determine whether to wait for standby catching up, if requested by user.
*
@ -495,15 +526,19 @@ void SyncRepCleanupAtProcExit(void)
*/
void SyncRepInitConfig(void)
{
int group;
int priority;
/*
* Determine if we are a potential sync standby and remember the result
* for handling replies from standby.
*/
priority = SyncRepGetStandbyPriority();
if (t_thrd.walsender_cxt.MyWalSnd->sync_standby_priority != priority) {
SyncRepGetStandbyGroupAndPriority(&group, &priority);
if (t_thrd.walsender_cxt.MyWalSnd->sync_standby_group != group ||
t_thrd.walsender_cxt.MyWalSnd->sync_standby_priority != priority) {
LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
t_thrd.walsender_cxt.MyWalSnd->sync_standby_group = group;
t_thrd.walsender_cxt.MyWalSnd->sync_standby_priority = priority;
/*
@ -513,8 +548,8 @@ void SyncRepInitConfig(void)
SyncRepCheckSyncStandbyAlive();
LWLockRelease(SyncRepLock);
ereport(DEBUG1, (errmsg("standby \"%s\" now has synchronous standby priority %d",
u_sess->attr.attr_common.application_name, priority)));
ereport(DEBUG1, (errmsg("standby \"%s\" now has synchronous standby group and priority: %d %d",
u_sess->attr.attr_common.application_name, group, priority)));
}
}
@ -571,13 +606,15 @@ void SyncRepReleaseWaiters(void)
*/
if (t_thrd.syncrep_cxt.announce_next_takeover && am_sync) {
t_thrd.syncrep_cxt.announce_next_takeover = false;
if (t_thrd.syncrep_cxt.SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY) {
ereport(LOG, (errmsg("standby \"%s\" is now a synchronous standby with priority %d",
if (GetWalsndSyncRepConfig(t_thrd.walsender_cxt.MyWalSnd)->syncrep_method == SYNC_REP_PRIORITY) {
ereport(LOG, (errmsg("standby \"%s\" is now a synchronous standby with group and priority: %d %d",
u_sess->attr.attr_common.application_name,
t_thrd.walsender_cxt.MyWalSnd->sync_standby_group,
t_thrd.walsender_cxt.MyWalSnd->sync_standby_priority)));
} else {
ereport(LOG, (errmsg("standby \"%s\" is now a candidate for quorum synchronous standby",
u_sess->attr.attr_common.application_name)));
ereport(LOG, (errmsg("standby \"%s\" is now a candidate for quorum synchronous standby in group: %d",
u_sess->attr.attr_common.application_name,
t_thrd.walsender_cxt.MyWalSnd->sync_standby_group)));
}
}
@ -620,6 +657,35 @@ void SyncRepReleaseWaiters(void)
(uint32)writePtr, numflush, (uint32)(flushPtr >> 32), (uint32)flushPtr)));
}
/*
* In a particular scenario, when most_available_sync is true, primary only wait
* the alive sync standbys, even if the quantity does not meet the configuration
* requirements.
*/
static bool allow_most_available_sync(const List* sync_standbys)
{
ListCell* lc = NULL;
List* per_group = NIL;
int gid = 0;
int alive = 0;
foreach(lc, sync_standbys) {
per_group = (List*)lfirst(lc);
if (!t_thrd.walsender_cxt.WalSndCtl->most_available_sync &&
list_length(per_group) < t_thrd.syncrep_cxt.SyncRepConfig[gid]->num_sync)
return false;
alive += list_length(per_group);
gid++;
}
if (t_thrd.walsender_cxt.WalSndCtl->most_available_sync && alive == 0) {
return false;
}
return true;
}
/*
* Calculate the synced Receive, Write, Flush and Apply positions among sync standbys.
*
@ -633,6 +699,8 @@ void SyncRepReleaseWaiters(void)
bool SyncRepGetSyncRecPtr(XLogRecPtr *receivePtr, XLogRecPtr *writePtr, XLogRecPtr *flushPtr, XLogRecPtr* replayPtr, bool *am_sync, bool check_am_sync)
{
List *sync_standbys = NIL;
List *per_group = NIL;
ListCell *lc = NULL;
*receivePtr = InvalidXLogRecPtr;
*writePtr = InvalidXLogRecPtr;
@ -645,14 +713,12 @@ bool SyncRepGetSyncRecPtr(XLogRecPtr *receivePtr, XLogRecPtr *writePtr, XLogRecP
/*
* Quick exit if we are not managing a sync standby (or not check for check_am_sync is false)
* or there are not enough synchronous standbys.
* but in a particular scenario, when most_available_sync is true, primary only wait the alive sync standbys
* if list_length(sync_standbys) doesn't satisfy t_thrd.syncrep_cxt.SyncRepConfig->num_sync.
* or most_available_sync is working and allow some standby nodes to be missing.
*/
if ((!(*am_sync) && check_am_sync) || t_thrd.syncrep_cxt.SyncRepConfig == NULL ||
(!t_thrd.walsender_cxt.WalSndCtl->most_available_sync &&
list_length(sync_standbys) < t_thrd.syncrep_cxt.SyncRepConfig->num_sync) ||
(t_thrd.walsender_cxt.WalSndCtl->most_available_sync && list_length((sync_standbys)) == 0)) {
list_free(sync_standbys);
if ((!(*am_sync) && check_am_sync) ||
t_thrd.syncrep_cxt.SyncRepConfig == NULL ||
!allow_most_available_sync(sync_standbys)) {
free_sync_standbys_list(sync_standbys);
return false;
}
@ -669,11 +735,20 @@ bool SyncRepGetSyncRecPtr(XLogRecPtr *receivePtr, XLogRecPtr *writePtr, XLogRecP
* we can use SyncRepGetOldestSyncRecPtr() to calculate the synced
* positions even in a quorum-based sync replication.
*/
if (t_thrd.syncrep_cxt.SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY) {
SyncRepGetOldestSyncRecPtr(receivePtr, writePtr, flushPtr, replayPtr, sync_standbys);
} else {
SyncRepGetNthLatestSyncRecPtr(receivePtr, writePtr, flushPtr, replayPtr, sync_standbys,
t_thrd.syncrep_cxt.SyncRepConfig->num_sync);
int i = 0;
foreach(lc, sync_standbys) {
per_group = (List*)lfirst(lc);
if (per_group == NIL) {
/* do nothing, this group of sync standbys are all offline or no sync standbys. */
} else if (t_thrd.syncrep_cxt.SyncRepConfig[i]->syncrep_method == SYNC_REP_PRIORITY) {
SyncRepGetOldestSyncRecPtr(receivePtr, writePtr, flushPtr, replayPtr, per_group);
} else {
SyncRepGetNthLatestSyncRecPtr(receivePtr, writePtr, flushPtr, replayPtr,
per_group, t_thrd.syncrep_cxt.SyncRepConfig[i]->num_sync);
}
list_free(per_group);
i++;
}
list_free(sync_standbys);
@ -700,7 +775,7 @@ static bool SyncRepGetSyncLeftTime(XLogRecPtr XactCommitLSN, TimestampTz* leftTi
/* Skip here if there is at lease one sync standby, or no standby in catchup. */
if (list_length(sync_standbys) > 0 || list_length(catchup_standbys) == 0) {
list_free(sync_standbys);
free_sync_standbys_list(sync_standbys);
list_free(catchup_standbys);
return false;
}
@ -723,7 +798,7 @@ static bool SyncRepGetSyncLeftTime(XLogRecPtr XactCommitLSN, TimestampTz* leftTi
}
}
list_free(sync_standbys);
free_sync_standbys_list(sync_standbys);
list_free(catchup_standbys);
return true;
}
@ -822,10 +897,14 @@ static void SyncRepGetNthLatestSyncRecPtr(XLogRecPtr* receivePtr, XLogRecPtr* wr
}
/* Get Nth latest Write, Flush, Apply positions */
*receivePtr = receive_array[nth - 1];
*writePtr = write_array[nth - 1];
*flushPtr = flush_array[nth - 1];
*replayPtr = apply_array[nth - 1];
if (XLogRecPtrIsInvalid(*writePtr) || XLByteLE(write_array[nth - 1], *writePtr))
*writePtr = write_array[nth - 1];
if (XLogRecPtrIsInvalid(*flushPtr) || XLByteLE(flush_array[nth - 1], *flushPtr))
*flushPtr = flush_array[nth - 1];
if (XLogRecPtrIsInvalid(*receivePtr) || XLByteLE(receive_array[nth - 1], *receivePtr))
*receivePtr = receive_array[nth - 1];
if (XLogRecPtrIsInvalid(*replayPtr) || XLByteLE(apply_array[nth - 1], *replayPtr))
*replayPtr = apply_array[nth - 1];
pfree(receive_array);
receive_array = NULL;
@ -855,47 +934,56 @@ static int cmp_lsn(const void *a, const void *b)
/*
* Check if we are in the list of sync standbys, and if so, determine
* priority sequence. Return priority if set, or zero to indicate that
* priority sequence. Return groupid and priority if set, or zero to indicate that
* we are not a potential sync standby.
*
* Compare the parameter SyncRepStandbyNames against the application_name
* for this WALSender, or allow any name if we find a wildcard "*".
*/
static int SyncRepGetStandbyPriority(void)
static void SyncRepGetStandbyGroupAndPriority(int* gid, int* prio)
{
const char *standby_name = NULL;
int priority;
int group;
bool found = false;
*gid = 0;
*prio = 0;
/*
* Since synchronous cascade replication is not allowed, we always set the
* priority of cascading walsender to zero.
*/
if (AM_WAL_STANDBY_SENDER)
return 0;
return;
if (!SyncStandbysDefined() || t_thrd.syncrep_cxt.SyncRepConfig == NULL || !SyncRepRequested())
return 0;
return;
standby_name = t_thrd.syncrep_cxt.SyncRepConfig->member_names;
for (priority = 1; priority <= t_thrd.syncrep_cxt.SyncRepConfig->nmembers; priority++) {
if (pg_strcasecmp(standby_name, u_sess->attr.attr_common.application_name) == 0 ||
strcmp(standby_name, "*") == 0) {
found = true;
break;
for (group = 0; group < t_thrd.syncrep_cxt.SyncRepConfigGroups && !found; group++) {
standby_name = t_thrd.syncrep_cxt.SyncRepConfig[group]->member_names;
for (priority = 1; priority <= t_thrd.syncrep_cxt.SyncRepConfig[group]->nmembers; priority++) {
if (pg_strcasecmp(standby_name, u_sess->attr.attr_common.application_name) == 0 ||
strcmp(standby_name, "*") == 0) {
Assert(!(group > 1 && strcmp(standby_name, "*") == 0));
found = true;
break;
}
standby_name += strlen(standby_name) + 1;
}
standby_name += strlen(standby_name) + 1;
}
if (!found) {
return 0;
return;
}
/*
* In quorum-based sync replication, all the standbys in the list
* have the same priority, one.
*/
return (t_thrd.syncrep_cxt.SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY) ? priority : 1;
*gid = group - 1;
*prio = (t_thrd.syncrep_cxt.SyncRepConfig[group - 1]->syncrep_method == SYNC_REP_PRIORITY) ? priority : 1;
return;
}
/*
@ -1116,9 +1204,16 @@ List *SyncRepGetSyncStandbys(bool *am_sync, List** catchup_standbys)
if (t_thrd.syncrep_cxt.SyncRepConfig == NULL)
return NIL;
return (t_thrd.syncrep_cxt.SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY)
? SyncRepGetSyncStandbysPriority(am_sync, catchup_standbys)
: SyncRepGetSyncStandbysQuorum(am_sync, catchup_standbys);
List* results = NIL;
for(int i = 0; i < t_thrd.syncrep_cxt.SyncRepConfigGroups; i++) {
if (t_thrd.syncrep_cxt.SyncRepConfig[i]->syncrep_method == SYNC_REP_PRIORITY) {
results = lappend(results, SyncRepGetSyncStandbysPriority(am_sync, i,catchup_standbys));
} else {
results = lappend(results, SyncRepGetSyncStandbysQuorum(am_sync, i, catchup_standbys));
}
}
return results;
}
/*
@ -1131,13 +1226,13 @@ List *SyncRepGetSyncStandbys(bool *am_sync, List** catchup_standbys)
* On return, *am_sync is set to true if this walsender is connecting to
* sync standby. Otherwise it's set to false.
*/
static List *SyncRepGetSyncStandbysQuorum(bool *am_sync, List** catchup_standbys)
static List *SyncRepGetSyncStandbysQuorum(bool *am_sync, int groupid, List **catchup_standbys)
{
List *result = NIL;
int i;
volatile WalSnd *walsnd = NULL; /* Use volatile pointer to prevent code rearrangement */
Assert(t_thrd.syncrep_cxt.SyncRepConfig->syncrep_method == SYNC_REP_QUORUM);
Assert(t_thrd.syncrep_cxt.SyncRepConfig[groupid]->syncrep_method == SYNC_REP_QUORUM);
for (i = 0; i < g_instance.attr.attr_storage.max_wal_senders; i++) {
walsnd = &t_thrd.walsender_cxt.WalSndCtl->walsnds[i];
@ -1147,7 +1242,7 @@ static List *SyncRepGetSyncStandbysQuorum(bool *am_sync, List** catchup_standbys
continue;
/* Must be synchronous */
if (walsnd->sync_standby_priority == 0)
if (walsnd->sync_standby_priority == 0 || walsnd->sync_standby_group != groupid)
continue;
if ((walsnd->state == WALSNDSTATE_CATCHUP || walsnd->peer_state == CATCHUP_STATE) &&
@ -1188,7 +1283,7 @@ static List *SyncRepGetSyncStandbysQuorum(bool *am_sync, List** catchup_standbys
* On return, *am_sync is set to true if this walsender is connecting to
* sync standby. Otherwise it's set to false.
*/
static List *SyncRepGetSyncStandbysPriority(bool *am_sync, List** catchup_standbys)
static List *SyncRepGetSyncStandbysPriority(bool *am_sync, int groupid, List **catchup_standbys)
{
List *result = NIL;
List *pending = NIL;
@ -1200,9 +1295,9 @@ static List *SyncRepGetSyncStandbysPriority(bool *am_sync, List** catchup_standb
bool am_in_pending = false;
volatile WalSnd *walsnd = NULL; /* Use volatile pointer to prevent code rearrangement */
Assert(t_thrd.syncrep_cxt.SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY);
Assert(t_thrd.syncrep_cxt.SyncRepConfig[groupid]->syncrep_method == SYNC_REP_PRIORITY);
lowest_priority = t_thrd.syncrep_cxt.SyncRepConfig->nmembers;
lowest_priority = t_thrd.syncrep_cxt.SyncRepConfig[groupid]->nmembers;
next_highest_priority = lowest_priority + 1;
/*
@ -1219,7 +1314,7 @@ static List *SyncRepGetSyncStandbysPriority(bool *am_sync, List** catchup_standb
/* Must be synchronous */
this_priority = walsnd->sync_standby_priority;
if (this_priority == 0)
if (this_priority == 0 || walsnd->sync_standby_group != groupid)
continue;
if ((walsnd->state == WALSNDSTATE_CATCHUP || walsnd->peer_state == CATCHUP_STATE) &&
@ -1244,7 +1339,7 @@ static List *SyncRepGetSyncStandbysPriority(bool *am_sync, List** catchup_standb
result = lappend_int(result, i);
if (am_sync != NULL && walsnd == t_thrd.walsender_cxt.MyWalSnd)
*am_sync = true;
if (list_length(result) == t_thrd.syncrep_cxt.SyncRepConfig->num_sync) {
if (list_length(result) == t_thrd.syncrep_cxt.SyncRepConfig[groupid]->num_sync) {
list_free(pending);
return result; /* Exit if got enough sync standbys */
}
@ -1269,7 +1364,7 @@ static List *SyncRepGetSyncStandbysPriority(bool *am_sync, List** catchup_standb
* Consider all pending standbys as sync if the number of them plus
* already-found sync ones is lower than the configuration requests.
*/
if (list_length(result) + list_length(pending) <= t_thrd.syncrep_cxt.SyncRepConfig->num_sync) {
if (list_length(result) + list_length(pending) <= t_thrd.syncrep_cxt.SyncRepConfig[groupid]->num_sync) {
bool needfree = (result != NIL && pending != NIL);
/*
@ -1315,7 +1410,7 @@ static List *SyncRepGetSyncStandbysPriority(bool *am_sync, List** catchup_standb
* starts because we know that the list has enough elements to
* reach SyncRepConfig->num_sync.
*/
if (list_length(result) == t_thrd.syncrep_cxt.SyncRepConfig->num_sync) {
if (list_length(result) == t_thrd.syncrep_cxt.SyncRepConfig[groupid]->num_sync) {
list_free(pending);
return result; /* Exit if got enough sync standbys */
}
@ -1489,6 +1584,166 @@ int init_gauss_cluster_config(void)
return 0;
}
/*
* =================================================================
* Analyze and transform GUC param synchronous_standby_names content
* Create result in session MEMORY_CONTEXT_STORAGE session.
* =================================================================
*/
static TransContext* create_transform_context()
{
TransContext* tcxt = (TransContext*)palloc(sizeof(TransContext));
tcxt->group_nums = list_length(t_thrd.syncrepgram_cxt.syncrep_parse_result);
Assert(tcxt->group_nums > 0);
tcxt->existers = NIL;
tcxt->conf = NULL;
tcxt->is_star = false;
return tcxt;
}
static void bind_transform_context(TransContext *tcxt, SyncRepConfigData *conf)
{
tcxt->conf = conf;
tcxt->is_star = false;
}
static void free_transform_context(TransContext *tcxt)
{
list_free(tcxt->existers);
pfree(tcxt);
}
static bool analyze_star_and_num(TransContext *tcxt)
{
SyncRepConfigData *conf = tcxt->conf;
char *p = conf->member_names;
for (int i = 0; i < conf->nmembers; i++) {
if (strcmp(p, "*") != 0) {
p += strlen(p) + 1;
continue;
}
if (tcxt->group_nums > 1) {
GUC_check_errdetail("'*' is not support when sync standby strategy is combinated.");
return false;
}
if (conf->nmembers > 1) {
GUC_check_errdetail("Please use the '*' separately, otherwise, it is semantic ambiguity.");
return false;
}
tcxt->is_star = true;
}
if (!tcxt->is_star && conf->num_sync > conf->nmembers) {
GUC_check_errdetail("The sync number must less or equals to the number of standby node names.");
return false;
}
return true;
}
static bool analyze_duplicate_names(TransContext *tcxt)
{
if (tcxt->is_star)
return true;
SyncRepConfigData *conf = tcxt->conf;
char *p = conf->member_names;
char *exister = NULL;
ListCell *lc = NULL;
for (int i = 0; i < conf->nmembers; i++) {
foreach(lc, tcxt->existers) {
exister = (char*)lfirst(lc);
if (pg_strcasecmp(p, exister) == 0) {
GUC_check_errdetail("Duplicate standby node name: %s", p);
return false;
}
}
tcxt->existers = lappend(tcxt->existers, p);
p += strlen(p) + 1;
}
return true;
}
static bool analyze_cluster_names(TransContext *tcxt)
{
if (tcxt->is_star)
return true;
SyncRepConfigData *conf = tcxt->conf;
if (t_thrd.role != WORKER ||
strcmp(u_sess->attr.attr_common.application_name, "gsql") != 0 ||
!has_static_config()) {
return true;
}
if (0 != init_gauss_cluster_config()) {
return true;
}
char* p = conf->member_names;
for (int i = 1; i <= conf->nmembers; i++) {
if (!CheckDataNameValue(p, t_thrd.proc_cxt.DataDir)) {
GUC_check_errdetail("invalid standby name: %s", p);
return false;
}
p += strlen(p) + 1;
}
return true;
}
static List* transform_synchronous_standby_names(TransContext* tcxt)
{
errno_t rc = EOK;
ListCell* lc = NULL;
List* results = NIL;
if (tcxt->group_nums > SYNC_REP_MAX_GROUPS) {
GUC_check_errdetail("Too much groups. Please no more than %d.", SYNC_REP_MAX_GROUPS);
return NIL;
}
foreach(lc, t_thrd.syncrepgram_cxt.syncrep_parse_result) {
SyncRepConfigData* conf = (SyncRepConfigData*)lfirst(lc);
bind_transform_context(tcxt, conf);
if (conf->syncrep_method == SYNC_REP_PRIORITY && tcxt->group_nums > 1) {
GUC_check_errdetail("FIRST rule is not support when sync standby strategy is combinated.");
return NIL;
}
if (!analyze_star_and_num(tcxt)) {
return NIL;
}
if (!analyze_duplicate_names(tcxt)) {
return NIL;
}
if (!analyze_cluster_names(tcxt)) {
return NIL;
}
}
// append into results
MemoryContext old_context = MemoryContextSwitchTo(SESS_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE));
foreach(lc, t_thrd.syncrepgram_cxt.syncrep_parse_result) {
SyncRepConfigData* conf = (SyncRepConfigData*)lfirst(lc);
SyncRepConfigData* result = (SyncRepConfigData*)palloc(conf->config_size);
rc = memcpy_s(result, conf->config_size, conf, conf->config_size);
securec_check(rc, "", "");
results = lappend(results, result);
}
(void)MemoryContextSwitchTo(old_context);
return results;
}
/*
* ===========================================================
* Synchronous Replication functions executed by any process
@ -1496,70 +1751,42 @@ int init_gauss_cluster_config(void)
*/
bool check_synchronous_standby_names(char **newval, void **extra, GucSource source)
{
errno_t rc = EOK;
if (*newval != NULL && (*newval)[0] != '\0') {
int parse_rc;
SyncRepConfigData *pconf = NULL;
syncrep_scanner_yyscan_t yyscanner;
char* data_dir = t_thrd.proc_cxt.DataDir;
char* p = NULL;
List* pconf = NIL;
/* Reset communication variables to ensure a fresh start */
t_thrd.syncrepgram_cxt.syncrep_parse_result = NULL;
t_thrd.syncrepgram_cxt.syncrep_parse_result = NIL;
/* Parse the synchronous_standby_names string */
yyscanner = syncrep_scanner_init(*newval);
parse_rc = syncrep_yyparse(yyscanner);
syncrep_scanner_finish(yyscanner);
if (parse_rc != 0 || t_thrd.syncrepgram_cxt.syncrep_parse_result == NULL) {
if (parse_rc != 0 || t_thrd.syncrepgram_cxt.syncrep_parse_result == NIL) {
GUC_check_errcode(ERRCODE_SYNTAX_ERROR);
GUC_check_errdetail("synchronous_standby_names parser failed");
return false;
}
/* GUC extra value must be malloc'd, not palloc'd */
pconf = (SyncRepConfigData *)MemoryContextAlloc(SESS_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE),
t_thrd.syncrepgram_cxt.syncrep_parse_result->config_size);
if (pconf == NULL)
return false;
rc = memcpy_s(pconf, t_thrd.syncrepgram_cxt.syncrep_parse_result->config_size,
t_thrd.syncrepgram_cxt.syncrep_parse_result,
t_thrd.syncrepgram_cxt.syncrep_parse_result->config_size);
securec_check(rc, "", "");
/*
* analyze and transform synchronous_standby_names content,
* make result in session MEMORY_CONTEXT_STORAGE.
*/
TransContext* tcxt = create_transform_context();
pconf = transform_synchronous_standby_names(tcxt);
free_transform_context(tcxt);
tcxt = NULL;
if (strcmp(pconf->member_names, "*") == 0) {
goto pass;
}
if (pconf->num_sync > pconf->nmembers) {
// The sync number must less or equals to the number of standby node names.
if (pconf == NIL) {
return false;
}
/* get current cluster information from cluster_staic_config */
if (t_thrd.role == WORKER && strcmp(u_sess->attr.attr_common.application_name, "gsql") == 0
&& has_static_config()) {
if (0 != init_gauss_cluster_config()) {
goto pass;
}
} else {
goto pass;
}
p = pconf->member_names;
for (int i = 1; i <= pconf->nmembers; i++) {
if (!CheckDataNameValue(p, data_dir)) {
return false;
}
p += strlen(p) + 1;
}
pass:
*extra = (void *)pconf;
if (t_thrd.syncrepgram_cxt.syncrep_parse_result) {
pfree(t_thrd.syncrepgram_cxt.syncrep_parse_result);
t_thrd.syncrepgram_cxt.syncrep_parse_result = NULL;
list_free_deep(t_thrd.syncrepgram_cxt.syncrep_parse_result);
t_thrd.syncrepgram_cxt.syncrep_parse_result = NIL;
}
/*
@ -1582,16 +1809,35 @@ void assign_synchronous_standby_names(const char *newval, void *extra)
* it should be safe to know the latest rep config ASAP for all sessions.
* If this assumption no longer holds, please move it to session level.
*/
pfree_ext(t_thrd.syncrep_cxt.SyncRepConfig);
if (extra != NULL) {
SyncRepConfigData *pconf = (SyncRepConfigData *)extra;
errno_t rc = EOK;
t_thrd.syncrep_cxt.SyncRepConfig = (SyncRepConfigData *)MemoryContextAlloc(
THREAD_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE), pconf->config_size);
rc = memcpy_s(t_thrd.syncrep_cxt.SyncRepConfig, pconf->config_size, pconf, pconf->config_size);
securec_check(rc, "", "");
for (int i = 0; i < t_thrd.syncrep_cxt.SyncRepConfigGroups; i++) {
pfree_ext(t_thrd.syncrep_cxt.SyncRepConfig[i]);
}
pfree_ext(t_thrd.syncrep_cxt.SyncRepConfig);
t_thrd.syncrep_cxt.SyncRepConfigGroups = 0;
if (extra == NULL)
return;
List* pconf_list = (List*)extra;
ListCell* lc = NULL;
int i = 0;
errno_t rc = EOK;
MemoryContext old_context = MemoryContextSwitchTo(THREAD_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE));
t_thrd.syncrep_cxt.SyncRepConfigGroups = list_length(pconf_list);
t_thrd.syncrep_cxt.SyncRepConfig = (SyncRepConfigData **)palloc(
t_thrd.syncrep_cxt.SyncRepConfigGroups * sizeof(SyncRepConfigData*));
foreach(lc, pconf_list) {
SyncRepConfigData* pconf = (SyncRepConfigData*)lfirst(lc);
t_thrd.syncrep_cxt.SyncRepConfig[i] = (SyncRepConfigData*)palloc(pconf->config_size);
rc = memcpy_s(t_thrd.syncrep_cxt.SyncRepConfig[i], pconf->config_size, pconf, pconf->config_size);
securec_check(rc, "", "");
i++;
}
(void)MemoryContextSwitchTo(old_context);
}
void assign_synchronous_commit(int newval, void *extra)

19
src/gausskernel/storage/replication/syncrep_gram.y
View File

@ -66,8 +66,8 @@ static void syncrep_yyerror(YYLTYPE *yylloc,
%token <str> NAME NUM JUNK ANY FIRST
%type <config> result standby_config
%type <list> standby_list
%type <config> standby_config_simplify standby_config_complete
%type <list> result standby_config standby_config_combination standby_list
%type <str> standby_name
%start result
@ -78,9 +78,22 @@ result:
;
standby_config:
standby_config_simplify { $$ = list_make1($1); }
| standby_config_combination { $$ = $1; }
;
standby_config_combination:
standby_config_complete { $$ = list_make1($1); }
| standby_config_combination ',' standby_config_complete { $$ = lappend($1, $3); }
;
standby_config_simplify:
standby_list { $$ = create_syncrep_config("1", $1, SYNC_REP_PRIORITY); }
| NUM '(' standby_list ')' { $$ = create_syncrep_config($1, $3, SYNC_REP_PRIORITY); pfree($1); }
| ANY NUM '(' standby_list ')' { $$ = create_syncrep_config($2, $4, SYNC_REP_QUORUM); pfree($2); }
;
standby_config_complete:
ANY NUM '(' standby_list ')' { $$ = create_syncrep_config($2, $4, SYNC_REP_QUORUM); pfree($2); }
| FIRST NUM '(' standby_list ')' { $$ = create_syncrep_config($2, $4, SYNC_REP_PRIORITY); pfree($2); }
;

44
src/gausskernel/storage/replication/walsender.cpp
View File

@ -3638,6 +3638,7 @@ static void InitWalSnd(void)
securec_check(rc, "", "");
rc = memset_s((void *)&walsnd->wal_sender_channel, sizeof(ReplConnInfo), 0, sizeof(ReplConnInfo));
securec_check(rc, "", "");
walsnd->sync_standby_group = 0;
walsnd->sync_standby_priority = 0;
walsnd->index = i;
walsnd->log_ctrl.sleep_time = 0;
@ -4411,8 +4412,7 @@ bool WalSndInProgress(int type)
bool WalSndQuorumInProgress(int type)
{
int i;
int num = 0;
int num_sync = t_thrd.syncrep_cxt.SyncRepConfig->num_sync;
int* nums = (int*)palloc0(t_thrd.syncrep_cxt.SyncRepConfigGroups * sizeof(int));
for (i = 0; i < g_instance.attr.attr_storage.max_wal_senders; i++) {
/* use volatile pointer to prevent code rearrangement */
@ -4420,15 +4420,19 @@ bool WalSndQuorumInProgress(int type)
SpinLockAcquire(&walsnd->mutex);
if (walsnd->pid != 0 && walsnd->pid != t_thrd.proc_cxt.MyProcPid &&
((walsnd->sendRole & type) == walsnd->sendRole)) {
num++;
nums[walsnd->sync_standby_group]++;
}
SpinLockRelease(&walsnd->mutex);
}
if (num_sync <= num) {
return true;
} else {
return false;
for (i = 0; i < t_thrd.syncrep_cxt.SyncRepConfigGroups; i++) {
if (t_thrd.syncrep_cxt.SyncRepConfig[i]->num_sync > nums[i]) {
pfree(nums);
return false;
}
}
pfree(nums);
return true;
}
/* check if there is all wal sender in type status. */
@ -4601,7 +4605,7 @@ static void set_xlog_location(ServerMode local_role, XLogRecPtr* sndWrite, XLogR
*/
Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_WAL_SENDERS_COLS 21
#define PG_STAT_GET_WAL_SENDERS_COLS 22
TupleDesc tupdesc;
Tuplestorestate *tupstore = NULL;
@ -4609,6 +4613,7 @@ Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
int i = 0;
volatile HaShmemData *hashmdata = t_thrd.postmaster_cxt.HaShmData;
List *sync_standbys = NIL;
ListCell *lc = NULL;
tupstore = BuildTupleResult(fcinfo, &tupdesc);
@ -4667,6 +4672,7 @@ Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
int j = 0;
errno_t rc = 0;
int ret = 0;
int group = 0;
int priority = 0;
SpinLockAcquire(&hashmdata->mutex);
@ -4699,8 +4705,10 @@ Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
syncStart = walsnd->syncPercentCountStart;
catchup_time[0] = walsnd->catchupTime[0];
catchup_time[1] = walsnd->catchupTime[1];
if (IS_DN_MULTI_STANDYS_MODE())
if (IS_DN_MULTI_STANDYS_MODE()) {
group = walsnd->sync_standby_group;
priority = walsnd->sync_standby_priority;
}
SpinLockRelease(&walsnd->mutex);
set_xlog_location(local_role, &sndWrite, &sndFlush, &sndReplay);
@ -4822,9 +4830,11 @@ Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
/* sync_state and sync_prority */
if (!SyncRepRequested()) {
values[j++] = CStringGetTextDatum("Async");
nulls[j++] = true;
values[j++] = Int32GetDatum(0);
} else {
values[j++] = CStringGetTextDatum("Sync");
nulls[j++] = true;
values[j++] = Int32GetDatum(sync_priority[i]);
}
} else {
@ -4844,14 +4854,19 @@ Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
* basically useless to report "sync" or "potential" as their sync
* states. We report just "quorum" for them.
*/
if (priority == 0)
if (priority == 0) {
values[j++] = CStringGetTextDatum("Async");
else if (list_member_int(sync_standbys, i)) {
values[j++] = t_thrd.syncrep_cxt.SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY
nulls[j++] = true;
} else if (list_member_int((List*)list_nth(sync_standbys, group), i)) {
values[j++] = GetWalsndSyncRepConfig(walsnd)->syncrep_method == SYNC_REP_PRIORITY
? CStringGetTextDatum("Sync")
: CStringGetTextDatum("Quorum");
} else
values[j++] = Int32GetDatum(group);
} else {
values[j++] = CStringGetTextDatum("Potential");
values[j++] = Int32GetDatum(group);
}
values[j++] = Int32GetDatum(priority);
}
@ -4869,6 +4884,9 @@ Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
tuplestore_putvalues(tupstore, tupdesc, values, nulls);
}
foreach(lc, sync_standbys) {
list_free((List*)lfirst(lc));
}
list_free(sync_standbys);
if (sync_priority != NULL) {
pfree(sync_priority);

29
src/include/catalog/upgrade_sql/rollback_catalog_maindb/rollback_catalog_maindb_92_301.sql Normal file
View File

@ -0,0 +1,29 @@
DROP FUNCTION IF EXISTS pg_catalog.pg_stat_get_wal_senders() CASCADE;
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 3099;
CREATE FUNCTION pg_catalog.pg_stat_get_wal_senders(
OUT pid bigint,
OUT sender_pid integer,
OUT local_role text,
OUT peer_role text,
OUT peer_state text,
OUT state text,
OUT catchup_start timestamp with time zone,
OUT catchup_end timestamp with time zone,
OUT sender_sent_location text,
OUT sender_write_location text,
OUT sender_flush_location text,
OUT sender_replay_location text,
OUT receiver_received_location text,
OUT receiver_write_location text,
OUT receiver_flush_location text,
OUT receiver_replay_location text,
OUT sync_percent text,
OUT sync_state text,
OUT sync_priority integer,
OUT sync_most_available text,
OUT channel text
) RETURNS SETOF record
STABLE NOT FENCED NOT SHIPPABLE ROWS 10
LANGUAGE internal AS 'pg_stat_get_wal_senders';
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 0;

29
src/include/catalog/upgrade_sql/rollback_catalog_otherdb/rollback_catalog_otherdb_92_301.sql Normal file
View File

@ -0,0 +1,29 @@
DROP FUNCTION IF EXISTS pg_catalog.pg_stat_get_wal_senders() CASCADE;
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 3099;
CREATE FUNCTION pg_catalog.pg_stat_get_wal_senders(
OUT pid bigint,
OUT sender_pid integer,
OUT local_role text,
OUT peer_role text,
OUT peer_state text,
OUT state text,
OUT catchup_start timestamp with time zone,
OUT catchup_end timestamp with time zone,
OUT sender_sent_location text,
OUT sender_write_location text,
OUT sender_flush_location text,
OUT sender_replay_location text,
OUT receiver_received_location text,
OUT receiver_write_location text,
OUT receiver_flush_location text,
OUT receiver_replay_location text,
OUT sync_percent text,
OUT sync_state text,
OUT sync_priority integer,
OUT sync_most_available text,
OUT channel text
) RETURNS SETOF record
STABLE NOT FENCED NOT SHIPPABLE ROWS 10
LANGUAGE internal AS 'pg_stat_get_wal_senders';
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 0;

30
src/include/catalog/upgrade_sql/upgrade_catalog_maindb/upgrade-post_catalog_maindb_92_301.sql Normal file
View File

@ -0,0 +1,30 @@
DROP FUNCTION IF EXISTS pg_catalog.pg_stat_get_wal_senders() CASCADE;
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 3099;
CREATE FUNCTION pg_catalog.pg_stat_get_wal_senders(
OUT pid bigint,
OUT sender_pid integer,
OUT local_role text,
OUT peer_role text,
OUT peer_state text,
OUT state text,
OUT catchup_start timestamp with time zone,
OUT catchup_end timestamp with time zone,
OUT sender_sent_location text,
OUT sender_write_location text,
OUT sender_flush_location text,
OUT sender_replay_location text,
OUT receiver_received_location text,
OUT receiver_write_location text,
OUT receiver_flush_location text,
OUT receiver_replay_location text,
OUT sync_percent text,
OUT sync_state text,
OUT sync_group integer,
OUT sync_priority integer,
OUT sync_most_available text,
OUT channel text
) RETURNS SETOF record
STABLE NOT FENCED NOT SHIPPABLE ROWS 10
LANGUAGE internal AS 'pg_stat_get_wal_senders';
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 0;

30
src/include/catalog/upgrade_sql/upgrade_catalog_otherdb/upgrade-post_catalog_otherdb_92_301.sql Normal file
View File

@ -0,0 +1,30 @@
DROP FUNCTION IF EXISTS pg_catalog.pg_stat_get_wal_senders() CASCADE;
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 3099;
CREATE FUNCTION pg_catalog.pg_stat_get_wal_senders(
OUT pid bigint,
OUT sender_pid integer,
OUT local_role text,
OUT peer_role text,
OUT peer_state text,
OUT state text,
OUT catchup_start timestamp with time zone,
OUT catchup_end timestamp with time zone,
OUT sender_sent_location text,
OUT sender_write_location text,
OUT sender_flush_location text,
OUT sender_replay_location text,
OUT receiver_received_location text,
OUT receiver_write_location text,
OUT receiver_flush_location text,
OUT receiver_replay_location text,
OUT sync_percent text,
OUT sync_state text,
OUT sync_group integer,
OUT sync_priority integer,
OUT sync_most_available text,
OUT channel text
) RETURNS SETOF record
STABLE NOT FENCED NOT SHIPPABLE ROWS 10
LANGUAGE internal AS 'pg_stat_get_wal_senders';
SET LOCAL inplace_upgrade_next_system_object_oids = IUO_PROC, 0;

5
src/include/knl/knl_thread.h
View File

@ -2220,7 +2220,7 @@ typedef struct knl_t_replscanner_context {
typedef struct knl_t_syncrepgram_context {
/* Result of parsing is returned in one of these two variables */
struct SyncRepConfigData* syncrep_parse_result;
List* syncrep_parse_result;
} knl_t_syncrepgram_context;
typedef struct knl_t_syncrepscanner_context {
@ -2230,7 +2230,8 @@ typedef struct knl_t_syncrepscanner_context {
} knl_t_syncrepscanner_context;
typedef struct knl_t_syncrep_context {
struct SyncRepConfigData* SyncRepConfig;
struct SyncRepConfigData** SyncRepConfig;
int SyncRepConfigGroups;
bool announce_next_takeover;
} knl_t_syncrep_context;

5
src/include/replication/syncrep.h
View File

@ -38,11 +38,16 @@
#define SYNC_REP_PRIORITY 0
#define SYNC_REP_QUORUM 1
#define SYNC_REP_MAX_GROUPS 256
extern volatile bool most_available_sync;
#define SyncStandbysDefined() \
(u_sess->attr.attr_storage.SyncRepStandbyNames != NULL && u_sess->attr.attr_storage.SyncRepStandbyNames[0] != '\0')
#define GetWalsndSyncRepConfig(walsnder) \
(t_thrd.syncrep_cxt.SyncRepConfig[(walsnder)->sync_standby_group])
/*
* Struct for the configuration of synchronous replication.
*

7
src/include/replication/walsender_private.h
View File

@ -109,10 +109,11 @@ typedef struct WalSnd {
Latch latch;
/*
* The priority order of the standby managed by this WALSender, as listed
* in synchronous_standby_names, or 0 if not-listed. Protected by
* SyncRepLock.
* The strategy group and priority order of the standby managed by this WALSender,
* as listed in synchronous_standby_names, or 0 if not-listed.
* Protected by SyncRepLock.
*/
uint8 sync_standby_group;
int sync_standby_priority;
int index;
XLogRecPtr arch_task_lsn;

150
src/test/regress/expected/sync_standy_names.out Normal file
View File

@ -0,0 +1,150 @@
show synchronous_standby_names;
synchronous_standby_names
---------------------------
*
(1 row)
-- single gram
alter system set synchronous_standby_names = '*'; -- suc
alter system set synchronous_standby_names = 'd1, *'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "d1, *"
DETAIL: Please use the '*' separately, otherwise, it is semantic ambiguity.
alter system set synchronous_standby_names = '*, *'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "*, *"
DETAIL: Please use the '*' separately, otherwise, it is semantic ambiguity.
alter system set synchronous_standby_names = 'd1'; -- suc
alter system set synchronous_standby_names = 'd1, d2, d3, d4'; -- suc
alter system set synchronous_standby_names = 'd1, d2, d1, d4'; -- err, duplicate name
ERROR: invalid value for parameter "synchronous_standby_names": "d1, d2, d1, d4"
DETAIL: Duplicate standby node name: d1
alter system set synchronous_standby_names = '2 (*)'; -- suc
alter system set synchronous_standby_names = '2 (d1, *)'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "2 (d1, *)"
DETAIL: Please use the '*' separately, otherwise, it is semantic ambiguity.
alter system set synchronous_standby_names = '2 (*, *)'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "2 (*, *)"
DETAIL: Please use the '*' separately, otherwise, it is semantic ambiguity.
alter system set synchronous_standby_names = '2 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = '2 (d1, d2, d1, d4)'; -- err, duplicate name
ERROR: invalid value for parameter "synchronous_standby_names": "2 (d1, d2, d1, d4)"
DETAIL: Duplicate standby node name: d1
alter system set synchronous_standby_names = '5 (d1, d2, d3, d4)'; -- err, requriement more than have
ERROR: invalid value for parameter "synchronous_standby_names": "5 (d1, d2, d3, d4)"
DETAIL: The sync number must less or equals to the number of standby node names.
alter system set synchronous_standby_names = '0 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = '-1 (d1, d2, d3, d4)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "-1 (d1, d2, d3, d4)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = 'first 2 (*)'; -- suc
alter system set synchronous_standby_names = 'first 2 (d1, *)'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "first 2 (d1, *)"
DETAIL: Please use the '*' separately, otherwise, it is semantic ambiguity.
alter system set synchronous_standby_names = 'first 2 (*, *)'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "first 2 (*, *)"
DETAIL: Please use the '*' separately, otherwise, it is semantic ambiguity.
alter system set synchronous_standby_names = 'first 2 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = 'first 2 (d1, d2, d1, d4)'; -- err, duplicate name
ERROR: invalid value for parameter "synchronous_standby_names": "first 2 (d1, d2, d1, d4)"
DETAIL: Duplicate standby node name: d1
alter system set synchronous_standby_names = 'first 5 (d1, d2, d3, d4)'; -- err, requriement more than have
ERROR: invalid value for parameter "synchronous_standby_names": "first 5 (d1, d2, d3, d4)"
DETAIL: The sync number must less or equals to the number of standby node names.
alter system set synchronous_standby_names = 'first 0 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = 'first -1 (d1, d2, d3, d4)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "first -1 (d1, d2, d3, d4)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = 'any 2 (*)'; -- suc
alter system set synchronous_standby_names = 'any 2 (d1, *'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, *"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = 'any 2 (*, *)'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (*, *)"
DETAIL: Please use the '*' separately, otherwise, it is semantic ambiguity.
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = 'any 2 (d1, d2, d1, d4)'; -- err, duplicate name
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d1, d4)"
DETAIL: Duplicate standby node name: d1
alter system set synchronous_standby_names = 'any 5 (d1, d2, d3, d4)'; -- err, requriement more than have
ERROR: invalid value for parameter "synchronous_standby_names": "any 5 (d1, d2, d3, d4)"
DETAIL: The sync number must less or equals to the number of standby node names.
alter system set synchronous_standby_names = 'any 0 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = 'any -1 (d1, d2, d3, d4)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "any -1 (d1, d2, d3, d4)"
DETAIL: synchronous_standby_names parser failed
-- combinate gram
alter system set synchronous_standby_names = 'd1, d2, d3, 1 (d5, d6)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "d1, d2, d3, 1 (d5, d6)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = '1 (d1, d2, d3), 1 (d5, d6)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "1 (d1, d2, d3), 1 (d5, d6)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = '*, 1 (d5, d6)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "*, 1 (d5, d6)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = '1 (d1, d2, d3), 1 (*)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "1 (d1, d2, d3), 1 (*)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = '1 (d1, d2, d3), 1 (*)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "1 (d1, d2, d3), 1 (*)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = 'any 2 (*), any 2 (d1, d2, d3, d4), any 2 (d5, d6, d7)'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (*), any 2 (d1, d2, d3, d4), any 2 (d5, d6, d7)"
DETAIL: '*' is not support when sync standby strategy is combinated.
alter system set synchronous_standby_names = 'any 2 (d, dd), any 2 (d1, *, d3, d4), any 2 (d5, d6, d7)'; -- err, semantic ambiguity
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d, dd), any 2 (d1, *, d3, d4), any 2 (d5, d6, d7)"
DETAIL: '*' is not support when sync standby strategy is combinated.
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), d5, d6, d7'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d3, d4), d5, d6, d7"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), 2 (d5, d1, d7)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d3, d4), 2 (d5, d1, d7)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), *'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d3, d4), *"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = '2 (d1, d2, d3, d4), first 2 (d5, d6, d7)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "2 (d1, d2, d3, d4), first 2 (d5, d6, d7)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = '2 (d1, d2, d3, d4), any 4 (d5, d6, d7)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "2 (d1, d2, d3, d4), any 4 (d5, d6, d7)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = '*, any 0 (d5, d6, d7)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "*, any 0 (d5, d6, d7)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = 'd1, d2, d3, d4, any 2 (d5, d6, d7)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "d1, d2, d3, d4, any 2 (d5, d6, d7)"
DETAIL: synchronous_standby_names parser failed
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 2 (d5, d6, d7)'; -- suc
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 2 (d5, d1, d7)'; -- err, duplicate name
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d3, d4), any 2 (d5, d1, d7)"
DETAIL: Duplicate standby node name: d1
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 2 (d5, d6, d6)'; -- err, duplicate name
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d3, d4), any 2 (d5, d6, d6)"
DETAIL: Duplicate standby node name: d6
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), first 2 (d5, d6, d7)'; -- err, not support first
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d3, d4), first 2 (d5, d6, d7)"
DETAIL: FIRST rule is not support when sync standby strategy is combinated.
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 4 (d5, d6, d7)'; -- err, requriement more than have
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d3, d4), any 4 (d5, d6, d7)"
DETAIL: The sync number must less or equals to the number of standby node names.
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 0 (d5, d6, d7)'; -- suc
alter system set synchronous_standby_names = 'first 2 (d1, d2, d3, d4), first 2 (d5, d6, d7)'; -- err, not support first
ERROR: invalid value for parameter "synchronous_standby_names": "first 2 (d1, d2, d3, d4), first 2 (d5, d6, d7)"
DETAIL: FIRST rule is not support when sync standby strategy is combinated.
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), , any 0 (d5, d6, d7)'; -- err, parser err
ERROR: invalid value for parameter "synchronous_standby_names": "any 2 (d1, d2, d3, d4), , any 0 (d5, d6, d7)"
DETAIL: synchronous_standby_names parser failed
-- recover
alter system set synchronous_standby_names = '*';
select pg_sleep(3);
pg_sleep
----------
(1 row)
show synchronous_standby_names;
synchronous_standby_names
---------------------------
*
(1 row)

2
src/test/regress/parallel_schedule0
View File

@ -10,7 +10,7 @@ test: analyze_commands
test: single_node_ddl
test: single_node_sqlbypass
test: median
test: sync_standy_names
# run tablespace by itself, and first, because it forces a checkpoint;
# we'd prefer not to have checkpoints later in the tests because that

62
src/test/regress/sql/sync_standy_names.sql Normal file
View File

@ -0,0 +1,62 @@
show synchronous_standby_names;
-- single gram
alter system set synchronous_standby_names = '*'; -- suc
alter system set synchronous_standby_names = 'd1, *'; -- err, semantic ambiguity
alter system set synchronous_standby_names = '*, *'; -- err, semantic ambiguity
alter system set synchronous_standby_names = 'd1'; -- suc
alter system set synchronous_standby_names = 'd1, d2, d3, d4'; -- suc
alter system set synchronous_standby_names = 'd1, d2, d1, d4'; -- err, duplicate name
alter system set synchronous_standby_names = '2 (*)'; -- suc
alter system set synchronous_standby_names = '2 (d1, *)'; -- err, semantic ambiguity
alter system set synchronous_standby_names = '2 (*, *)'; -- err, semantic ambiguity
alter system set synchronous_standby_names = '2 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = '2 (d1, d2, d1, d4)'; -- err, duplicate name
alter system set synchronous_standby_names = '5 (d1, d2, d3, d4)'; -- err, requriement more than have
alter system set synchronous_standby_names = '0 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = '-1 (d1, d2, d3, d4)'; -- err, parser err
alter system set synchronous_standby_names = 'first 2 (*)'; -- suc
alter system set synchronous_standby_names = 'first 2 (d1, *)'; -- err, semantic ambiguity
alter system set synchronous_standby_names = 'first 2 (*, *)'; -- err, semantic ambiguity
alter system set synchronous_standby_names = 'first 2 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = 'first 2 (d1, d2, d1, d4)'; -- err, duplicate name
alter system set synchronous_standby_names = 'first 5 (d1, d2, d3, d4)'; -- err, requriement more than have
alter system set synchronous_standby_names = 'first 0 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = 'first -1 (d1, d2, d3, d4)'; -- err, parser err
alter system set synchronous_standby_names = 'any 2 (*)'; -- suc
alter system set synchronous_standby_names = 'any 2 (d1, *'; -- err, semantic ambiguity
alter system set synchronous_standby_names = 'any 2 (*, *)'; -- err, semantic ambiguity
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = 'any 2 (d1, d2, d1, d4)'; -- err, duplicate name
alter system set synchronous_standby_names = 'any 5 (d1, d2, d3, d4)'; -- err, requriement more than have
alter system set synchronous_standby_names = 'any 0 (d1, d2, d3, d4)'; -- suc
alter system set synchronous_standby_names = 'any -1 (d1, d2, d3, d4)'; -- err, parser err
-- combinate gram
alter system set synchronous_standby_names = 'd1, d2, d3, 1 (d5, d6)'; -- err, parser err
alter system set synchronous_standby_names = '1 (d1, d2, d3), 1 (d5, d6)'; -- err, parser err
alter system set synchronous_standby_names = '*, 1 (d5, d6)'; -- err, parser err
alter system set synchronous_standby_names = '1 (d1, d2, d3), 1 (*)'; -- err, parser err
alter system set synchronous_standby_names = '1 (d1, d2, d3), 1 (*)'; -- err, parser err
alter system set synchronous_standby_names = 'any 2 (*), any 2 (d1, d2, d3, d4), any 2 (d5, d6, d7)'; -- err, semantic ambiguity
alter system set synchronous_standby_names = 'any 2 (d, dd), any 2 (d1, *, d3, d4), any 2 (d5, d6, d7)'; -- err, semantic ambiguity
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), d5, d6, d7'; -- err, parser err
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), 2 (d5, d1, d7)'; -- err, parser err
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), *'; -- err, parser err
alter system set synchronous_standby_names = '2 (d1, d2, d3, d4), first 2 (d5, d6, d7)'; -- err, parser err
alter system set synchronous_standby_names = '2 (d1, d2, d3, d4), any 4 (d5, d6, d7)'; -- err, parser err
alter system set synchronous_standby_names = '*, any 0 (d5, d6, d7)'; -- err, parser err
alter system set synchronous_standby_names = 'd1, d2, d3, d4, any 2 (d5, d6, d7)'; -- err, parser err
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 2 (d5, d6, d7)'; -- suc
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 2 (d5, d1, d7)'; -- err, duplicate name
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 2 (d5, d6, d6)'; -- err, duplicate name
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), first 2 (d5, d6, d7)'; -- err, not support first
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 4 (d5, d6, d7)'; -- err, requriement more than have
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), any 0 (d5, d6, d7)'; -- suc
alter system set synchronous_standby_names = 'first 2 (d1, d2, d3, d4), first 2 (d5, d6, d7)'; -- err, not support first
alter system set synchronous_standby_names = 'any 2 (d1, d2, d3, d4), , any 0 (d5, d6, d7)'; -- err, parser err
-- recover
alter system set synchronous_standby_names = '*';
select pg_sleep(3);
show synchronous_standby_names;