Elgato_dark/TDengine
1
0
Fork 0
mirror of https://github.com/taosdata/TDengine synced 2026-05-24 10:09:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4
TDengine/source/libs/executor/src/dataInserter.c
xinsheng Ren a50ffcd58d
Merge pull request #32073 from taosdata/fix/TD-36848/setNullType 
fix: set null type
2025-07-21 16:03:27 +08:00

2488 lines
84 KiB
C
Raw Blame History

/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include "dataSinkInt.h"
#include "dataSinkMgt.h"
#include "executor.h"
#include "executorInt.h"
#include "functionMgt.h"
#include "libs/new-stream/stream.h"
#include "osAtomic.h"
#include "osMemory.h"
#include "osSemaphore.h"
#include "planner.h"
#include "query.h"
#include "querytask.h"
#include "storageapi.h"
#include "taoserror.h"
#include "tarray.h"
#include "tcompression.h"
#include "tdatablock.h"
#include "tdataformat.h"
#include "tglobal.h"
#include "thash.h"
#include "tmsg.h"
#include "tqueue.h"
extern SDataSinkStat gDataSinkStat;
SHashObj* gStreamGrpTableHash = NULL;
typedef struct SSubmitRes {
int64_t affectedRows;
int32_t code;
SSubmitRsp2* pRsp;
} SSubmitRes;
typedef struct SSubmitTbDataMsg {
int32_t vgId;
int32_t len;
void* pData;
} SSubmitTbDataMsg;
static void destroySSubmitTbDataMsg(void* p) {
if (p == NULL) return;
SSubmitTbDataMsg* pVg = p;
taosMemoryFree(pVg->pData);
taosMemoryFree(pVg);
}
typedef struct SDataInserterHandle {
SDataSinkHandle sink;
SDataSinkManager* pManager;
STSchema* pSchema;
SQueryInserterNode* pNode;
SSubmitRes submitRes;
SInserterParam* pParam;
SArray* pDataBlocks;
SHashObj* pCols;
int32_t status;
bool queryEnd;
bool fullOrderColList;
uint64_t useconds;
uint64_t cachedSize;
uint64_t flags;
TdThreadMutex mutex;
tsem_t ready;
bool explain;
bool isStbInserter;
SSchemaWrapper* pTagSchema;
const char* dbFName;
SHashObj* dbVgInfoMap;
SUseDbRsp* pRsp;
} SDataInserterHandle;
typedef struct SSubmitRspParam {
SDataInserterHandle* pInserter;
void* putParam;
} SSubmitRspParam;
typedef struct SInsertTableRes {
int64_t uid;
int64_t vgid;
int32_t version;
char* tbname;
} SInsertTableRes;
static void freeCacheTbInfo(void* p) {
SInsertTableRes* pTbRes = (SInsertTableRes*)p;
if (pTbRes->tbname) {
taosMemFree(pTbRes->tbname);
pTbRes->tbname = NULL;
}
}
int32_t initInserterGrpInfo() {
gStreamGrpTableHash = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
if (NULL == gStreamGrpTableHash) {
qError("failed to create stream group table hash");
return terrno;
}
taosHashSetFreeFp(gStreamGrpTableHash, freeCacheTbInfo);
return TSDB_CODE_SUCCESS;
}
void destroyInserterGrpInfo() {
static int8_t destoryGrpInfo = 0;
int8_t flag = atomic_val_compare_exchange_8(&destoryGrpInfo, 0, 1);
if (flag != 0) {
return;
}
if (NULL != gStreamGrpTableHash) {
taosHashCleanup(gStreamGrpTableHash);
gStreamGrpTableHash = NULL;
}
}
static int32_t checkResAndGetTableId(const SSubmitRes* pSubmitRes, int8_t tbType, SInsertTableRes* res) {
int32_t code = TSDB_CODE_SUCCESS;
if (!pSubmitRes->pRsp) {
stError("create table response is NULL");
return TSDB_CODE_MND_STREAM_INTERNAL_ERROR;
}
if (pSubmitRes->pRsp->aCreateTbRsp->size != 1) {
stError("create table response size is not 1");
return TSDB_CODE_MND_STREAM_INTERNAL_ERROR;
}
SVCreateTbRsp* pCreateTbRsp = taosArrayGet(pSubmitRes->pRsp->aCreateTbRsp, 0);
if (pCreateTbRsp->code != 0 && pCreateTbRsp->code != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
stError("create table failed, code:%d", pCreateTbRsp->code);
return pCreateTbRsp->code;
}
if (!pCreateTbRsp->pMeta || pCreateTbRsp->pMeta->tuid == 0) {
stError("create table can not get tuid");
return TSDB_CODE_MND_STREAM_INTERNAL_ERROR;
}
res->uid = pCreateTbRsp->pMeta->tuid;
res->version = pCreateTbRsp->pMeta->sversion;
res->vgid = pCreateTbRsp->pMeta->vgId;
stDebug("inserter callback, uid:%" PRId64 " vgid: %" PRId64 ", version: %d", res->uid, res->vgid, res->version);
return TSDB_CODE_SUCCESS;
}
static int32_t saveCreateGrpTableInfo(SStreamDataInserterInfo* pInserterInfo, const SSubmitRes* pSubmitRes,
int8_t tbType) {
int32_t code = TSDB_CODE_SUCCESS;
int64_t key[2] = {pInserterInfo->streamId, pInserterInfo->groupId};
SInsertTableRes* pTbRes = taosHashGet(gStreamGrpTableHash, key, sizeof(key));
if (NULL == pTbRes) {
return TSDB_CODE_MND_STREAM_INTERNAL_ERROR;
}
code = checkResAndGetTableId(pSubmitRes, tbType, pTbRes);
if (code) {
return code;
}
return TSDB_CODE_SUCCESS;
}
static int32_t initTableInfo(SStreamDataInserterInfo* pInserterInfo) {
int32_t code = TSDB_CODE_SUCCESS;
SInsertTableRes res = {0};
res.tbname = taosStrdup(pInserterInfo->tbName);
if (res.tbname == NULL) {
stError("failed to allocate memory for table name");
return terrno;
}
int64_t key[2] = {pInserterInfo->streamId, pInserterInfo->groupId};
code = taosHashPut(gStreamGrpTableHash, key, sizeof(key), &res, sizeof(SInsertTableRes));
if(code == TSDB_CODE_DUP_KEY) {
taosMemFree(res.tbname);
return TSDB_CODE_SUCCESS;
}
if (code != 0 && code != TSDB_CODE_DUP_KEY) {
return code;
}
return TSDB_CODE_SUCCESS;
}
static bool colsIsSupported(const STableMetaRsp* pTableMetaRsp, const SStreamInserterParam* pInserterParam) {
SArray* pCreatingFields = pInserterParam->pFields;
if (pTableMetaRsp->numOfColumns < pCreatingFields->size) {
return false;
}
for (int32_t i = 0; i < pCreatingFields->size; ++i) {
SFieldWithOptions* pField = taosArrayGet(pCreatingFields, i);
if (NULL == pField) {
stError("isSupportedSTableSchema: failed to get field from array");
return false;
}
if (strncmp(pTableMetaRsp->pSchemas[i].name, pField->name, TSDB_COL_NAME_LEN) != 0) {
return false;
}
if (pTableMetaRsp->pSchemas[i].type != pField->type || pTableMetaRsp->pSchemas[i].bytes != pField->bytes) {
return false;
}
}
return true;
}
static bool TagsIsSupported(const STableMetaRsp* pTableMetaRsp, const SStreamInserterParam* pInserterParam) {
SArray* pCreatingTags = pInserterParam->pTagFields;
if (pTableMetaRsp->numOfTags < pCreatingTags->size) {
return false;
}
int32_t tagIndexOffset = -1;
SFieldWithOptions* pField = taosArrayGet(pCreatingTags, 0);
if (NULL == pField) {
stError("isSupportedSTableSchema: failed to get field from array");
return false;
}
for (int32_t i = 0; i < pTableMetaRsp->numOfColumns + pTableMetaRsp->numOfTags; ++i) {
if (strncmp(pTableMetaRsp->pSchemas[i].name, pField->name, TSDB_COL_NAME_LEN) != 0) {
tagIndexOffset = i;
break;
}
}
if (tagIndexOffset == -1) {
stError("isSupportedSTableSchema: failed to get tag index");
return false;
}
for (int32_t i = 0; i < pTableMetaRsp->numOfTags; ++i) {
int32_t index = i + tagIndexOffset;
SFieldWithOptions* pField = taosArrayGet(pCreatingTags, i);
if (NULL == pField) {
stError("isSupportedSTableSchema: failed to get field from array");
return false;
}
if (strncmp(pTableMetaRsp->pSchemas[index].name, pField->name, TSDB_COL_NAME_LEN) != 0) {
return false;
}
if (pTableMetaRsp->pSchemas[index].type != pField->type || pTableMetaRsp->pSchemas[index].bytes != pField->bytes) {
return false;
}
}
return true;
}
static bool isSupportedSTableSchema(const STableMetaRsp* pTableMetaRsp, const SStreamInserterParam* pInserterParam) {
if (!colsIsSupported(pTableMetaRsp, pInserterParam)) {
return false;
}
if (!TagsIsSupported(pTableMetaRsp, pInserterParam)) {
return false;
}
return true;
}
static bool isSupportedNTableSchema(const STableMetaRsp* pTableMetaRsp, const SStreamInserterParam* pInserterParam) {
return colsIsSupported(pTableMetaRsp, pInserterParam);
}
static int32_t checkAndSaveCreateGrpTableInfo(SDataInserterHandle* pInserthandle,
SStreamDataInserterInfo* pInserterInfo) {
int32_t code = TSDB_CODE_SUCCESS;
SSubmitRes* pSubmitRes = &pInserthandle->submitRes;
int8_t tbType = pInserthandle->pParam->streamInserterParam->tbType;
SVCreateTbRsp* pCreateTbRsp = taosArrayGet(pSubmitRes->pRsp->aCreateTbRsp, 0);
SSchema* pExistRow = pCreateTbRsp->pMeta->pSchemas;
SStreamInserterParam* pInserterParam = pInserthandle->pParam->streamInserterParam;
if (tbType == TSDB_CHILD_TABLE || tbType == TSDB_SUPER_TABLE) {
if (!isSupportedSTableSchema(pCreateTbRsp->pMeta, pInserterParam)) {
stError("create table failed, schema is not supported");
return TSDB_CODE_STREAM_INSERT_SCHEMA_NOT_MATCH;
}
} else if (tbType == TSDB_NORMAL_TABLE) {
if (!isSupportedNTableSchema(pCreateTbRsp->pMeta, pInserterParam)) {
stError("create table failed, schema is not supported");
return TSDB_CODE_STREAM_INSERT_SCHEMA_NOT_MATCH;
}
} else {
stError("checkAndSaveCreateGrpTableInfo failed, tbType:%d is not supported", tbType);
return TSDB_CODE_MND_STREAM_INTERNAL_ERROR;
}
int64_t key[2] = {pInserterInfo->streamId, pInserterInfo->groupId};
SInsertTableRes* pTbRes = taosHashGet(gStreamGrpTableHash, key, sizeof(key));
if (NULL == pTbRes) {
return TSDB_CODE_MND_STREAM_INTERNAL_ERROR;
}
SInsertTableRes res = {0};
code = checkResAndGetTableId(pSubmitRes, tbType, pTbRes);
if (code) {
return code;
}
return TSDB_CODE_SUCCESS;
}
int32_t inserterCallback(void* param, SDataBuf* pMsg, int32_t code) {
SSubmitRspParam* pParam = (SSubmitRspParam*)param;
SDataInserterHandle* pInserter = pParam->pInserter;
int32_t code2 = 0;
if (code) {
pInserter->submitRes.code = code;
} else {
pInserter->submitRes.code = TSDB_CODE_SUCCESS;
}
SDecoder coder = {0};
if (code == TSDB_CODE_SUCCESS) {
pInserter->submitRes.pRsp = taosMemoryCalloc(1, sizeof(SSubmitRsp2));
if (NULL == pInserter->submitRes.pRsp) {
pInserter->submitRes.code = terrno;
goto _return;
}
tDecoderInit(&coder, pMsg->pData, pMsg->len);
code = tDecodeSSubmitRsp2(&coder, pInserter->submitRes.pRsp);
if (code) {
tDestroySSubmitRsp2(pInserter->submitRes.pRsp, TSDB_MSG_FLG_DECODE);
taosMemoryFree(pInserter->submitRes.pRsp);
pInserter->submitRes.code = code;
goto _return;
}
if (pInserter->submitRes.pRsp->affectedRows > 0) {
SArray* pCreateTbList = pInserter->submitRes.pRsp->aCreateTbRsp;
int32_t numOfTables = taosArrayGetSize(pCreateTbList);
for (int32_t i = 0; i < numOfTables; ++i) {
SVCreateTbRsp* pRsp = taosArrayGet(pCreateTbList, i);
if (NULL == pRsp) {
pInserter->submitRes.code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _return;
}
if (TSDB_CODE_SUCCESS != pRsp->code) {
code = pRsp->code;
tDestroySSubmitRsp2(pInserter->submitRes.pRsp, TSDB_MSG_FLG_DECODE);
taosMemoryFree(pInserter->submitRes.pRsp);
pInserter->submitRes.code = code;
goto _return;
}
}
}
if (pParam->putParam != NULL && ((SStreamDataInserterInfo*)pParam->putParam)->isAutoCreateTable) {
saveCreateGrpTableInfo((SStreamDataInserterInfo*)pParam->putParam, &pInserter->submitRes,
pInserter->pParam->streamInserterParam->tbType);
}
pInserter->submitRes.affectedRows += pInserter->submitRes.pRsp->affectedRows;
qDebug("submit rsp received, affectedRows:%d, total:%" PRId64, pInserter->submitRes.pRsp->affectedRows,
pInserter->submitRes.affectedRows);
tDestroySSubmitRsp2(pInserter->submitRes.pRsp, TSDB_MSG_FLG_DECODE);
taosMemoryFree(pInserter->submitRes.pRsp);
}
if (TSDB_CODE_TDB_TABLE_ALREADY_EXIST == code) {
pInserter->submitRes.code = code;
if (pParam->putParam != NULL && ((SStreamDataInserterInfo*)pParam->putParam)->isAutoCreateTable) {
pInserter->submitRes.pRsp = taosMemoryCalloc(1, sizeof(SSubmitRsp2));
if (NULL == pInserter->submitRes.pRsp) {
pInserter->submitRes.code = terrno;
goto _return;
}
tDecoderInit(&coder, pMsg->pData, pMsg->len);
code2 = tDecodeSSubmitRsp2(&coder, pInserter->submitRes.pRsp);
if (code2 == TSDB_CODE_SUCCESS) {
code2 = checkAndSaveCreateGrpTableInfo(pInserter, (SStreamDataInserterInfo*)pParam->putParam);
}
tDestroySSubmitRsp2(pInserter->submitRes.pRsp, TSDB_MSG_FLG_DECODE);
taosMemoryFree(pInserter->submitRes.pRsp);
if (code2) {
pInserter->submitRes.code = code2;
goto _return;
}
}
}
_return:
tDecoderClear(&coder);
code2 = tsem_post(&pInserter->ready);
if (code2 < 0) {
qError("tsem_post inserter ready failed, error:%s", tstrerror(code2));
if (TSDB_CODE_SUCCESS == code) {
pInserter->submitRes.code = code2;
}
}
taosMemoryFree(pMsg->pData);
return TSDB_CODE_SUCCESS;
}
void freeUseDbOutput_tmp(void* ppOutput) {
SUseDbOutput* pOut = *(SUseDbOutput**)ppOutput;
if (NULL == ppOutput) {
return;
}
if (pOut->dbVgroup) {
freeVgInfo(pOut->dbVgroup);
}
taosMemFree(pOut);
*(SUseDbOutput**)ppOutput = NULL;
}
static int32_t processUseDbRspForInserter(void* param, SDataBuf* pMsg, int32_t code) {
int32_t lino = 0;
SDBVgInfoReq* pVgInfoReq = (SDBVgInfoReq*)param;
if (TSDB_CODE_SUCCESS != code) {
// pInserter->pTaskInfo->code = rpcCvtErrCode(code);
// if (pInserter->pTaskInfo->code != code) {
// qError("load db info rsp received, error:%s, cvted error:%s", tstrerror(code),
// tstrerror(pInserter->pTaskInfo->code));
// } else {
// qError("load db info rsp received, error:%s", tstrerror(code));
// }
goto _return;
}
pVgInfoReq->pRsp = taosMemoryMalloc(sizeof(SUseDbRsp));
QUERY_CHECK_NULL(pVgInfoReq->pRsp, code, lino, _return, terrno);
code = tDeserializeSUseDbRsp(pMsg->pData, (int32_t)pMsg->len, pVgInfoReq->pRsp);
QUERY_CHECK_CODE(code, lino, _return);
_return:
taosMemoryFreeClear(pMsg->pData);
taosMemoryFreeClear(pMsg->pEpSet);
if (code != 0){
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
int ret = tsem_post(&pVgInfoReq->ready);
if (ret != 0) {
qError("%s failed code: %d", __func__, ret);
}
return code;
}
int inserterVgInfoComp(const void* lp, const void* rp) {
SVgroupInfo* pLeft = (SVgroupInfo*)lp;
SVgroupInfo* pRight = (SVgroupInfo*)rp;
if (pLeft->hashBegin < pRight->hashBegin) {
return -1;
} else if (pLeft->hashBegin > pRight->hashBegin) {
return 1;
}
return 0;
}
static int32_t buildDbVgInfoMap(void* clientRpc, const char* dbFName, SUseDbOutput* output) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
char* buf1 = NULL;
SUseDbReq* pReq = NULL;
SDBVgInfoReq dbVgInfoReq = {0};
code = tsem_init(&dbVgInfoReq.ready, 0, 0);
if (code != TSDB_CODE_SUCCESS) {
qError("tsem_init failed, error:%s", tstrerror(code));
return code;
}
pReq = taosMemoryMalloc(sizeof(SUseDbReq));
QUERY_CHECK_NULL(pReq, code, lino, _return, terrno);
tstrncpy(pReq->db, dbFName, TSDB_DB_FNAME_LEN);
QUERY_CHECK_CODE(code, lino, _return);
int32_t contLen = tSerializeSUseDbReq(NULL, 0, pReq);
buf1 = taosMemoryCalloc(1, contLen);
QUERY_CHECK_NULL(buf1, code, lino, _return, terrno);
int32_t tempRes = tSerializeSUseDbReq(buf1, contLen, pReq);
if (tempRes < 0) {
QUERY_CHECK_CODE(terrno, lino, _return);
}
SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
QUERY_CHECK_NULL(pMsgSendInfo, code, lino, _return, terrno);
SEpSet pEpSet = {0};
QUERY_CHECK_CODE(getCurrentMnodeEpset(&pEpSet), lino, _return);
pMsgSendInfo->param = &dbVgInfoReq;
pMsgSendInfo->msgInfo.pData = buf1;
buf1 = NULL;
pMsgSendInfo->msgInfo.len = contLen;
pMsgSendInfo->msgType = TDMT_MND_GET_DB_INFO;
pMsgSendInfo->fp = processUseDbRspForInserter;
// pMsgSendInfo->requestId = pTaskInfo->id.queryId;
code = asyncSendMsgToServer(clientRpc, &pEpSet, NULL, pMsgSendInfo);
QUERY_CHECK_CODE(code, lino, _return);
code = tsem_wait(&dbVgInfoReq.ready);
QUERY_CHECK_CODE(code, lino, _return);
code = queryBuildUseDbOutput(output, dbVgInfoReq.pRsp);
QUERY_CHECK_CODE(code, lino, _return);
output->dbVgroup->vgArray = taosArrayInit(dbVgInfoReq.pRsp->vgNum, sizeof(SVgroupInfo));
if (NULL == output->dbVgroup->vgArray) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _return);
}
void* pIter = taosHashIterate(output->dbVgroup->vgHash, NULL);
while (pIter) {
if (NULL == taosArrayPush(output->dbVgroup->vgArray, pIter)) {
taosHashCancelIterate(output->dbVgroup->vgHash, pIter);
return terrno;
}
pIter = taosHashIterate(output->dbVgroup->vgHash, pIter);
}
taosArraySort(output->dbVgroup->vgArray, inserterVgInfoComp);
_return:
if (code) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
taosMemoryFree(buf1);
}
taosMemoryFree(pReq);
tsem_destroy(&dbVgInfoReq.ready);
if (dbVgInfoReq.pRsp) {
tFreeSUsedbRsp(dbVgInfoReq.pRsp);
taosMemoryFreeClear(dbVgInfoReq.pRsp);
}
return code;
}
int32_t inserterBuildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTag, int64_t suid, const char* sname,
SArray* tagName, uint8_t tagNum, int32_t ttl) {
pTbReq->type = TD_CHILD_TABLE;
pTbReq->ctb.pTag = (uint8_t*)pTag;
pTbReq->name = taosStrdup(tname);
if (!pTbReq->name) return terrno;
pTbReq->ctb.suid = suid;
pTbReq->ctb.tagNum = tagNum;
if (sname) {
pTbReq->ctb.stbName = taosStrdup(sname);
if (!pTbReq->ctb.stbName) {
taosMemoryFree(pTbReq->name);
return terrno;
}
}
pTbReq->ctb.tagName = taosArrayDup(tagName, NULL);
if (!pTbReq->ctb.tagName) return terrno;
pTbReq->ttl = ttl;
pTbReq->commentLen = -1;
return TSDB_CODE_SUCCESS;
}
int32_t inserterHashValueComp(void const* lp, void const* rp) {
uint32_t* key = (uint32_t*)lp;
SVgroupInfo* pVg = (SVgroupInfo*)rp;
if (*key < pVg->hashBegin) {
return -1;
} else if (*key > pVg->hashEnd) {
return 1;
}
return 0;
}
int32_t inserterGetVgInfo(SDBVgInfo* dbInfo, char* tbName, SVgroupInfo* pVgInfo) {
if (NULL == dbInfo) {
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
if (NULL == dbInfo->vgArray) {
qError("empty db vgArray, hashSize:%d", taosHashGetSize(dbInfo->vgHash));
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
uint32_t hashValue =
taosGetTbHashVal(tbName, (int32_t)strlen(tbName), dbInfo->hashMethod, dbInfo->hashPrefix, dbInfo->hashSuffix);
SVgroupInfo* vgInfo = taosArraySearch(dbInfo->vgArray, &hashValue, inserterHashValueComp, TD_EQ);
if (NULL == vgInfo) {
qError("no hash range found for hash value [%u], table:%s, numOfVgId:%d", hashValue, tbName,
(int32_t)taosArrayGetSize(dbInfo->vgArray));
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
*pVgInfo = *vgInfo;
qDebug("insert get vgInfo, tbName:%s vgId:%d epset(%s:%d)", tbName, pVgInfo->vgId, pVgInfo->epSet.eps[0].fqdn,
pVgInfo->epSet.eps[0].port);
return TSDB_CODE_SUCCESS;
}
int32_t inserterGetVgId(SDBVgInfo* dbInfo, char* tbName, int32_t* vgId) {
SVgroupInfo vgInfo = {0};
int32_t code = inserterGetVgInfo(dbInfo, tbName, &vgInfo);
if (code != TSDB_CODE_SUCCESS) {
qError("inserterGetVgId failed, code:%d", code);
return code;
}
*vgId = vgInfo.vgId;
return TSDB_CODE_SUCCESS;
}
int32_t inserterGetDbVgInfo(SDataInserterHandle* pInserter, const char* dbFName, SDBVgInfo** dbVgInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t line = 0;
SUseDbOutput* output = NULL;
// QRY_PARAM_CHECK(dbVgInfo);
// QRY_PARAM_CHECK(pInserter);
// QRY_PARAM_CHECK(name);
if (pInserter->dbVgInfoMap == NULL) {
pInserter->dbVgInfoMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if (pInserter->dbVgInfoMap == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
SUseDbOutput** find = (SUseDbOutput**)taosHashGet(pInserter->dbVgInfoMap, dbFName, strlen(dbFName));
if (find == NULL) {
output = taosMemoryMalloc(sizeof(SUseDbOutput));
if (output == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
code = buildDbVgInfoMap(pInserter->pParam->readHandle->pMsgCb->clientRpc, dbFName, output);
QUERY_CHECK_CODE(code, line, _return);
code = taosHashPut(pInserter->dbVgInfoMap, dbFName, strlen(dbFName), &output, POINTER_BYTES);
QUERY_CHECK_CODE(code, line, _return);
} else {
output = *find;
}
*dbVgInfo = output->dbVgroup;
return code;
_return:
qError("%s failed at line %d since %s", __func__, line, tstrerror(code));
freeUseDbOutput_tmp(&output);
return code;
}
int32_t getExistVgInfo(SDataInserterHandle* pInserter, SStreamInserterParam* pInsertParam,
SStreamDataInserterInfo* pInserterInfo, SVgroupInfo* pVgInfo) {
return getDbVgInfoForExec(pInserter->pParam->readHandle->pMsgCb->clientRpc, pInsertParam->dbFName,
pInserterInfo->tbName, pVgInfo);
}
static int32_t sendSubmitRequest(SDataInserterHandle* pInserter, void* putParam, void* pMsg, int32_t msgLen,
void* pTransporter, SEpSet* pEpset) {
// send the fetch remote task result reques
SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
if (NULL == pMsgSendInfo) {
taosMemoryFreeClear(pMsg);
return terrno;
}
SSubmitRspParam* pParam = taosMemoryCalloc(1, sizeof(SSubmitRspParam));
if (NULL == pParam) {
taosMemoryFreeClear(pMsg);
taosMemoryFreeClear(pMsgSendInfo);
return terrno;
}
pParam->pInserter = pInserter;
pParam->putParam = putParam;
pMsgSendInfo->param = pParam;
pMsgSendInfo->paramFreeFp = taosAutoMemoryFree;
pMsgSendInfo->msgInfo.pData = pMsg;
pMsgSendInfo->msgInfo.len = msgLen;
pMsgSendInfo->msgType = TDMT_VND_SUBMIT;
pMsgSendInfo->fp = inserterCallback;
return asyncSendMsgToServer(pTransporter, pEpset, NULL, pMsgSendInfo);
}
static int32_t submitReqToMsg(int32_t vgId, SSubmitReq2* pReq, void** pData, int32_t* pLen) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t len = 0;
void* pBuf = NULL;
tEncodeSize(tEncodeSubmitReq, pReq, len, code);
if (TSDB_CODE_SUCCESS == code) {
SEncoder encoder;
len += sizeof(SSubmitReq2Msg);
pBuf = taosMemoryMalloc(len);
if (NULL == pBuf) {
return terrno;
}
((SSubmitReq2Msg*)pBuf)->header.vgId = htonl(vgId);
((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len);
((SSubmitReq2Msg*)pBuf)->version = htobe64(1);
tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));
code = tEncodeSubmitReq(&encoder, pReq);
tEncoderClear(&encoder);
}
if (TSDB_CODE_SUCCESS == code) {
*pData = pBuf;
*pLen = len;
} else {
taosMemoryFree(pBuf);
}
return code;
}
int32_t buildSubmitReqFromStbBlock(SDataInserterHandle* pInserter, SHashObj* pHash, const SSDataBlock* pDataBlock,
const STSchema* pTSchema, int64_t uid, int32_t vgId, tb_uid_t suid) {
SArray* pVals = NULL;
SArray* pTagVals = NULL;
int32_t numOfBlks = 0;
terrno = TSDB_CODE_SUCCESS;
int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
int32_t rows = pDataBlock->info.rows;
if (!pTagVals && !(pTagVals = taosArrayInit(colNum, sizeof(STagVal)))) {
goto _end;
}
if (!pVals && !(pVals = taosArrayInit(colNum, sizeof(SColVal)))) {
goto _end;
}
SDBVgInfo* dbInfo = NULL;
int32_t code = inserterGetDbVgInfo(pInserter, pInserter->dbFName, &dbInfo);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
goto _end;
}
for (int32_t j = 0; j < rows; ++j) {
SSubmitTbData tbData = {0};
if (!(tbData.aRowP = taosArrayInit(rows, sizeof(SRow*)))) {
goto _end;
}
tbData.suid = suid;
tbData.uid = uid;
tbData.sver = pTSchema->version;
int64_t lastTs = TSKEY_MIN;
taosArrayClear(pVals);
int32_t offset = 0;
taosArrayClear(pTagVals);
tbData.uid = 0;
tbData.pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
if (NULL == tbData.pCreateTbReq) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
tbData.flags |= SUBMIT_REQ_AUTO_CREATE_TABLE;
SColumnInfoData* tbname = taosArrayGet(pDataBlock->pDataBlock, 0);
if (NULL == tbname) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
qError("Insert into stable must have tbname column");
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
if (tbname->info.type != TSDB_DATA_TYPE_BINARY) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
qError("tbname column must be binary");
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
if (colDataIsNull_s(tbname, j)) {
qError("insert into stable tbname column is null");
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
void* data = colDataGetVarData(tbname, j);
SValue sv = (SValue){TSDB_DATA_TYPE_VARCHAR, .nData = varDataLen(data), .pData = varDataVal(data)};
SColVal cv = COL_VAL_VALUE(0, sv);
char tbFullName[TSDB_TABLE_FNAME_LEN];
char tableName[TSDB_TABLE_FNAME_LEN];
memcpy(tableName, sv.pData, sv.nData);
tableName[sv.nData] = '\0';
int32_t len = snprintf(tbFullName, TSDB_TABLE_FNAME_LEN, "%s.%s", pInserter->dbFName, tableName);
if (len >= TSDB_TABLE_FNAME_LEN) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
qError("table name too long after format, len:%d, maxLen:%d", len, TSDB_TABLE_FNAME_LEN);
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
int32_t vgIdForTbName = 0;
code = inserterGetVgId(dbInfo, tbFullName, &vgIdForTbName);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
SSubmitReq2* pReq = taosHashGet(pHash, &vgIdForTbName, sizeof(int32_t));
if (pReq == NULL) {
pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2));
if (NULL == pReq) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
taosHashPut(pHash, &vgIdForTbName, sizeof(int32_t), pReq, sizeof(SSubmitReq2));
}
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
SArray* TagNames = taosArrayInit(8, TSDB_COL_NAME_LEN);
if (!TagNames) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
for (int32_t i = 0; i < pInserter->pTagSchema->nCols; ++i) {
SSchema* tSchema = &pInserter->pTagSchema->pSchema[i];
int16_t colIdx = tSchema->colId;
int16_t* slotId = taosHashGet(pInserter->pCols, &colIdx, sizeof(colIdx));
if (NULL == slotId) {
continue;
}
if (NULL == taosArrayPush(TagNames, tSchema->name)) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
colIdx = *slotId;
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, colIdx);
if (NULL == pColInfoData) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
// void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_VARCHAR: {
if (pColInfoData->info.type != tSchema->type) {
qError("tag:%d type:%d in block dismatch with schema tag:%d type:%d", colIdx, pColInfoData->info.type, i,
tSchema->type);
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
if (colDataIsNull_s(pColInfoData, j)) {
continue;
} else {
void* data = colDataGetVarData(pColInfoData, j);
STagVal tv = (STagVal){
.cid = tSchema->colId, .type = tSchema->type, .nData = varDataLen(data), .pData = varDataVal(data)};
if (NULL == taosArrayPush(pTagVals, &tv)) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
}
break;
}
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_MEDIUMBLOB:
qError("the tag type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
terrno = TSDB_CODE_APP_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
break;
default:
if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
if (colDataIsNull_s(pColInfoData, j)) {
continue;
} else {
void* data = colDataGetData(pColInfoData, j);
STagVal tv = {.cid = tSchema->colId, .type = tSchema->type};
memcpy(&tv.i64, data, tSchema->bytes);
if (NULL == taosArrayPush(pTagVals, &tv)) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
}
} else {
uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
terrno = TSDB_CODE_APP_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
break;
}
}
STag* pTag = NULL;
code = tTagNew(pTagVals, 1, false, &pTag);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
qError("failed to create tag, error:%s", tstrerror(code));
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
code = inserterBuildCreateTbReq(tbData.pCreateTbReq, tableName, pTag, suid, pInserter->pNode->tableName, TagNames,
pInserter->pTagSchema->nCols, TSDB_DEFAULT_TABLE_TTL);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
qError("failed to build create table request, error:%s", tstrerror(code));
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
for (int32_t k = 0; k < pTSchema->numOfCols; ++k) {
int16_t colIdx = k;
const STColumn* pCol = &pTSchema->columns[k];
int16_t* slotId = taosHashGet(pInserter->pCols, &pCol->colId, sizeof(pCol->colId));
if (NULL == slotId) {
continue;
}
colIdx = *slotId;
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, colIdx);
if (NULL == pColInfoData) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_VARCHAR: {
if (pColInfoData->info.type != pCol->type) {
qError("column:%d type:%d in block dismatch with schema col:%d type:%d", colIdx, pColInfoData->info.type, k,
pCol->type);
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
if (colDataIsNull_s(pColInfoData, j)) {
SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
if (NULL == taosArrayPush(pVals, &cv)) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
} else {
void* data = colDataGetVarData(pColInfoData, j);
SValue sv = (SValue){.type = pCol->type, .nData = varDataLen(data), .pData = varDataVal(data)};
SColVal cv = COL_VAL_VALUE(pCol->colId, sv);
if (NULL == taosArrayPush(pVals, &cv)) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
}
break;
}
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_MEDIUMBLOB:
qError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
terrno = TSDB_CODE_APP_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
break;
default:
if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
if (colDataIsNull_s(pColInfoData, j)) {
if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId) {
qError("Primary timestamp column should not be null");
terrno = TSDB_CODE_PAR_INCORRECT_TIMESTAMP_VAL;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
if (NULL == taosArrayPush(pVals, &cv)) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
} else {
// if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && !needSortMerge) {
// if (*(int64_t*)var <= lastTs) {
// needSortMerge = true;
// } else {
// lastTs = *(int64_t*)var;
// }
// }
SValue sv = {.type = pCol->type};
valueSetDatum(&sv, sv.type, var, tDataTypes[pCol->type].bytes);
SColVal cv = COL_VAL_VALUE(pCol->colId, sv);
if (NULL == taosArrayPush(pVals, &cv)) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
}
} else {
uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
terrno = TSDB_CODE_APP_ERROR;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
break;
}
}
SRow* pRow = NULL;
if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
if (NULL == taosArrayPush(tbData.aRowP, &pRow)) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
if (NULL == taosArrayPush(pReq->aSubmitTbData, &tbData)) {
goto _end;
}
}
_end:
taosArrayDestroy(pTagVals);
taosArrayDestroy(pVals);
return terrno;
}
int32_t buildSubmitReqFromBlock(SDataInserterHandle* pInserter, SSubmitReq2** ppReq, const SSDataBlock* pDataBlock,
const STSchema* pTSchema, int64_t* uid, int32_t* vgId, tb_uid_t* suid) {
SSubmitReq2* pReq = *ppReq;
SArray* pVals = NULL;
SArray* pTagVals = NULL;
int32_t numOfBlks = 0;
char* tableName = NULL;
int32_t code = 0, lino = 0;
terrno = TSDB_CODE_SUCCESS;
if (NULL == pReq) {
if (!(pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2)))) {
goto _end;
}
if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
goto _end;
}
}
int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
int32_t rows = pDataBlock->info.rows;
SSubmitTbData tbData = {0};
if (!(tbData.aRowP = taosArrayInit(rows, sizeof(SRow*)))) {
goto _end;
}
tbData.suid = *suid;
tbData.uid = *uid;
tbData.sver = pTSchema->version;
if (!pVals && !(pVals = taosArrayInit(colNum, sizeof(SColVal)))) {
taosArrayDestroy(tbData.aRowP);
goto _end;
}
if (pInserter->isStbInserter) {
if (!pTagVals && !(pTagVals = taosArrayInit(colNum, sizeof(STagVal)))) {
taosArrayDestroy(tbData.aRowP);
goto _end;
}
}
int64_t lastTs = TSKEY_MIN;
bool needSortMerge = false;
for (int32_t j = 0; j < rows; ++j) { // iterate by row
taosArrayClear(pVals);
int32_t offset = 0;
// 处理超级表的tbname和tags
if (pInserter->isStbInserter) {
taosArrayClear(pTagVals);
tbData.uid = 0;
*uid = 0;
tbData.pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
tbData.flags |= SUBMIT_REQ_AUTO_CREATE_TABLE;
SColumnInfoData* tbname = taosArrayGet(pDataBlock->pDataBlock, 0);
if (NULL == tbname) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
qError("Insert into stable must have tbname column");
goto _end;
}
if (tbname->info.type != TSDB_DATA_TYPE_BINARY) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
qError("tbname column must be binary");
goto _end;
}
if (colDataIsNull_s(tbname, j)) {
qError("insert into stable tbname column is null");
goto _end;
}
void* data = colDataGetVarData(tbname, j);
SValue sv = (SValue){TSDB_DATA_TYPE_VARCHAR, .nData = varDataLen(data),
.pData = varDataVal(data)}; // address copy, no value
SColVal cv = COL_VAL_VALUE(0, sv);
// 获取子表vgId
SDBVgInfo* dbInfo = NULL;
code = inserterGetDbVgInfo(pInserter, pInserter->dbFName, &dbInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
char tbFullName[TSDB_TABLE_FNAME_LEN];
taosMemoryFreeClear(tableName);
tableName = taosMemoryCalloc(1, sv.nData + 1);
TSDB_CHECK_NULL(tableName, code, lino, _end, terrno);
tstrncpy(tableName, sv.pData, sv.nData);
tableName[sv.nData] = '\0';
int32_t len = snprintf(tbFullName, TSDB_TABLE_FNAME_LEN, "%s.%s", pInserter->dbFName, tableName);
if (len >= TSDB_TABLE_FNAME_LEN) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
qError("table name too long after format, len:%d, maxLen:%d", len, TSDB_TABLE_FNAME_LEN);
goto _end;
}
code = inserterGetVgId(dbInfo, tbFullName, vgId);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
goto _end;
}
// 解析tag
SArray* TagNames = taosArrayInit(8, TSDB_COL_NAME_LEN);
if (!TagNames) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _end;
}
for (int32_t i = 0; i < pInserter->pTagSchema->nCols; ++i) {
SSchema* tSchema = &pInserter->pTagSchema->pSchema[i];
int16_t colIdx = tSchema->colId;
if (NULL == taosArrayPush(TagNames, tSchema->name)) {
goto _end;
}
int16_t* slotId = taosHashGet(pInserter->pCols, &colIdx, sizeof(colIdx));
if (NULL == slotId) {
continue;
}
colIdx = *slotId;
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, colIdx);
if (NULL == pColInfoData) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _end;
}
// void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_VARCHAR: { // TSDB_DATA_TYPE_BINARY
if (pColInfoData->info.type != tSchema->type) {
qError("tag:%d type:%d in block dismatch with schema tag:%d type:%d", colIdx, pColInfoData->info.type, i,
tSchema->type);
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _end;
}
if (colDataIsNull_s(pColInfoData, j)) {
continue;
} else {
void* data = colDataGetVarData(pColInfoData, j);
STagVal tv = (STagVal){.cid = tSchema->colId,
.type = tSchema->type,
.nData = varDataLen(data),
.pData = varDataVal(data)}; // address copy, no value
if (NULL == taosArrayPush(pTagVals, &tv)) {
goto _end;
}
}
break;
}
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_MEDIUMBLOB:
qError("the tag type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
terrno = TSDB_CODE_APP_ERROR;
goto _end;
break;
default:
if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
if (colDataIsNull_s(pColInfoData, j)) {
continue;
} else {
void* data = colDataGetData(pColInfoData, j);
STagVal tv = {.cid = tSchema->colId, .type = tSchema->type};
memcpy(&tv.i64, data, tSchema->bytes);
if (NULL == taosArrayPush(pTagVals, &tv)) {
goto _end;
}
}
} else {
uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
terrno = TSDB_CODE_APP_ERROR;
goto _end;
}
break;
}
}
STag* pTag = NULL;
code = tTagNew(pTagVals, 1, false, &pTag);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
qError("failed to create tag, error:%s", tstrerror(code));
goto _end;
}
inserterBuildCreateTbReq(tbData.pCreateTbReq, tableName, pTag, *suid, pInserter->pNode->tableName, TagNames,
pInserter->pTagSchema->nCols, TSDB_DEFAULT_TABLE_TTL);
}
for (int32_t k = 0; k < pTSchema->numOfCols; ++k) { // iterate by column
int16_t colIdx = k;
const STColumn* pCol = &pTSchema->columns[k];
if (!pInserter->fullOrderColList) {
int16_t* slotId = taosHashGet(pInserter->pCols, &pCol->colId, sizeof(pCol->colId));
if (NULL == slotId) {
continue;
}
colIdx = *slotId;
}
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, colIdx);
if (NULL == pColInfoData) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _end;
}
void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_VARCHAR: { // TSDB_DATA_TYPE_BINARY
if (pColInfoData->info.type != pCol->type) {
qError("column:%d type:%d in block dismatch with schema col:%d type:%d", colIdx, pColInfoData->info.type, k,
pCol->type);
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _end;
}
if (colDataIsNull_s(pColInfoData, j)) {
SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
if (NULL == taosArrayPush(pVals, &cv)) {
goto _end;
}
} else {
void* data = colDataGetVarData(pColInfoData, j);
SValue sv = (SValue){
.type = pCol->type, .nData = varDataLen(data), .pData = varDataVal(data)}; // address copy, no value
SColVal cv = COL_VAL_VALUE(pCol->colId, sv);
if (NULL == taosArrayPush(pVals, &cv)) {
goto _end;
}
}
break;
}
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_MEDIUMBLOB:
qError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
terrno = TSDB_CODE_APP_ERROR;
goto _end;
break;
default:
if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
if (colDataIsNull_s(pColInfoData, j)) {
if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId) {
qError("Primary timestamp column should not be null");
terrno = TSDB_CODE_PAR_INCORRECT_TIMESTAMP_VAL;
goto _end;
}
SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type); // should use pCol->type
if (NULL == taosArrayPush(pVals, &cv)) {
goto _end;
}
} else {
if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && !needSortMerge) {
if (*(int64_t*)var <= lastTs) {
needSortMerge = true;
} else {
lastTs = *(int64_t*)var;
}
}
SValue sv = {.type = pCol->type};
valueSetDatum(&sv, sv.type, var, tDataTypes[pCol->type].bytes);
SColVal cv = COL_VAL_VALUE(pCol->colId, sv);
if (NULL == taosArrayPush(pVals, &cv)) {
goto _end;
}
}
} else {
uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
terrno = TSDB_CODE_APP_ERROR;
goto _end;
}
break;
}
}
SRow* pRow = NULL;
if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) {
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end;
}
if (NULL == taosArrayPush(tbData.aRowP, &pRow)) {
goto _end;
}
}
if (needSortMerge) {
if ((tRowSort(tbData.aRowP) != TSDB_CODE_SUCCESS) ||
(terrno = tRowMerge(tbData.aRowP, (STSchema*)pTSchema, PREFER_NON_NULL)) != 0) {
goto _end;
}
}
if (NULL == taosArrayPush(pReq->aSubmitTbData, &tbData)) {
goto _end;
}
_end:
taosMemoryFreeClear(tableName);
taosArrayDestroy(pTagVals);
taosArrayDestroy(pVals);
if (terrno != 0) {
*ppReq = NULL;
if (pReq) {
tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq);
}
return terrno;
}
*ppReq = pReq;
return TSDB_CODE_SUCCESS;
}
int32_t dataBlocksToSubmitReqArray(SDataInserterHandle* pInserter, SArray* pMsgs) {
const SArray* pBlocks = pInserter->pDataBlocks;
const STSchema* pTSchema = pInserter->pSchema;
int64_t uid = pInserter->pNode->tableId;
int64_t suid = pInserter->pNode->stableId;
int32_t vgId = pInserter->pNode->vgId;
int32_t sz = taosArrayGetSize(pBlocks);
int32_t code = 0;
SHashObj* pHash = NULL;
void* iterator = NULL;
for (int32_t i = 0; i < sz; i++) {
SSDataBlock* pDataBlock = taosArrayGetP(pBlocks, i); // pDataBlock select查询到的结果
if (NULL == pDataBlock) {
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
if (pHash == NULL) {
pHash = taosHashInit(sz * pDataBlock->info.rows, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false,
HASH_ENTRY_LOCK);
if (NULL == pHash) {
return terrno;
}
}
code = buildSubmitReqFromStbBlock(pInserter, pHash, pDataBlock, pTSchema, uid, vgId, suid);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
}
size_t keyLen = 0;
while ((iterator = taosHashIterate(pHash, iterator))) {
SSubmitReq2* pReq = (SSubmitReq2*)iterator;
int32_t* ctbVgId = taosHashGetKey(iterator, &keyLen);
SSubmitTbDataMsg* pMsg = taosMemoryCalloc(1, sizeof(SSubmitTbDataMsg));
if (NULL == pMsg) {
code = terrno;
goto _end;
}
code = submitReqToMsg(*ctbVgId, pReq, &pMsg->pData, &pMsg->len);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
if (NULL == taosArrayPush(pMsgs, &pMsg)) {
code = terrno;
goto _end;
}
}
_end:
if (pHash != NULL) {
while ((iterator = taosHashIterate(pHash, iterator))) {
SSubmitReq2* pReq = (SSubmitReq2*)iterator;
if (pReq) {
tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
// taosMemoryFree(pReq);
}
}
taosHashCleanup(pHash);
}
return code;
}
int32_t dataBlocksToSubmitReq(SDataInserterHandle* pInserter, void** pMsg, int32_t* msgLen) {
const SArray* pBlocks = pInserter->pDataBlocks;
const STSchema* pTSchema = pInserter->pSchema;
int64_t uid = pInserter->pNode->tableId;
int64_t suid = pInserter->pNode->stableId;
int32_t vgId = pInserter->pNode->vgId;
int32_t sz = taosArrayGetSize(pBlocks);
int32_t code = 0;
SSubmitReq2* pReq = NULL;
for (int32_t i = 0; i < sz; i++) {
SSDataBlock* pDataBlock = taosArrayGetP(pBlocks, i); // pDataBlock select查询到的结果
if (NULL == pDataBlock) {
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
code = buildSubmitReqFromBlock(pInserter, &pReq, pDataBlock, pTSchema, &uid, &vgId, &suid);
if (code) {
if (pReq) {
tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq);
}
return code;
}
}
code = submitReqToMsg(vgId, pReq, pMsg, msgLen);
tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq);
return code;
}
static int32_t getStreamTableId(SStreamDataInserterInfo* pInserterInfo, SInsertTableRes* pTbInfo) {
int64_t key[2] = {pInserterInfo->streamId, pInserterInfo->groupId};
SInsertTableRes* pTbRes = taosHashGet(gStreamGrpTableHash, key, sizeof(key));
if (NULL == pTbRes) {
return TSDB_CODE_STREAM_INSERT_TBINFO_NOT_FOUND;
}
pTbInfo->uid = pTbRes->uid;
pTbInfo->version = pTbRes->version;
pTbInfo->vgid = pTbRes->vgid;
pTbInfo->tbname = pTbRes->tbname;
return TSDB_CODE_SUCCESS;
}
int32_t buildNormalTableCreateReq(SDataInserterHandle* pInserter, SStreamInserterParam* pInsertParam,
SSubmitTbData* tbData, SVgroupInfo* vgInfo) {
int32_t code = TSDB_CODE_SUCCESS;
tbData->suid = 0;
tbData->pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
if (NULL == tbData->pCreateTbReq) {
goto _end;
}
tbData->flags |= SUBMIT_REQ_AUTO_CREATE_TABLE;
tbData->pCreateTbReq->type = TSDB_NORMAL_TABLE;
tbData->pCreateTbReq->flags |= (TD_CREATE_NORMAL_TB_IN_STREAM | TD_CREATE_IF_NOT_EXISTS);
tbData->pCreateTbReq->uid = 0;
tbData->sver = pInsertParam->sver;
tbData->pCreateTbReq->name = taosStrdup(pInsertParam->tbname);
if (!tbData->pCreateTbReq->name) return terrno;
code = getDbVgInfoForExec(pInserter->pParam->readHandle->pMsgCb->clientRpc, pInsertParam->dbFName,
pInsertParam->tbname, vgInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
int32_t numOfCols = pInsertParam->pFields->size;
tbData->pCreateTbReq->ntb.schemaRow.nCols = numOfCols;
tbData->pCreateTbReq->ntb.schemaRow.version = pInsertParam->pSchema->version;
tbData->pCreateTbReq->ntb.schemaRow.pSchema = taosMemoryCalloc(numOfCols, sizeof(SSchema));
if (NULL == tbData->pCreateTbReq->ntb.schemaRow.pSchema) {
goto _end;
}
for (int32_t i = 0; i < numOfCols; ++i) {
SFieldWithOptions* pField = taosArrayGet(pInsertParam->pFields, i);
if (NULL == pField) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _end;
}
tbData->pCreateTbReq->ntb.schemaRow.pSchema[i].colId = i + 1;
tbData->pCreateTbReq->ntb.schemaRow.pSchema[i].type = pField->type;
tbData->pCreateTbReq->ntb.schemaRow.pSchema[i].bytes = pField->bytes;
tbData->pCreateTbReq->ntb.schemaRow.pSchema[i].flags = pField->flags;
if (i == 0 && pField->type != TSDB_DATA_TYPE_TIMESTAMP) {
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
qError("buildNormalTableCreateReq, the first column must be timestamp.");
goto _end;
}
if (i == 0) {
tbData->pCreateTbReq->ntb.schemaRow.pSchema[i].flags |= COL_IS_KEY;
}
snprintf(tbData->pCreateTbReq->ntb.schemaRow.pSchema[i].name, TSDB_COL_NAME_LEN, "%s", pField->name);
}
return TSDB_CODE_SUCCESS;
_end:
return code;
}
// reference tBuildTSchema funciton
static int32_t buildTSchmaFromInserter(SStreamInserterParam* pInsertParam, STSchema** ppTSchema) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t numOfCols = pInsertParam->pFields->size;
STSchema* pTSchema = taosMemoryCalloc(1, sizeof(STSchema) + sizeof(STColumn) * numOfCols);
if (NULL == pTSchema) {
return terrno;
}
if (pInsertParam->tbType == TSDB_NORMAL_TABLE) {
pTSchema->version =
1; // normal table version start from 1, if has exist table, it will be reset by resetInserterTbVersion
} else {
pTSchema->version = pInsertParam->sver;
}
pTSchema->numOfCols = numOfCols;
SFieldWithOptions* pField = taosArrayGet(pInsertParam->pFields, 0);
if (NULL == pField) {
code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _end;
}
pTSchema->columns[0].colId = PRIMARYKEY_TIMESTAMP_COL_ID;
pTSchema->columns[0].type = pField->type;
pTSchema->columns[0].flags = pField->flags | COL_IS_KEY;
pTSchema->columns[0].bytes = TYPE_BYTES[pField->type];
pTSchema->columns[0].offset = -1;
pTSchema->tlen = 0;
pTSchema->flen = 0;
for (int32_t i = 1; i < numOfCols; ++i) {
SFieldWithOptions* pField = taosArrayGet(pInsertParam->pFields, i);
if (NULL == pField) {
code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
goto _end;
}
pTSchema->columns[i].colId = i + 1;
pTSchema->columns[i].type = pField->type;
pTSchema->columns[i].flags = pField->flags;
pTSchema->columns[i].bytes = pField->bytes;
pTSchema->columns[i].offset = pTSchema->flen;
if (IS_VAR_DATA_TYPE(pField->type)) {
pTSchema->columns[i].bytes = pField->bytes;
pTSchema->tlen += (TYPE_BYTES[pField->type] + pField->bytes);
} else {
pTSchema->columns[i].bytes = TYPE_BYTES[pField->type];
pTSchema->tlen += TYPE_BYTES[pField->type];
}
pTSchema->flen += TYPE_BYTES[pField->type];
}
#if 1
pTSchema->tlen += (int32_t)TD_BITMAP_BYTES(numOfCols);
#endif
_end:
if (code != TSDB_CODE_SUCCESS) {
taosMemoryFree(pTSchema);
*ppTSchema = NULL;
} else {
*ppTSchema = pTSchema;
}
return code;
}
static int32_t getTagValsFromStreamInserterInfo(SStreamDataInserterInfo* pInserterInfo, int32_t preCols,
SArray** ppTagVals) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t nTags = pInserterInfo->pTagVals->size;
*ppTagVals = taosArrayInit(nTags, sizeof(STagVal));
if (!ppTagVals) {
return terrno;
}
for (int32_t i = 0; i < pInserterInfo->pTagVals->size; ++i) {
SStreamTagInfo* pTagInfo = taosArrayGet(pInserterInfo->pTagVals, i);
STagVal tagVal = {
.cid = preCols + i + 1,
.type = pTagInfo->val.data.type,
};
if (!pTagInfo->val.isNull) {
if (IS_VAR_DATA_TYPE(pTagInfo->val.data.type)) {
tagVal.nData = pTagInfo->val.data.nData;
tagVal.pData = pTagInfo->val.data.pData;
} else {
tagVal.i64 = pTagInfo->val.data.val;
}
if (NULL == taosArrayPush(*ppTagVals, &tagVal)) {
code = terrno;
goto _end;
}
}
}
_end:
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(*ppTagVals);
*ppTagVals = NULL;
}
return code;
}
static int32_t buildStreamSubTableCreateReq(SDataInserterHandle* pInserter, SStreamInserterParam* pInsertParam,
SStreamDataInserterInfo* pInserterInfo, SSubmitTbData* tbData,
SVgroupInfo* vgInfo) {
int32_t code = TSDB_CODE_SUCCESS;
STag* pTag = NULL;
SArray* pTagVals = NULL;
SArray* TagNames = NULL;
if (pInsertParam->pTagFields == NULL) {
stError("buildStreamSubTableCreateReq, pTagFields is NULL");
return TSDB_CODE_STREAM_INTERNAL_ERROR;
}
if (pInserterInfo->pTagVals == NULL || pInserterInfo->pTagVals->size == 0) {
stError("buildStreamSubTableCreateReq, pTagVals is NULL");
return TSDB_CODE_STREAM_INTERNAL_ERROR;
}
if (pInsertParam->suid <= 0 || pInsertParam->sver <= 0) {
stError("buildStreamSubTableCreateReq, suid:%" PRId64
", sver:%d"
" must be greater than 0",
pInsertParam->suid, pInsertParam->sver);
return TSDB_CODE_STREAM_INTERNAL_ERROR;
}
int32_t nTags = pInserterInfo->pTagVals->size;
TagNames = taosArrayInit(nTags, TSDB_COL_NAME_LEN);
if (!TagNames) {
code = terrno;
goto _end;
}
for (int32_t i = 0; i < nTags; ++i) {
SFieldWithOptions* pField = taosArrayGet(pInsertParam->pTagFields, i);
if (NULL == taosArrayPush(TagNames, pField->name)) {
goto _end;
}
}
tbData->flags |= SUBMIT_REQ_AUTO_CREATE_TABLE;
tbData->uid = 0;
tbData->suid = pInsertParam->suid;
tbData->sver = pInsertParam->sver;
tbData->pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
if (NULL == tbData->pCreateTbReq) {
code = terrno;
goto _end;
}
tbData->pCreateTbReq->type = TSDB_CHILD_TABLE;
tbData->pCreateTbReq->flags |= (TD_CREATE_SUB_TB_IN_STREAM | TD_CREATE_IF_NOT_EXISTS);
code = getDbVgInfoForExec(pInserter->pParam->readHandle->pMsgCb->clientRpc, pInsertParam->dbFName,
pInserterInfo->tbName, vgInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
code = getTagValsFromStreamInserterInfo(pInserterInfo, pInsertParam->pFields->size, &pTagVals);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
code = tTagNew(pTagVals, pInsertParam->sver, false, &pTag);
if (code != TSDB_CODE_SUCCESS) {
qError("failed to create tag, error:%s", tstrerror(code));
goto _end;
}
code = inserterBuildCreateTbReq(tbData->pCreateTbReq, pInserterInfo->tbName, pTag, tbData->suid,
pInsertParam->stbname, TagNames, nTags, TSDB_DEFAULT_TABLE_TTL);
if (code != TSDB_CODE_SUCCESS) {
qError("failed to build create table request, error:%s", tstrerror(code));
goto _end;
}
_end:
if (code != TSDB_CODE_SUCCESS) {
if (tbData->pCreateTbReq) {
taosMemoryFree(tbData->pCreateTbReq->name);
taosMemoryFree(tbData->pCreateTbReq);
}
}
if (TagNames) {
taosArrayDestroy(TagNames);
}
if (pTagVals) {
taosArrayDestroy(pTagVals);
}
return code;
}
static int32_t buildInsertData(SStreamInserterParam* pInsertParam, const SSDataBlock* pDataBlock,
SSubmitTbData* tbData) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t rows = pDataBlock->info.rows;
int32_t numOfCols = pInsertParam->pFields->size;
int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
STSchema* pTSchema = pInsertParam->pSchema;
int64_t lastTs = TSKEY_MIN;
bool needSortMerge = false;
SArray* pVals = NULL;
if (!(pVals = taosArrayInit(colNum, sizeof(SColVal)))) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
for (int32_t j = 0; j < rows; ++j) { // iterate by row
taosArrayClear(pVals);
bool tsIsNull = false;
for (int32_t k = 0; k < numOfCols; ++k) { // iterate by column
int16_t colIdx = k + 1;
SFieldWithOptions* pCol = taosArrayGet(pInsertParam->pFields, k);
if (PRIMARYKEY_TIMESTAMP_COL_ID != colIdx && TSDB_DATA_TYPE_NULL == pCol->type) {
SColVal cv = COL_VAL_NULL(colIdx, pCol->type);
if (NULL == taosArrayPush(pVals, &cv)) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
continue;
}
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
if (NULL == pColInfoData) {
code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
QUERY_CHECK_CODE(code, lino, _end);
}
void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_VARCHAR: { // TSDB_DATA_TYPE_BINARY
if (pColInfoData->info.type != pCol->type) {
qError("tb:%s column:%d type:%d in block dismatch with schema col:%d type:%d", pInsertParam->tbname, k,
pColInfoData->info.type, k, pCol->type);
code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
QUERY_CHECK_CODE(code, lino, _end);
}
if (colDataIsNull_s(pColInfoData, j)) {
SColVal cv = COL_VAL_NULL(colIdx, pCol->type);
if (NULL == taosArrayPush(pVals, &cv)) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
} else {
if (pColInfoData->pData == NULL) {
qError("build insert tb:%s, column:%d data is NULL in block", pInsertParam->tbname, k);
code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
QUERY_CHECK_CODE(code, lino, _end);
}
void* data = colDataGetVarData(pColInfoData, j);
SValue sv = (SValue){
.type = pCol->type, .nData = varDataLen(data), .pData = varDataVal(data)}; // address copy, no value
SColVal cv = COL_VAL_VALUE(colIdx, sv);
if (NULL == taosArrayPush(pVals, &cv)) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
}
break;
}
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_MEDIUMBLOB:
qError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
code = TSDB_CODE_APP_ERROR;
QUERY_CHECK_CODE(code, lino, _end);
break;
default:
if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
if (colDataIsNull_s(pColInfoData, j)) {
if (PRIMARYKEY_TIMESTAMP_COL_ID == colIdx) {
tsIsNull = true;
qInfo("Primary timestamp column should not be null, skip this row");
break;
}
SColVal cv = COL_VAL_NULL(colIdx, pCol->type); // should use pCol->type
if (NULL == taosArrayPush(pVals, &cv)) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
} else {
if (PRIMARYKEY_TIMESTAMP_COL_ID == colIdx && !needSortMerge) {
if (*(int64_t*)var <= lastTs) {
needSortMerge = true;
} else {
lastTs = *(int64_t*)var;
}
}
SValue sv = {.type = pCol->type};
valueSetDatum(&sv, sv.type, var, tDataTypes[pCol->type].bytes);
SColVal cv = COL_VAL_VALUE(colIdx, sv);
if (NULL == taosArrayPush(pVals, &cv)) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
}
} else {
uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
code = TSDB_CODE_APP_ERROR;
QUERY_CHECK_CODE(code, lino, _end);
}
break;
}
if(tsIsNull) break; // skip remaining columns because the primary key is null
}
if(tsIsNull) continue; // skip this row if primary key is null
SRow* pRow = NULL;
if ((code = tRowBuild(pVals, pTSchema, &pRow)) != TSDB_CODE_SUCCESS) {
QUERY_CHECK_CODE(code, lino, _end);
}
if (NULL == taosArrayPush(tbData->aRowP, &pRow)) {
taosMemFree(pRow);
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
}
if(taosArrayGetSize(tbData->aRowP) == 0) {
stDebug("no valid data to insert, skip this block");
code = TSDB_CODE_STREAM_NO_DATA;
}
if (needSortMerge) {
if ((tRowSort(tbData->aRowP) != TSDB_CODE_SUCCESS) ||
(code = tRowMerge(tbData->aRowP, (STSchema*)pTSchema, KEEP_CONSISTENCY)) != 0) {
QUERY_CHECK_CODE(code, lino, _end);
}
}
_end:
taosArrayDestroy(pVals);
if (code != TSDB_CODE_SUCCESS) {
tDestroySubmitTbData(tbData, TSDB_MSG_FLG_ENCODE);
}
return code;
}
// todo 和 buildStreamSubmitReqFromBlock 总的公共部分提取接口，待其他修改稳定后进行防止多人修改冲突
int32_t buildStreamSubmitReqFromBlock(SDataInserterHandle* pInserter, SStreamDataInserterInfo* pInserterInfo,
SSubmitReq2** ppReq, const SSDataBlock* pDataBlock, SVgroupInfo* vgInfo) {
SSubmitReq2* pReq = *ppReq;
int32_t numOfBlks = 0;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamInserterParam* pInsertParam = pInserter->pParam->streamInserterParam;
if (NULL == pReq) {
if (!(pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2)))) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
*ppReq = pReq;
if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
}
STSchema* pTSchema = pInsertParam->pSchema;
int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
int32_t rows = pDataBlock->info.rows;
SSubmitTbData tbData = {0};
if (!(tbData.aRowP = taosArrayInit(rows, sizeof(SRow*)))) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
if (pInserterInfo->isAutoCreateTable) {
if (pInsertParam->tbType == TSDB_NORMAL_TABLE) {
code = buildNormalTableCreateReq(pInserter, pInsertParam, &tbData, vgInfo);
} else if (pInsertParam->tbType == TSDB_SUPER_TABLE) {
code = buildStreamSubTableCreateReq(pInserter, pInsertParam, pInserterInfo, &tbData, vgInfo);
} else {
code = TSDB_CODE_MND_STREAM_INTERNAL_ERROR;
stError("buildStreamSubmitReqFromBlock, unknown table type %d", pInsertParam->tbType);
}
QUERY_CHECK_CODE(code, lino, _end);
code = initTableInfo(pInserterInfo);
QUERY_CHECK_CODE(code, lino, _end);
} else {
SInsertTableRes tbInfo = {0};
code = getStreamTableId(pInserterInfo, &tbInfo);
QUERY_CHECK_CODE(code, lino, _end);
pInserterInfo->tbName = tbInfo.tbname; // pInserterInfo->tbName wouldn't be delete
tbData.uid = tbInfo.uid;
tbData.sver = tbInfo.version;
code = getExistVgInfo(pInserter, pInsertParam, pInserterInfo, vgInfo);
QUERY_CHECK_CODE(code, lino, _end);
if (pInsertParam->tbType == TSDB_SUPER_TABLE) {
tbData.suid = pInsertParam->suid;
tbData.sver = pInsertParam->sver;
}
}
stDebug("[data inserter], Handle:%p, STREAM:0x%" PRIx64 " GROUP:%" PRId64 " tbname:%s autoCreate:%d uid:%" PRId64
" suid:%" PRId64 " sver:%d vgid:%d",
pInserter, pInserterInfo->streamId, pInserterInfo->groupId, pInserterInfo->tbName,
pInserterInfo->isAutoCreateTable, tbData.uid, tbData.suid, tbData.sver, vgInfo->vgId);
code = buildInsertData(pInsertParam, pDataBlock, &tbData);
QUERY_CHECK_CODE(code, lino, _end);
if (NULL == taosArrayPush(pReq->aSubmitTbData, &tbData)) {
code = terrno;
tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
QUERY_CHECK_CODE(code, lino, _end);
}
_end:
if (code != 0) {
if (tbData.aRowP) {
taosArrayDestroy(tbData.aRowP);
}
}
return code;
}
int32_t streamDataBlocksToSubmitReq(SDataInserterHandle* pInserter, SStreamDataInserterInfo* pInserterInfo, void** pMsg,
int32_t* msgLen, SVgroupInfo* vgInfo) {
const SArray* pBlocks = pInserter->pDataBlocks;
int32_t sz = taosArrayGetSize(pBlocks);
int32_t code = 0;
SSubmitReq2* pReq = NULL;
for (int32_t i = 0; i < sz; i++) {
SSDataBlock* pDataBlock = taosArrayGetP(pBlocks, i); // pDataBlock select查询到的结果
if (NULL == pDataBlock) {
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
stDebug("[data inserter], Handle:%p, STREAM:0x%" PRIx64 " GROUP:%" PRId64
" tbname:%s autoCreate:%d block: %d/%d rows:%" PRId64,
pInserter, pInserterInfo->streamId, pInserterInfo->groupId, pInserterInfo->tbName,
pInserterInfo->isAutoCreateTable, i + 1, sz, pDataBlock->info.rows);
code = buildStreamSubmitReqFromBlock(pInserter, pInserterInfo, &pReq, pDataBlock, vgInfo);
if (code) {
if (pReq) {
tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq);
}
return code;
}
}
code = submitReqToMsg(vgInfo->vgId, pReq, pMsg, msgLen);
tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq);
stDebug("[data inserter], submit req, vgid:%d, TREAM:0x%" PRIx64 " GROUP:%" PRId64 " tbname:%s autoCreate:%d code:%d ",
vgInfo->vgId, pInserterInfo->streamId, pInserterInfo->groupId, pInserterInfo->tbName,
pInserterInfo->isAutoCreateTable, code);
return code;
}
static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;
if (!pInserter->explain) {
if (NULL == taosArrayPush(pInserter->pDataBlocks, &pInput->pData)) {
return terrno;
}
if (pInserter->isStbInserter) {
SArray* pMsgs = taosArrayInit(4, sizeof(POINTER_BYTES));
if (NULL == pMsgs) {
return terrno;
}
int32_t code = dataBlocksToSubmitReqArray(pInserter, pMsgs);
if (code) {
taosArrayDestroyP(pMsgs, destroySSubmitTbDataMsg);
return code;
}
taosArrayClear(pInserter->pDataBlocks);
for (int32_t i = 0; i < taosArrayGetSize(pMsgs); ++i) {
SSubmitTbDataMsg* pMsg = taosArrayGetP(pMsgs, i);
code = sendSubmitRequest(pInserter, NULL, pMsg->pData, pMsg->len,
pInserter->pParam->readHandle->pMsgCb->clientRpc, &pInserter->pNode->epSet);
if (code) {
for (int j = i + 1; j < taosArrayGetSize(pMsgs); ++j) {
SSubmitTbDataMsg* pMsg2 = taosArrayGetP(pMsgs, j);
destroySSubmitTbDataMsg(pMsg2);
}
taosArrayDestroy(pMsgs);
return code;
}
QRY_ERR_RET(tsem_wait(&pInserter->ready));
if (pInserter->submitRes.code) {
for (int j = i + 1; j < taosArrayGetSize(pMsgs); ++j) {
SSubmitTbDataMsg* pMsg2 = taosArrayGetP(pMsgs, j);
destroySSubmitTbDataMsg(pMsg2);
}
taosArrayDestroy(pMsgs);
return pInserter->submitRes.code;
}
}
taosArrayDestroy(pMsgs);
} else {
void* pMsg = NULL;
int32_t msgLen = 0;
int32_t code = dataBlocksToSubmitReq(pInserter, &pMsg, &msgLen);
if (code) {
return code;
}
taosArrayClear(pInserter->pDataBlocks);
code = sendSubmitRequest(pInserter, NULL, pMsg, msgLen, pInserter->pParam->readHandle->pMsgCb->clientRpc,
&pInserter->pNode->epSet);
if (code) {
return code;
}
QRY_ERR_RET(tsem_wait(&pInserter->ready));
if (pInserter->submitRes.code) {
return pInserter->submitRes.code;
}
}
}
*pContinue = true;
return TSDB_CODE_SUCCESS;
}
static int32_t resetInserterTbVersion(SDataInserterHandle* pInserter, const SInputData* pInput) {
SInsertTableRes pTbInfo = {0};
int32_t code = getStreamTableId(pInput->pStreamDataInserterInfo, &pTbInfo);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
stDebug("resetInserterTbVersion, streamId:0x%" PRIx64 " groupId:%" PRId64 " tbName:%s, uid:%" PRId64 ", version:%d",
pInput->pStreamDataInserterInfo->streamId, pInput->pStreamDataInserterInfo->groupId,
pInput->pStreamDataInserterInfo->tbName, pTbInfo.uid, pTbInfo.version);
pInserter->pParam->streamInserterParam->pSchema->version = pTbInfo.version;
if (pInserter->pParam->streamInserterParam->tbType != TSDB_NORMAL_TABLE) {
pInserter->pParam->streamInserterParam->sver = pTbInfo.version;
}
return code;
}
static int32_t putStreamDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
int32_t code = 0;
int32_t lino = 0;
SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;
if (!pInserter || !pInserter->pParam || !pInserter->pParam->streamInserterParam) {
stError("putStreamDataBlock invalid param, pInserter: %p, pParam:%p", pInserter, pInserter ? pInserter->pParam : NULL);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
if (!pInserter->explain) {
code = TSDB_CODE_SUCCESS;
if (NULL == taosArrayPush(pInserter->pDataBlocks, &pInput->pData)) {
return terrno;
}
void* pMsg = NULL;
int32_t msgLen = 0;
SVgroupInfo vgInfo = {0};
if (pInserter->pParam->streamInserterParam->pSchema == NULL) {
code = buildTSchmaFromInserter(pInserter->pParam->streamInserterParam,
&pInserter->pParam->streamInserterParam->pSchema);
QUERY_CHECK_CODE(code, lino, _return);
}
code = streamDataBlocksToSubmitReq(pInserter, pInput->pStreamDataInserterInfo, &pMsg, &msgLen, &vgInfo);
QUERY_CHECK_CODE(code, lino, _return);
code = sendSubmitRequest(pInserter, pInput->pStreamDataInserterInfo, pMsg, msgLen,
pInserter->pParam->readHandle->pMsgCb->clientRpc, &vgInfo.epSet);
QUERY_CHECK_CODE(code, lino, _return);
code = tsem_wait(&pInserter->ready);
QUERY_CHECK_CODE(code, lino, _return);
if (pInserter->submitRes.code == TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
pInput->pStreamDataInserterInfo->isAutoCreateTable = false;
code = resetInserterTbVersion(pInserter, pInput);
QUERY_CHECK_CODE(code, lino, _return);
code = streamDataBlocksToSubmitReq(pInserter, pInput->pStreamDataInserterInfo, &pMsg, &msgLen, &vgInfo);
QUERY_CHECK_CODE(code, lino, _return);
code = sendSubmitRequest(pInserter, pInput->pStreamDataInserterInfo, pMsg, msgLen,
pInserter->pParam->readHandle->pMsgCb->clientRpc, &vgInfo.epSet);
QUERY_CHECK_CODE(code, lino, _return);
code = tsem_wait(&pInserter->ready);
QUERY_CHECK_CODE(code, lino, _return);
}
if ((pInserter->submitRes.code == TSDB_CODE_TDB_TABLE_NOT_EXIST &&
!pInput->pStreamDataInserterInfo->isAutoCreateTable) || pInserter->submitRes.code == TSDB_CODE_VND_INVALID_VGROUP_ID) {
rmDbVgInfoFromCache(pInserter->pParam->streamInserterParam->dbFName);
stInfo("putStreamDataBlock, stream inserter table info not found, groupId:%" PRId64
", tbName:%s. so reset dbVgInfo",
pInput->pStreamDataInserterInfo->groupId, pInput->pStreamDataInserterInfo->tbName);
code = TSDB_CODE_STREAM_INSERT_TBINFO_NOT_FOUND;
QUERY_CHECK_CODE(code, lino, _return);
}
if (pInserter->submitRes.code) {
code = pInserter->submitRes.code;
stError("submitRes err:%s, code:%0x", tstrerror(pInserter->submitRes.code), pInserter->submitRes.code);
QUERY_CHECK_CODE(code, lino, _return);
}
*pContinue = true;
_return:
taosArrayClear(pInserter->pDataBlocks);
if(code == TSDB_CODE_STREAM_NO_DATA) {
stDebug("putStreamDataBlock, no valid data to insert, skip this block");
code = TSDB_CODE_SUCCESS;
} else if (code) {
stError("submitRes err:%s, code:%0x lino:%d", tstrerror(code), code, lino);
return code;
}
return code;
}
return TSDB_CODE_SUCCESS;
}
static void endPut(struct SDataSinkHandle* pHandle, uint64_t useconds) {
SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;
(void)taosThreadMutexLock(&pInserter->mutex);
pInserter->queryEnd = true;
pInserter->useconds = useconds;
(void)taosThreadMutexUnlock(&pInserter->mutex);
}
static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, int64_t* pRawLen, bool* pQueryEnd) {
SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;
*pLen = pDispatcher->submitRes.affectedRows;
qDebug("got total affectedRows %" PRId64, *pLen);
}
static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;
(void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pInserter->cachedSize);
taosArrayDestroy(pInserter->pDataBlocks);
taosMemoryFree(pInserter->pSchema);
if (pInserter->pParam->streamInserterParam) {
destroyStreamInserterParam(pInserter->pParam->streamInserterParam);
taosMemoryFree(pInserter->pParam->readHandle); // only for stream
}
taosMemoryFree(pInserter->pParam);
taosHashCleanup(pInserter->pCols);
nodesDestroyNode((SNode*)pInserter->pNode);
pInserter->pNode = NULL;
(void)taosThreadMutexDestroy(&pInserter->mutex);
taosMemoryFree(pInserter->pManager);
if (pInserter->dbVgInfoMap) {
taosHashCleanup(pInserter->dbVgInfoMap);
}
return TSDB_CODE_SUCCESS;
}
static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {
SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;
*size = atomic_load_64(&pDispatcher->cachedSize);
return TSDB_CODE_SUCCESS;
}
static int32_t getSinkFlags(struct SDataSinkHandle* pHandle, uint64_t* pFlags) {
SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;
*pFlags = atomic_load_64(&pDispatcher->flags);
return TSDB_CODE_SUCCESS;
}
int32_t createDataInserter(SDataSinkManager* pManager, SDataSinkNode** ppDataSink, DataSinkHandle* pHandle,
void* pParam) {
SDataSinkNode* pDataSink = *ppDataSink;
SDataInserterHandle* inserter = taosMemoryCalloc(1, sizeof(SDataInserterHandle));
if (NULL == inserter) {
taosMemoryFree(pParam);
goto _return;
}
SQueryInserterNode* pInserterNode = (SQueryInserterNode*)pDataSink;
inserter->sink.fPut = putDataBlock;
inserter->sink.fEndPut = endPut;
inserter->sink.fGetLen = getDataLength;
inserter->sink.fGetData = NULL;
inserter->sink.fDestroy = destroyDataSinker;
inserter->sink.fGetCacheSize = getCacheSize;
inserter->sink.fGetFlags = getSinkFlags;
inserter->pManager = pManager;
inserter->pNode = pInserterNode;
inserter->pParam = pParam;
inserter->status = DS_BUF_EMPTY;
inserter->queryEnd = false;
inserter->explain = pInserterNode->explain;
*ppDataSink = NULL;
int64_t suid = 0;
int32_t code = pManager->pAPI->metaFn.getTableSchema(inserter->pParam->readHandle->vnode, pInserterNode->tableId,
&inserter->pSchema, &suid, &inserter->pTagSchema);
if (code) {
terrno = code;
goto _return;
}
pManager->pAPI->metaFn.getBasicInfo(inserter->pParam->readHandle->vnode, &inserter->dbFName, NULL, NULL, NULL);
if (pInserterNode->tableType == TSDB_SUPER_TABLE) {
inserter->isStbInserter = true;
}
if (pInserterNode->stableId != suid) {
terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
goto _return;
}
inserter->pDataBlocks = taosArrayInit(1, POINTER_BYTES);
if (NULL == inserter->pDataBlocks) {
goto _return;
}
QRY_ERR_JRET(taosThreadMutexInit(&inserter->mutex, NULL));
inserter->fullOrderColList = pInserterNode->pCols->length == inserter->pSchema->numOfCols;
inserter->pCols = taosHashInit(pInserterNode->pCols->length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT),
false, HASH_NO_LOCK);
if (NULL == inserter->pCols) {
goto _return;
}
SNode* pNode = NULL;
int32_t i = 0;
bool foundTbname = false;
FOREACH(pNode, pInserterNode->pCols) {
if (pNode->type == QUERY_NODE_FUNCTION && ((SFunctionNode*)pNode)->funcType == FUNCTION_TYPE_TBNAME) {
int16_t colId = 0;
int16_t slotId = 0;
QRY_ERR_JRET(taosHashPut(inserter->pCols, &colId, sizeof(colId), &slotId, sizeof(slotId)));
foundTbname = true;
continue;
}
SColumnNode* pCol = (SColumnNode*)pNode;
QRY_ERR_JRET(taosHashPut(inserter->pCols, &pCol->colId, sizeof(pCol->colId), &pCol->slotId, sizeof(pCol->slotId)));
if (inserter->fullOrderColList && pCol->colId != inserter->pSchema->columns[i].colId) {
inserter->fullOrderColList = false;
}
++i;
}
if (inserter->isStbInserter && !foundTbname) {
QRY_ERR_JRET(TSDB_CODE_PAR_TBNAME_ERROR);
}
QRY_ERR_JRET(tsem_init(&inserter->ready, 0, 0));
inserter->dbVgInfoMap = NULL;
*pHandle = inserter;
return TSDB_CODE_SUCCESS;
_return:
if (inserter) {
(void)destroyDataSinker((SDataSinkHandle*)inserter);
taosMemoryFree(inserter);
} else {
taosMemoryFree(pManager);
}
nodesDestroyNode((SNode*)*ppDataSink);
*ppDataSink = NULL;
return terrno;
}
static TdThreadOnce g_dbVgInfoMgrInit = PTHREAD_ONCE_INIT;
SDBVgInfoMgr g_dbVgInfoMgr = {0};
void dbVgInfoMgrInitOnce() {
g_dbVgInfoMgr.dbVgInfoMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
if (g_dbVgInfoMgr.dbVgInfoMap == NULL) {
stError("%s failed at line %d, error:%s", __FUNCTION__, __LINE__, tstrerror(terrno));
return;
}
taosHashSetFreeFp(g_dbVgInfoMgr.dbVgInfoMap, freeUseDbOutput_tmp);
}
int32_t createStreamDataInserter(SDataSinkManager* pManager, DataSinkHandle* pHandle, void* pParam) {
int32_t code = TSDB_CODE_SUCCESS, lino = 0;
TAOS_UNUSED(taosThreadOnce(&g_dbVgInfoMgrInit, dbVgInfoMgrInitOnce));
TSDB_CHECK_NULL(g_dbVgInfoMgr.dbVgInfoMap, code, lino, _exit, terrno);
SDataInserterHandle* inserter = taosMemoryCalloc(1, sizeof(SDataInserterHandle));
TSDB_CHECK_NULL(inserter, code, lino, _exit, terrno);
inserter->sink.fPut = putStreamDataBlock;
inserter->sink.fEndPut = endPut;
inserter->sink.fGetLen = getDataLength;
inserter->sink.fGetData = NULL;
inserter->sink.fDestroy = destroyDataSinker;
inserter->sink.fGetCacheSize = getCacheSize;
inserter->sink.fGetFlags = getSinkFlags;
inserter->pManager = pManager;
inserter->pNode = NULL;
inserter->pParam = pParam;
inserter->status = DS_BUF_EMPTY;
inserter->queryEnd = false;
inserter->explain = false;
inserter->pDataBlocks = taosArrayInit(1, POINTER_BYTES);
TSDB_CHECK_NULL(inserter->pDataBlocks, code, lino, _exit, terrno);
TAOS_CHECK_EXIT(taosThreadMutexInit(&inserter->mutex, NULL));
TAOS_CHECK_EXIT(tsem_init(&inserter->ready, 0, 0));
inserter->dbVgInfoMap = NULL;
*pHandle = inserter;
return TSDB_CODE_SUCCESS;
_exit:
if (inserter) {
(void)destroyDataSinker((SDataSinkHandle*)inserter);
taosMemoryFree(inserter);
} else {
taosMemoryFree(pManager);
}
if (code) {
stError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
}
return code;
}
int32_t getDbVgInfoByTbName(void* clientRpc, const char* dbFName, SDBVgInfo** dbVgInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t line = 0;
SUseDbOutput* output = NULL;
SUseDbOutput** find = (SUseDbOutput**)taosHashGet(g_dbVgInfoMgr.dbVgInfoMap, dbFName, strlen(dbFName));
if (find == NULL) {
output = taosMemoryCalloc(1, sizeof(SUseDbOutput));
if (output == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
code = buildDbVgInfoMap(clientRpc, dbFName, output);
QUERY_CHECK_CODE(code, line, _return);
code = taosHashPut(g_dbVgInfoMgr.dbVgInfoMap, dbFName, strlen(dbFName), &output, POINTER_BYTES);
if (code == TSDB_CODE_DUP_KEY) {
code = TSDB_CODE_SUCCESS;
// another thread has put the same dbFName, so we need to free the output
freeUseDbOutput_tmp(&output);
find = (SUseDbOutput**)taosHashGet(g_dbVgInfoMgr.dbVgInfoMap, dbFName, strlen(dbFName));
if (find == NULL) {
QUERY_CHECK_CODE(code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, line, _return);
}
output = *find;
}
QUERY_CHECK_CODE(code, line, _return);
} else {
output = *find;
}
*dbVgInfo = output->dbVgroup;
return code;
_return:
qError("%s failed at line %d since %s", __func__, line, tstrerror(code));
freeUseDbOutput_tmp(&output);
return code;
}
int32_t getDbVgInfoForExec(void* clientRpc, const char* dbFName, const char* tbName, SVgroupInfo* pVgInfo) {
SDBVgInfo* dbInfo = NULL;
int32_t code = 0, lino = 0;
char tbFullName[TSDB_TABLE_FNAME_LEN];
snprintf(tbFullName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbFName, tbName);
taosRLockLatch(&g_dbVgInfoMgr.lock);
TAOS_CHECK_EXIT(getDbVgInfoByTbName(clientRpc, dbFName, &dbInfo));
TAOS_CHECK_EXIT(inserterGetVgInfo(dbInfo, tbFullName, pVgInfo));
_exit:
taosRUnLockLatch(&g_dbVgInfoMgr.lock);
if (code) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
void rmDbVgInfoFromCache(const char* dbFName) {
taosWLockLatch(&g_dbVgInfoMgr.lock);
taosHashRemove(g_dbVgInfoMgr.dbVgInfoMap, dbFName, strlen(dbFName));
taosWUnLockLatch(&g_dbVgInfoMgr.lock);
}
Reference in a new issue View git blame Copy permalink