hcq/openGauss-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

!6737 修复HTAP备节点偶现获取列信息问题

Browse Source 
Merge pull request !6737 from sovyyvos/master
This commit is contained in:
opengauss_bot
2024-11-29 06:09:28 +00:00
committed by Gitee
parent 466faceb39 54a66623d6
commit 6f9a7bb9fc
3 changed files with 24 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/gausskernel/optimizer/commands/tablecmds.cpp
View File

@ -22394,16 +22394,13 @@ static void ATExecIMCSTORED(Relation rel, List* colList)
{
Oid relOid = RelationGetRelid(rel);
int2vector* imcsAtts = NULL;
List* colNames = NIL;
int imcsNatts = 0;
int nameLength = 0;
CheckForEnableImcs(rel, colList, imcsAtts, &imcsNatts);
/* standbynode populate */
if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
GetColNamesForStandBy(rel, imcsAtts, imcsNatts, colNames, &nameLength);
CreateImcsDescForPrimaryNode(rel, imcsAtts, imcsNatts);
SendImcstoredRequest(relOid, InvalidOid, colNames, nameLength, TYPE_IMCSTORED);
SendImcstoredRequest(relOid, InvalidOid, imcsAtts->values, imcsNatts, TYPE_IMCSTORED);
} else {
AlterTableEnableImcstore(rel, imcsAtts, imcsNatts);
}
@ -22434,9 +22431,7 @@ static void ATExecModifyPartitionIMCSTORED(Relation rel, const char* partName, L
Oid partOid = InvalidOid;
Oid relOid = RelationGetRelid(rel);
int2vector* imcsAtts = NULL;
List* colNames = NIL;
int imcsNatts = 0;
int nameLength = 0;
partOid = ImcsPartNameGetPartOid(relOid, partName);
CheckForEnableImcs(rel, colList, imcsAtts, &imcsNatts, partOid);
@ -22456,9 +22451,8 @@ static void ATExecModifyPartitionIMCSTORED(Relation rel, const char* partName, L
{
/* populate on standbynode */
if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
GetColNamesForStandBy(partRel, imcsAtts, imcsNatts, colNames, &nameLength);
CreateImcsDescForPrimaryNode(partRel, imcsAtts, imcsNatts);
SendImcstoredRequest(relOid, partOid, colNames, nameLength, TYPE_PARTITION_IMCSTORED);
SendImcstoredRequest(relOid, partOid, imcsAtts->values, imcsNatts, TYPE_PARTITION_IMCSTORED);
} else {
/* populate on currrent node */
AlterTableEnableImcstore(partRel, imcsAtts, imcsNatts);

63
src/gausskernel/storage/htap/imcs_ctlg.cpp
View File

@ -582,17 +582,6 @@ void UnPopulateImcs(Relation rel)
PG_END_TRY();
}
void GetColNamesForStandBy(Relation rel, int2vector* imcsAttsNum, int imcsNatts, List* &colList, int* nameLength)
{
Assert(imcsAttsNum != NULL && imcsNatts > 0);
FormData_pg_attribute *rel_atts = rel->rd_att->attrs;
for (int i = 0; i < imcsNatts; i++) {
AttrNumber attnumber = imcsAttsNum->values[i];
colList = lappend(colList, makeString(NameStr(rel_atts[attnumber - 1].attname)));
*nameLength += (strlen(NameStr(rel_atts[attnumber - 1].attname)) + 1);
}
}
void PopulateImcsOnStandby(Oid relOid, StringInfo inputMsg)
{
int imcsNatts = 0;
@ -742,12 +731,9 @@ void ParsePopulateImcsParam(
rc = memcpy_s(&count, sizeof(int), pq_getmsgbytes(inputMsg, sizeof(int)), sizeof(int));
securec_check(rc, "\0", "\0");
char **colList = NULL;
if (count > 0) {
colList = (char **)palloc(count * sizeof(char *));
for (int i = 0; i < count; i++)
colList[i] = (char *)pq_getmsgstring(inputMsg);
}
int2 *attsNums = (int2*)palloc(sizeof(int2) * count);
rc = memcpy_s(attsNums, sizeof(int2) * count, pq_getmsgbytes(inputMsg, sizeof(int2) * count), sizeof(int2) * count);
securec_check(rc, "\0", "\0");
/* get current lsn from primary */
rc = memcpy_s(currentLsn, sizeof(XLogRecPtr), pq_getmsgbytes(inputMsg, sizeof(XLogRecPtr)),
@ -756,14 +742,9 @@ void ParsePopulateImcsParam(
ereport(DEBUG1, (errmsg("Received lsn for HTAP population.")));
pq_getmsgend(inputMsg);
int j = 0;
imcsAttsNum = buildint2vector(NULL, count);
for (int cnt = 0; cnt < count; cnt++) {
AttrNumber attnum = get_attnum(relOid, colList[cnt]);
imcsAttsNum->values[j] = attnum;
j++;
}
imcsAttsNum = buildint2vector(attsNums, count);
*imcsNatts = count;
pfree(attsNums);
return;
}
@ -975,7 +956,8 @@ static void HandlePgxcReceive(int connCount, PGXCNodeHandle **tempConnections)
pfree_ext(tempConnections);
}
static void PackBasicImcstoredRequest(PGXCNodeHandle *temp_connection, SendPopulateParams* populateParams)
static void PackBasicImcstoredRequest(
PGXCNodeHandle *temp_connection, SendPopulateParams* populateParams, int imcsNatts)
{
errno_t ss_rc = EOK;
int msglen = populateParams->msglen;
@ -1004,21 +986,15 @@ static void PackBasicImcstoredRequest(PGXCNodeHandle *temp_connection, SendPopul
securec_check(ss_rc, "\0", "\0");
temp_connection->outEnd += sizeof(Oid);
int colCount = list_length(populateParams->columnlist);
if (colCount > 0) {
if (imcsNatts > 0) {
ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
temp_connection->outSize - temp_connection->outEnd, &colCount, sizeof(int));
temp_connection->outSize - temp_connection->outEnd, &imcsNatts, sizeof(int));
securec_check(ss_rc, "\0", "\0");
temp_connection->outEnd += sizeof(int);
ListCell *cell;
foreach (cell, populateParams->columnlist) {
char *col_name = strVal(lfirst(cell));
int name_len = strlen(col_name) + 1;
ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
temp_connection->outSize - temp_connection->outEnd, col_name, name_len);
securec_check(ss_rc, "\0", "\0");
temp_connection->outEnd += name_len;
}
ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
temp_connection->outSize - temp_connection->outEnd, populateParams->attsNums, sizeof(int2) * imcsNatts);
securec_check(ss_rc, "\0", "\0");
temp_connection->outEnd += sizeof(int2) * imcsNatts;
XLogRecPtr lsn = t_thrd.shemem_ptr_cxt.XLogCtl->LogwrtRqst.Write;
ss_rc = memcpy_s(temp_connection->outBuffer + temp_connection->outEnd,
temp_connection->outSize - temp_connection->outEnd, &lsn, sizeof(XLogRecPtr));
@ -1027,7 +1003,7 @@ static void PackBasicImcstoredRequest(PGXCNodeHandle *temp_connection, SendPopul
}
}
void SendImcstoredRequest(Oid relOid, Oid specifyPartOid, List *columnlist, int length, int type)
void SendImcstoredRequest(Oid relOid, Oid specifyPartOid, int2* attsNums, int imcsNatts, int type)
{
int connCount = 0;
PGXCNodeHandle** connections = NULL;
@ -1036,10 +1012,11 @@ void SendImcstoredRequest(Oid relOid, Oid specifyPartOid, List *columnlist, int
/* init send populate params */
populateParams.relOid = relOid;
populateParams.partOid = specifyPartOid;
populateParams.columnlist = columnlist;
populateParams.attsNums = attsNums;
populateParams.imcstoreType = type;
populateParams.msglen =
sizeof(int) + sizeof(int) + sizeof(Oid) + sizeof(Oid) + sizeof(int) + length + sizeof(XLogRecPtr);
sizeof(int) + sizeof(int) + sizeof(Oid) + sizeof(Oid) + sizeof(int) +
imcsNatts * sizeof(int2) + sizeof(XLogRecPtr);
connections = GetStandbyConnections(&connCount);
PGXCNodeHandle **temp_connections = NULL;
@ -1058,7 +1035,7 @@ void SendImcstoredRequest(Oid relOid, Oid specifyPartOid, List *columnlist, int
temp_connections[i]->remoteNodeName, temp_connections[i]->gsock.idx, temp_connections[i]->gsock.sid,
temp_connections[i]->state);
PackBasicImcstoredRequest(temp_connections[i], &populateParams);
PackBasicImcstoredRequest(temp_connections[i], &populateParams, imcsNatts);
if (pgxc_node_flush(temp_connections[i]) != 0) {
temp_connections[i]->state = DN_CONNECTION_STATE_ERROR_FATAL;
@ -1081,7 +1058,7 @@ void SendUnImcstoredRequest(Oid relOid, Oid specifyPartOid, int type)
/* init send populate params */
populateParams.relOid = relOid;
populateParams.partOid = specifyPartOid;
populateParams.columnlist = NULL;
populateParams.attsNums = NULL;
populateParams.imcstoreType = type;
populateParams.msglen =
sizeof(int) + sizeof(int) + sizeof(Oid) + sizeof(Oid);
@ -1102,7 +1079,7 @@ void SendUnImcstoredRequest(Oid relOid, Oid specifyPartOid, int type)
temp_connections[i]->remoteNodeName, temp_connections[i]->gsock.idx, temp_connections[i]->gsock.sid,
temp_connections[i]->state);
PackBasicImcstoredRequest(temp_connections[i], &populateParams);
PackBasicImcstoredRequest(temp_connections[i], &populateParams, 0);
if (pgxc_node_flush(temp_connections[i]) != 0) {
temp_connections[i]->state = DN_CONNECTION_STATE_ERROR_FATAL;
ereport(ERROR, (errcode(ERRCODE_CONNECTION_EXCEPTION),

7
src/include/access/htap/imcs_ctlg.h
View File

@ -54,7 +54,7 @@
typedef struct SendStandbyPopulateParams {
Oid relOid;
Oid partOid;
List *columnlist;
int2 *attsNums;
int msglen;
int imcstoreType;
} SendPopulateParams;
@ -112,9 +112,6 @@ extern void AlterTableDisableImcstore(Relation rel);
extern void UnPopulateImcs(Relation rel);
extern void GetColNamesForStandBy(
Relation rel, int2vector* imcsAttsNum, int imcsNatts, List* &colList, int* nameLength);
extern void PopulateImcsOnStandby(Oid relOid, StringInfo inputMsg);
extern void PopulateImcsForPartitionOnStandby(Oid relOid, Oid partOid, StringInfo inputMsg);
@ -128,7 +125,7 @@ extern void ParsePopulateImcsParam(
extern PGXCNodeHandle **GetStandbyConnections(int *connCount);
extern void SendImcstoredRequest(Oid relOid, Oid specifyPartOid, List *columnlist, int length, int type);
extern void SendImcstoredRequest(Oid relOid, Oid specifyPartOid, int2* attsNums, int imcsNatts, int type);
extern void SendUnImcstoredRequest(Oid relOid, Oid specifyPartOid, int type);