xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix:error in multi thread in schemaless

This commit is contained in:
wangmm0220 2022-06-11 19:40:56 +08:00
parent 40b841f704
commit 9bccbbf865
3 changed files with 38 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
source/client/src/clientSml.c
View File

@ -473,6 +473,21 @@ static int32_t smlProcessSchemaAction(SSmlHandle* info, SSchema* schemaField, SH
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t smlCheckMeta(SSchema* schema, int32_t length, SArray* cols){
SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
for(uint16_t i = 0; i < length; i++){
taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES);
}
for(int32_t i = 0; i < taosArrayGetSize(cols); i++){
SSmlKv* kv = (SSmlKv*)taosArrayGetP(cols, i);
if(taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL){
return -1;
}
}
return 0;
}
static int32_t smlModifyDBSchemas(SSmlHandle* info) { static int32_t smlModifyDBSchemas(SSmlHandle* info) {
int32_t code = 0; int32_t code = 0;
SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp); SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
@ -483,6 +498,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
while (tableMetaSml) { while (tableMetaSml) {
SSmlSTableMeta* sTableData = *tableMetaSml; SSmlSTableMeta* sTableData = *tableMetaSml;
STableMeta *pTableMeta = NULL; STableMeta *pTableMeta = NULL;
bool needCheckMeta = false; // for multi thread
size_t superTableLen = 0; size_t superTableLen = 0;
void *superTable = taosHashGetKey(tableMetaSml, &superTableLen); void *superTable = taosHashGetKey(tableMetaSml, &superTableLen);
@ -533,6 +549,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto end; goto end;
} }
needCheckMeta = true;
} else { } else {
uError("SML:0x%"PRIx64" load table meta error: %s", info->id, tstrerror(code)); uError("SML:0x%"PRIx64" load table meta error: %s", info->id, tstrerror(code));
goto end; goto end;
@ -544,6 +561,20 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
uError("SML:0x%"PRIx64" catalogGetSTableMeta failed. super table name %s", info->id, (char*)superTable); uError("SML:0x%"PRIx64" catalogGetSTableMeta failed. super table name %s", info->id, (char*)superTable);
goto end; goto end;
} }
if(needCheckMeta){
code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags, sTableData->tags);
if (code != TSDB_CODE_SUCCESS) {
uError("SML:0x%"PRIx64" check tag failed. super table name %s", info->id, (char*)superTable);
goto end;
}
code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols);
if (code != TSDB_CODE_SUCCESS) {
uError("SML:0x%"PRIx64" check cols failed. super table name %s", info->id, (char*)superTable);
goto end;
}
}
sTableData->tableMeta = pTableMeta; sTableData->tableMeta = pTableMeta;
tableMetaSml = (SSmlSTableMeta**)taosHashIterate(info->superTables, tableMetaSml); tableMetaSml = (SSmlSTableMeta**)taosHashIterate(info->superTables, tableMetaSml);
@ -2368,6 +2399,7 @@ static void smlInsertCallback(void* param, void* res, int32_t code) {
SRequestObj *pRequest = (SRequestObj *)res; SRequestObj *pRequest = (SRequestObj *)res;
SSmlHandle* info = (SSmlHandle *)param; SSmlHandle* info = (SSmlHandle *)param;
uDebug("SML:0x%"PRIx64" result. code:%d, msg:%s", info->id, pRequest->code, pRequest->msgBuf);
// lock // lock
if(code != TSDB_CODE_SUCCESS){ if(code != TSDB_CODE_SUCCESS){
taosThreadSpinLock(&info->params->lock); taosThreadSpinLock(&info->params->lock);
@ -2497,7 +2529,7 @@ end:
taosThreadSpinDestroy(&params.lock); taosThreadSpinDestroy(&params.lock);
tsem_destroy(&params.sem); tsem_destroy(&params.sem);
((STscObj *)taos)->schemalessType = 0; ((STscObj *)taos)->schemalessType = 0;
uDebug("result:%s", request->msgBuf); uDebug("resultend:%s", request->msgBuf);
return (TAOS_RES*)request; return (TAOS_RES*)request;
} }

2
source/libs/parser/src/parInsert.c
View File

@ -2119,9 +2119,11 @@ static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SS
isOrdered = false; isOrdered = false;
} }
if (index < 0) { if (index < 0) {
uError("smlBoundColumnData. index:%d", index);
return TSDB_CODE_SML_INVALID_DATA; return TSDB_CODE_SML_INVALID_DATA;
} }
if (pColList->cols[index].valStat == VAL_STAT_HAS) { if (pColList->cols[index].valStat == VAL_STAT_HAS) {
uError("smlBoundColumnData. already set. index:%d", index);
return TSDB_CODE_SML_INVALID_DATA; return TSDB_CODE_SML_INVALID_DATA;
} }
lastColIdx = index; lastColIdx = index;

3
source/libs/parser/src/parTranslater.c
View File

@ -4945,6 +4945,9 @@ static int32_t buildModifyVnodeArray(STranslateContext* pCxt, SAlterTableStmt* p
static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) { static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) {
SAlterTableStmt* pStmt = (SAlterTableStmt*)pQuery->pRoot; SAlterTableStmt* pStmt = (SAlterTableStmt*)pQuery->pRoot;
int32_t code = checkSchemalessDb(pCxt, pStmt->dbName); int32_t code = checkSchemalessDb(pCxt, pStmt->dbName);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
STableMeta* pTableMeta = NULL; STableMeta* pTableMeta = NULL;
code = getTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pTableMeta); code = getTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pTableMeta);
if (TSDB_CODE_SUCCESS != code) { if (TSDB_CODE_SUCCESS != code) {