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

enh:[TS-5441] cost too long in tmq write meta data by cache meta and vg info

This commit is contained in:
wangmm0220 2024-10-28 15:02:03 +08:00
parent 94a70c9389
commit 0100d383ee
1 changed files with 159 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
source/client/src/clientRawBlockWrite.c
View File

@ -163,7 +163,7 @@ static void buildCreateTableJson(SSchemaWrapper* schemaRow, SSchemaWrapper* sche
} }
RAW_FALSE_CHECK(cJSON_AddItemToObject(json, "tags", tags)); RAW_FALSE_CHECK(cJSON_AddItemToObject(json, "tags", tags));
end: end:
*pJson = json; *pJson = json;
} }
@ -197,7 +197,7 @@ static int32_t setCompressOption(cJSON* json, uint32_t para) {
return code; return code;
} }
end: end:
return code; return code;
} }
static void buildAlterSTableJson(void* alterData, int32_t alterDataLen, cJSON** pJson) { static void buildAlterSTableJson(void* alterData, int32_t alterDataLen, cJSON** pJson) {
@ -338,7 +338,7 @@ static void buildAlterSTableJson(void* alterData, int32_t alterDataLen, cJSON**
break; break;
} }
end: end:
tFreeSMAltertbReq(&req); tFreeSMAltertbReq(&req);
*pJson = json; *pJson = json;
} }
@ -485,7 +485,7 @@ static void buildChildElement(cJSON* json, SVCreateTbReq* pCreateReq) {
RAW_FALSE_CHECK(cJSON_AddItemToArray(tags, tag)); RAW_FALSE_CHECK(cJSON_AddItemToArray(tags, tag));
} }
end: end:
RAW_FALSE_CHECK(cJSON_AddItemToObject(json, "tags", tags)); RAW_FALSE_CHECK(cJSON_AddItemToObject(json, "tags", tags));
taosArrayDestroy(pTagVals); taosArrayDestroy(pTagVals);
} }
@ -514,7 +514,7 @@ static void buildCreateCTableJson(SVCreateTbReq* pCreateReq, int32_t nReqs, cJSO
} }
RAW_FALSE_CHECK(cJSON_AddItemToObject(json, "createList", createList)); RAW_FALSE_CHECK(cJSON_AddItemToObject(json, "createList", createList));
end: end:
*pJson = json; *pJson = json;
} }
@ -585,7 +585,7 @@ static void processAutoCreateTable(SMqDataRsp* rsp, char** string) {
*string = cJSON_PrintUnformatted(pJson); *string = cJSON_PrintUnformatted(pJson);
cJSON_Delete(pJson); cJSON_Delete(pJson);
end: end:
uDebug("auto created table return, sql json:%s", *string); uDebug("auto created table return, sql json:%s", *string);
for (int i = 0; decoder && pCreateReq && i < rsp->createTableNum; i++) { for (int i = 0; decoder && pCreateReq && i < rsp->createTableNum; i++) {
tDecoderClear(&decoder[i]); tDecoderClear(&decoder[i]);
@ -989,7 +989,7 @@ static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) {
code = pRequest->code; code = pRequest->code;
end: end:
uDebug(LOG_ID_TAG " create stable return, msg:%s", LOG_ID_VALUE, tstrerror(code)); uDebug(LOG_ID_TAG " create stable return, msg:%s", LOG_ID_VALUE, tstrerror(code));
destroyRequest(pRequest); destroyRequest(pRequest);
tFreeSMCreateStbReq(&pReq); tFreeSMCreateStbReq(&pReq);
@ -1088,7 +1088,7 @@ static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) {
code = pRequest->code; code = pRequest->code;
end: end:
uDebug(LOG_ID_TAG " drop stable return, msg:%s", LOG_ID_VALUE, tstrerror(code)); uDebug(LOG_ID_TAG " drop stable return, msg:%s", LOG_ID_VALUE, tstrerror(code));
destroyRequest(pRequest); destroyRequest(pRequest);
tDecoderClear(&coder); tDecoderClear(&coder);
@ -1269,7 +1269,7 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) {
code = pRequest->code; code = pRequest->code;
end: end:
uDebug(LOG_ID_TAG " create table return, msg:%s", LOG_ID_VALUE, tstrerror(code)); uDebug(LOG_ID_TAG " create table return, msg:%s", LOG_ID_VALUE, tstrerror(code));
tDeleteSVCreateTbBatchReq(&req); tDeleteSVCreateTbBatchReq(&req);
@ -1395,7 +1395,7 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) {
} }
code = pRequest->code; code = pRequest->code;
end: end:
uDebug(LOG_ID_TAG " drop table return, msg:%s", LOG_ID_VALUE, tstrerror(code)); uDebug(LOG_ID_TAG " drop table return, msg:%s", LOG_ID_VALUE, tstrerror(code));
taosHashCleanup(pVgroupHashmap); taosHashCleanup(pVgroupHashmap);
destroyRequest(pRequest); destroyRequest(pRequest);
@ -1433,7 +1433,7 @@ static int32_t taosDeleteData(TAOS* taos, void* meta, int32_t metaLen) {
} }
taos_free_result(res); taos_free_result(res);
end: end:
uDebug("connId:0x%" PRIx64 " delete data sql:%s, code:%s", *(int64_t*)taos, sql, tstrerror(code)); uDebug("connId:0x%" PRIx64 " delete data sql:%s, code:%s", *(int64_t*)taos, sql, tstrerror(code));
tDecoderClear(&coder); tDecoderClear(&coder);
return code; return code;
@ -1543,7 +1543,7 @@ static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) {
code = handleAlterTbExecRes(pRes->res, pCatalog); code = handleAlterTbExecRes(pRes->res, pCatalog);
} }
} }
end: end:
uDebug(LOG_ID_TAG " alter table return, meta:%p, len:%d, msg:%s", LOG_ID_VALUE, meta, metaLen, tstrerror(code)); uDebug(LOG_ID_TAG " alter table return, meta:%p, len:%d, msg:%s", LOG_ID_VALUE, meta, metaLen, tstrerror(code));
taosArrayDestroy(pArray); taosArrayDestroy(pArray);
if (pVgData) taosMemoryFreeClear(pVgData->pData); if (pVgData) taosMemoryFreeClear(pVgData->pData);
@ -1608,7 +1608,7 @@ int taos_write_raw_block_with_fields_with_reqid(TAOS* taos, int rows, char* pDat
launchQueryImpl(pRequest, pQuery, true, NULL); launchQueryImpl(pRequest, pQuery, true, NULL);
code = pRequest->code; code = pRequest->code;
end: end:
uDebug(LOG_ID_TAG " write raw block with field return, msg:%s", LOG_ID_VALUE, tstrerror(code)); uDebug(LOG_ID_TAG " write raw block with field return, msg:%s", LOG_ID_VALUE, tstrerror(code));
taosMemoryFreeClear(pTableMeta); taosMemoryFreeClear(pTableMeta);
qDestroyQuery(pQuery); qDestroyQuery(pQuery);
@ -1668,7 +1668,7 @@ int taos_write_raw_block_with_reqid(TAOS* taos, int rows, char* pData, const cha
launchQueryImpl(pRequest, pQuery, true, NULL); launchQueryImpl(pRequest, pQuery, true, NULL);
code = pRequest->code; code = pRequest->code;
end: end:
uDebug(LOG_ID_TAG " write raw block return, msg:%s", LOG_ID_VALUE, tstrerror(code)); uDebug(LOG_ID_TAG " write raw block return, msg:%s", LOG_ID_VALUE, tstrerror(code));
taosMemoryFreeClear(pTableMeta); taosMemoryFreeClear(pTableMeta);
qDestroyQuery(pQuery); qDestroyQuery(pQuery);
@ -1708,7 +1708,8 @@ static int32_t buildCreateTbMap(SMqDataRsp* rsp, SHashObj* pHashObj) {
goto end; goto end;
} }
if (taosHashGet(pHashObj, pCreateReq.name, strlen(pCreateReq.name)) == NULL) { if (taosHashGet(pHashObj, pCreateReq.name, strlen(pCreateReq.name)) == NULL) {
RAW_RETURN_CHECK(taosHashPut(pHashObj, pCreateReq.name, strlen(pCreateReq.name), &pCreateReq, sizeof(SVCreateTbReq))); RAW_RETURN_CHECK(
taosHashPut(pHashObj, pCreateReq.name, strlen(pCreateReq.name), &pCreateReq, sizeof(SVCreateTbReq)));
} else { } else {
tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE); tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);
pCreateReq = (SVCreateTbReq){0}; pCreateReq = (SVCreateTbReq){0};
@ -1734,34 +1735,34 @@ static SHashObj* writeRawCache = NULL;
static int8_t initFlag = 0; static int8_t initFlag = 0;
static int8_t initedFlag = WRITE_RAW_INIT_START; static int8_t initedFlag = WRITE_RAW_INIT_START;
typedef struct{ typedef struct {
SHashObj* pVgHash; SHashObj* pVgHash;
SHashObj* pNameHash; SHashObj* pNameHash;
SHashObj* pMetaHash; SHashObj* pMetaHash;
}rawCacheInfo; } rawCacheInfo;
typedef struct{ typedef struct {
SVgroupInfo vgInfo; SVgroupInfo vgInfo;
int64_t uid; int64_t uid;
int64_t suid; int64_t suid;
}tbInfo; } tbInfo;
static void tmqFreeMeta(void *data){ static void tmqFreeMeta(void* data) {
STableMeta* pTableMeta = *(STableMeta**)data; STableMeta* pTableMeta = *(STableMeta**)data;
taosMemoryFree(pTableMeta); taosMemoryFree(pTableMeta);
} }
static void freeRawCache(void *data) { static void freeRawCache(void* data) {
rawCacheInfo* pRawCache = (rawCacheInfo*)data; rawCacheInfo* pRawCache = (rawCacheInfo*)data;
taosHashCleanup(pRawCache->pMetaHash); taosHashCleanup(pRawCache->pMetaHash);
taosHashCleanup(pRawCache->pNameHash); taosHashCleanup(pRawCache->pNameHash);
taosHashCleanup(pRawCache->pVgHash); taosHashCleanup(pRawCache->pVgHash);
} }
static int32_t initRawCacheHash(){ static int32_t initRawCacheHash() {
if (writeRawCache == NULL){ if (writeRawCache == NULL) {
writeRawCache = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK); writeRawCache = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
if (writeRawCache == NULL){ if (writeRawCache == NULL) {
return terrno; return terrno;
} }
taosHashSetFreeFp(writeRawCache, freeRawCache); taosHashSetFreeFp(writeRawCache, freeRawCache);
@ -1769,7 +1770,7 @@ static int32_t initRawCacheHash(){
return 0; return 0;
} }
static bool needRefreshMeta(void* rawData, STableMeta* pTableMeta, SSchemaWrapper* pSW){ static bool needRefreshMeta(void* rawData, STableMeta* pTableMeta, SSchemaWrapper* pSW) {
char* p = (char*)rawData; char* p = (char*)rawData;
// | version | total length | total rows | blankFill | total columns | flag seg| block group id | column schema | each // | version | total length | total rows | blankFill | total columns | flag seg| block group id | column schema | each
// column length | // column length |
@ -1799,16 +1800,15 @@ static bool needRefreshMeta(void* rawData, STableMeta* pTableMeta, SSchemaWrappe
} }
fields += sizeof(int8_t) + sizeof(int32_t); fields += sizeof(int8_t) + sizeof(int32_t);
if (j == pTableMeta->tableInfo.numOfColumns) if (j == pTableMeta->tableInfo.numOfColumns) return true;
return true;
} }
return false; return false;
} }
static int32_t getRawCache(SHashObj **pVgHash, SHashObj **pNameHash, SHashObj **pMetaHash, void *key) { static int32_t getRawCache(SHashObj** pVgHash, SHashObj** pNameHash, SHashObj** pMetaHash, void* key) {
int32_t code = 0; int32_t code = 0;
void* cacheInfo = taosHashGet(writeRawCache, &key, POINTER_BYTES); void* cacheInfo = taosHashGet(writeRawCache, &key, POINTER_BYTES);
if (cacheInfo == NULL){ if (cacheInfo == NULL) {
*pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK); *pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
RAW_NULL_CHECK(*pVgHash); RAW_NULL_CHECK(*pVgHash);
*pNameHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK); *pNameHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
@ -1819,7 +1819,7 @@ static int32_t getRawCache(SHashObj **pVgHash, SHashObj **pNameHash, SHashObj **
rawCacheInfo info = {*pVgHash, *pNameHash, *pMetaHash}; rawCacheInfo info = {*pVgHash, *pNameHash, *pMetaHash};
RAW_RETURN_CHECK(taosHashPut(writeRawCache, &key, POINTER_BYTES, &info, sizeof(rawCacheInfo))); RAW_RETURN_CHECK(taosHashPut(writeRawCache, &key, POINTER_BYTES, &info, sizeof(rawCacheInfo)));
} else { } else {
rawCacheInfo *info = (rawCacheInfo *)cacheInfo; rawCacheInfo* info = (rawCacheInfo*)cacheInfo;
*pVgHash = info->pVgHash; *pVgHash = info->pVgHash;
*pNameHash = info->pNameHash; *pNameHash = info->pNameHash;
*pMetaHash = info->pMetaHash; *pMetaHash = info->pMetaHash;
@ -1833,7 +1833,7 @@ end:
return code; return code;
} }
static int32_t buildRawRequest(TAOS* taos, SRequestObj** pRequest, SCatalog** pCatalog, SRequestConnInfo *conn){ static int32_t buildRawRequest(TAOS* taos, SRequestObj** pRequest, SCatalog** pCatalog, SRequestConnInfo* conn) {
int32_t code = 0; int32_t code = 0;
RAW_RETURN_CHECK(createRequest(*(int64_t*)taos, TSDB_SQL_INSERT, 0, pRequest)); RAW_RETURN_CHECK(createRequest(*(int64_t*)taos, TSDB_SQL_INSERT, 0, pRequest));
(*pRequest)->syncQuery = true; (*pRequest)->syncQuery = true;
@ -1852,8 +1852,9 @@ end:
return code; return code;
} }
typedef int32_t _raw_decode_func_(SDecoder *pDecoder, SMqDataRsp *pRsp); typedef int32_t _raw_decode_func_(SDecoder* pDecoder, SMqDataRsp* pRsp);
static int32_t decodeRawData(SDecoder *decoder, void* data, int32_t dataLen, _raw_decode_func_ func, SMqRspObj* rspObj){ static int32_t decodeRawData(SDecoder* decoder, void* data, int32_t dataLen, _raw_decode_func_ func,
SMqRspObj* rspObj) {
int8_t dataVersion = *(int8_t*)data; int8_t dataVersion = *(int8_t*)data;
if (dataVersion >= MQ_DATA_RSP_VERSION) { if (dataVersion >= MQ_DATA_RSP_VERSION) {
data = POINTER_SHIFT(data, sizeof(int8_t) + sizeof(int32_t)); data = POINTER_SHIFT(data, sizeof(int8_t) + sizeof(int32_t));
@ -1869,9 +1870,9 @@ static int32_t decodeRawData(SDecoder *decoder, void* data, int32_t dataLen, _ra
return code; return code;
} }
static int32_t processCacheMeta(SHashObj *pVgHash, SHashObj *pNameHash, SHashObj *pMetaHash, SVCreateTbReq* pCreateReqDst, static int32_t processCacheMeta(SHashObj* pVgHash, SHashObj* pNameHash, SHashObj* pMetaHash,
SCatalog* pCatalog, SRequestConnInfo* conn, SName* pName, SVCreateTbReq* pCreateReqDst, SCatalog* pCatalog, SRequestConnInfo* conn, SName* pName,
STableMeta** pMeta, SSchemaWrapper* pSW, void* rawData, int32_t retry){ STableMeta** pMeta, SSchemaWrapper* pSW, void* rawData, int32_t retry) {
int32_t code = 0; int32_t code = 0;
STableMeta* pTableMeta = NULL; STableMeta* pTableMeta = NULL;
tbInfo* tmpInfo = (tbInfo*)taosHashGet(pNameHash, pName->tname, strlen(pName->tname)); tbInfo* tmpInfo = (tbInfo*)taosHashGet(pNameHash, pName->tname, strlen(pName->tname));
@ -1884,13 +1885,13 @@ static int32_t processCacheMeta(SHashObj *pVgHash, SHashObj *pNameHash, SHashObj
} }
RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, conn, pName, &pTableMeta)); RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, conn, pName, &pTableMeta));
info.uid = pTableMeta->uid; info.uid = pTableMeta->uid;
if (pTableMeta->tableType == TSDB_CHILD_TABLE){ if (pTableMeta->tableType == TSDB_CHILD_TABLE) {
info.suid = pTableMeta->suid; info.suid = pTableMeta->suid;
} else { } else {
info.suid = pTableMeta->uid; info.suid = pTableMeta->uid;
} }
code = taosHashPut(pMetaHash, &info.suid, LONG_BYTES, &pTableMeta, POINTER_BYTES); code = taosHashPut(pMetaHash, &info.suid, LONG_BYTES, &pTableMeta, POINTER_BYTES);
if (code != 0){ if (code != 0) {
taosMemoryFree(pTableMeta); taosMemoryFree(pTableMeta);
goto end; goto end;
} }
@ -1902,20 +1903,21 @@ static int32_t processCacheMeta(SHashObj *pVgHash, SHashObj *pNameHash, SHashObj
RAW_RETURN_CHECK(taosHashPut(pNameHash, pName->tname, strlen(pName->tname), &info, sizeof(tbInfo))); RAW_RETURN_CHECK(taosHashPut(pNameHash, pName->tname, strlen(pName->tname), &info, sizeof(tbInfo)));
tmpInfo = (tbInfo*)taosHashGet(pNameHash, pName->tname, strlen(pName->tname)); tmpInfo = (tbInfo*)taosHashGet(pNameHash, pName->tname, strlen(pName->tname));
RAW_RETURN_CHECK(taosHashPut(pVgHash, &info.vgInfo.vgId, sizeof(info.vgInfo.vgId), &info.vgInfo, sizeof(SVgroupInfo))); RAW_RETURN_CHECK(
taosHashPut(pVgHash, &info.vgInfo.vgId, sizeof(info.vgInfo.vgId), &info.vgInfo, sizeof(SVgroupInfo)));
} }
if (pTableMeta == NULL || retry > 0){ if (pTableMeta == NULL || retry > 0) {
STableMeta** pTableMetaTmp = (STableMeta**)taosHashGet(pMetaHash, &tmpInfo->suid, LONG_BYTES); STableMeta** pTableMetaTmp = (STableMeta**)taosHashGet(pMetaHash, &tmpInfo->suid, LONG_BYTES);
if (pTableMetaTmp == NULL || retry > 0 || needRefreshMeta(rawData, *pTableMetaTmp, pSW)) { if (pTableMetaTmp == NULL || retry > 0 || needRefreshMeta(rawData, *pTableMetaTmp, pSW)) {
RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, conn, pName, &pTableMeta)); RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, conn, pName, &pTableMeta));
code = taosHashPut(pMetaHash, &tmpInfo->suid, LONG_BYTES, &pTableMeta, POINTER_BYTES); code = taosHashPut(pMetaHash, &tmpInfo->suid, LONG_BYTES, &pTableMeta, POINTER_BYTES);
if (code != 0){ if (code != 0) {
taosMemoryFree(pTableMeta); taosMemoryFree(pTableMeta);
goto end; goto end;
} }
}else{ } else {
pTableMeta = *pTableMetaTmp; pTableMeta = *pTableMetaTmp;
pTableMeta->uid = tmpInfo->uid; pTableMeta->uid = tmpInfo->uid;
pTableMeta->vgId = tmpInfo->vgInfo.vgId; pTableMeta->vgId = tmpInfo->vgInfo.vgId;
@ -1927,7 +1929,7 @@ end:
return code; return code;
} }
static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen){ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SQuery* pQuery = NULL; SQuery* pQuery = NULL;
SMqRspObj rspObj = {0}; SMqRspObj rspObj = {0};
@ -1940,12 +1942,12 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen){
uDebug(LOG_ID_TAG " write raw data, data:%p, dataLen:%d", LOG_ID_VALUE, data, dataLen); uDebug(LOG_ID_TAG " write raw data, data:%p, dataLen:%d", LOG_ID_VALUE, data, dataLen);
RAW_RETURN_CHECK(decodeRawData(&decoder, data, dataLen, tDecodeMqDataRsp, &rspObj)); RAW_RETURN_CHECK(decodeRawData(&decoder, data, dataLen, tDecodeMqDataRsp, &rspObj));
SHashObj *pVgHash = NULL; SHashObj* pVgHash = NULL;
SHashObj *pNameHash = NULL; SHashObj* pNameHash = NULL;
SHashObj *pMetaHash = NULL; SHashObj* pMetaHash = NULL;
RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos)); RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos));
int retry = 0; int retry = 0;
while(1){ while (1) {
RAW_RETURN_CHECK(smlInitHandle(&pQuery)); RAW_RETURN_CHECK(smlInitHandle(&pQuery));
uDebug(LOG_ID_TAG " write raw meta data block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum); uDebug(LOG_ID_TAG " write raw meta data block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum);
while (++rspObj.resIter < rspObj.dataRsp.blockNum) { while (++rspObj.resIter < rspObj.dataRsp.blockNum) {
@ -1968,8 +1970,8 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen){
tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN); tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);
STableMeta* pTableMeta = NULL; STableMeta* pTableMeta = NULL;
RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, NULL, pCatalog, &conn, RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, NULL, pCatalog, &conn, &pName, &pTableMeta, pSW,
&pName, &pTableMeta, pSW, rawData, retry)); rawData, retry));
char err[ERR_MSG_LEN] = {0}; char err[ERR_MSG_LEN] = {0};
code = rawBlockBindData(pQuery, pTableMeta, rawData, NULL, pSW, pSW->nCols, true, err, ERR_MSG_LEN, true); code = rawBlockBindData(pQuery, pTableMeta, rawData, NULL, pSW, pSW->nCols, true, err, ERR_MSG_LEN, true);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
@ -1991,7 +1993,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen){
break; break;
} }
end: end:
uDebug(LOG_ID_TAG " write raw data return, msg:%s", LOG_ID_VALUE, tstrerror(code)); uDebug(LOG_ID_TAG " write raw data return, msg:%s", LOG_ID_VALUE, tstrerror(code));
tDeleteMqDataRsp(&rspObj.dataRsp); tDeleteMqDataRsp(&rspObj.dataRsp);
tDecoderClear(&decoder); tDecoderClear(&decoder);
@ -2019,12 +2021,12 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
RAW_NULL_CHECK(pCreateTbHash); RAW_NULL_CHECK(pCreateTbHash);
RAW_RETURN_CHECK(buildCreateTbMap(&rspObj.dataRsp, pCreateTbHash)); RAW_RETURN_CHECK(buildCreateTbMap(&rspObj.dataRsp, pCreateTbHash));
SHashObj *pVgHash = NULL; SHashObj* pVgHash = NULL;
SHashObj *pNameHash = NULL; SHashObj* pNameHash = NULL;
SHashObj *pMetaHash = NULL; SHashObj* pMetaHash = NULL;
RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos)); RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos));
int retry = 0; int retry = 0;
while(1){ while (1) {
RAW_RETURN_CHECK(smlInitHandle(&pQuery)); RAW_RETURN_CHECK(smlInitHandle(&pQuery));
uDebug(LOG_ID_TAG " write raw meta data block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum); uDebug(LOG_ID_TAG " write raw meta data block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum);
while (++rspObj.resIter < rspObj.dataRsp.blockNum) { while (++rspObj.resIter < rspObj.dataRsp.blockNum) {
@ -2049,10 +2051,11 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
// find schema data info // find schema data info
SVCreateTbReq* pCreateReqDst = (SVCreateTbReq*)taosHashGet(pCreateTbHash, pName.tname, strlen(pName.tname)); SVCreateTbReq* pCreateReqDst = (SVCreateTbReq*)taosHashGet(pCreateTbHash, pName.tname, strlen(pName.tname));
STableMeta* pTableMeta = NULL; STableMeta* pTableMeta = NULL;
RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, pCreateReqDst, pCatalog, &conn, RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, pCreateReqDst, pCatalog, &conn, &pName,
&pName, &pTableMeta, pSW, rawData, retry)); &pTableMeta, pSW, rawData, retry));
char err[ERR_MSG_LEN] = {0}; char err[ERR_MSG_LEN] = {0};
code = rawBlockBindData(pQuery, pTableMeta, rawData, pCreateReqDst, pSW, pSW->nCols, true, err, ERR_MSG_LEN, true); code =
rawBlockBindData(pQuery, pTableMeta, rawData, pCreateReqDst, pSW, pSW->nCols, true, err, ERR_MSG_LEN, true);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
SET_ERROR_MSG("table:%s, err:%s", pName.tname, err); SET_ERROR_MSG("table:%s, err:%s", pName.tname, err);
goto end; goto end;
@ -2147,7 +2150,7 @@ static void processBatchMetaToJson(SMqBatchMetaRsp* pMsgRsp, char** string) {
*string = fullStr; *string = fullStr;
return; return;
end: end:
cJSON_Delete(pJson); cJSON_Delete(pJson);
tDeleteMqBatchMetaRsp(&rsp); tDeleteMqBatchMetaRsp(&rsp);
} }
@ -2170,7 +2173,7 @@ char* tmq_get_json_meta(TAOS_RES* res) {
processSimpleMeta(&rspObj->metaRsp, &pJson); processSimpleMeta(&rspObj->metaRsp, &pJson);
string = cJSON_PrintUnformatted(pJson); string = cJSON_PrintUnformatted(pJson);
cJSON_Delete(pJson); cJSON_Delete(pJson);
} else{ } else {
uError("tmq_get_json_meta res:%d, invalid type", *(int8_t*)res); uError("tmq_get_json_meta res:%d, invalid type", *(int8_t*)res);
} }
@ -2230,7 +2233,7 @@ static int32_t encodeMqDataRsp(__encode_func__* encodeFunc, SMqDataRsp* rspObj,
raw->raw = buf; raw->raw = buf;
raw->raw_len = len; raw->raw_len = len;
return code; return code;
FAILED: FAILED:
tEncoderClear(&encoder); tEncoderClear(&encoder);
taosMemoryFree(buf); taosMemoryFree(buf);
return code; return code;
@ -2282,7 +2285,7 @@ void tmq_free_raw(tmq_raw_data raw) {
(void)memset(terrMsg, 0, ERR_MSG_LEN); (void)memset(terrMsg, 0, ERR_MSG_LEN);
} }
static int32_t writeRawInit(){ static int32_t writeRawInit() {
while (atomic_load_8(&initedFlag) == WRITE_RAW_INIT_START) { while (atomic_load_8(&initedFlag) == WRITE_RAW_INIT_START) {
int8_t old = atomic_val_compare_exchange_8(&initFlag, 0, 1); int8_t old = atomic_val_compare_exchange_8(&initFlag, 0, 1);
if (old == 0) { if (old == 0) {
@ -2296,7 +2299,7 @@ static int32_t writeRawInit(){
} }
} }
if (atomic_load_8(&initedFlag) == WRITE_RAW_INIT_FAIL){ if (atomic_load_8(&initedFlag) == WRITE_RAW_INIT_FAIL) {
return TSDB_CODE_INTERNAL_ERROR; return TSDB_CODE_INTERNAL_ERROR;
} }
return 0; return 0;
@ -2321,7 +2324,7 @@ static int32_t writeRawImpl(TAOS* taos, void* buf, uint32_t len, uint16_t type)
return taosDropTable(taos, buf, len); return taosDropTable(taos, buf, len);
} else if (type == TDMT_VND_DELETE) { } else if (type == TDMT_VND_DELETE) {
return taosDeleteData(taos, buf, len); return taosDeleteData(taos, buf, len);
} else if (type == RES_TYPE__TMQ_METADATA){ } else if (type == RES_TYPE__TMQ_METADATA) {
return tmqWriteRawMetaDataImpl(taos, buf, len); return tmqWriteRawMetaDataImpl(taos, buf, len);
} else if (type == RES_TYPE__TMQ) { } else if (type == RES_TYPE__TMQ) {
return tmqWriteRawDataImpl(taos, buf, len); return tmqWriteRawDataImpl(taos, buf, len);
@ -2374,7 +2377,7 @@ static int32_t tmqWriteBatchMetaDataImpl(TAOS* taos, void* meta, int32_t metaLen
} }
} }
end: end:
tDeleteMqBatchMetaRsp(&rsp); tDeleteMqBatchMetaRsp(&rsp);
return code; return code;
} }