1466 lines
52 KiB
C
1466 lines
52 KiB
C
/*
|
|
* 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 "tcommon.h"
|
|
#include "tq.h"
|
|
|
|
typedef struct STableSinkInfo {
|
|
uint64_t uid;
|
|
tstr name;
|
|
} STableSinkInfo;
|
|
|
|
static bool hasOnlySubmitData(const SArray* pBlocks, int32_t numOfBlocks);
|
|
static int32_t tsAscendingSortFn(const void* p1, const void* p2);
|
|
static int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDataBlock, char* stbFullName,
|
|
SSubmitTbData* pTableData);
|
|
static int32_t doBuildAndSendDeleteMsg(SVnode* pVnode, char* stbFullName, SSDataBlock* pDataBlock, SStreamTask* pTask,
|
|
int64_t suid);
|
|
static int32_t doBuildAndSendSubmitMsg(SVnode* pVnode, SStreamTask* pTask, SSubmitReq2* pReq, int32_t numOfBlocks);
|
|
static int32_t buildSubmitMsgImpl(SSubmitReq2* pSubmitReq, int32_t vgId, void** pMsg, int32_t* msgLen);
|
|
static int32_t doConvertRows(SSubmitTbData* pTableData, const STSchema* pTSchema, SSDataBlock* pDataBlock,
|
|
int64_t earlyTs, const char* id);
|
|
static int32_t doWaitForDstTableCreated(SVnode* pVnode, SStreamTask* pTask, STableSinkInfo* pTableSinkInfo,
|
|
const char* dstTableName, int64_t* uid);
|
|
|
|
static bool isValidDstChildTable(SMetaReader* pReader, int32_t vgId, const char* ctbName, int64_t suid);
|
|
static int32_t initCreateTableMsg(SVCreateTbReq* pCreateTableReq, uint64_t suid, const char* stbFullName,
|
|
int32_t numOfTags);
|
|
static int32_t createDefaultTagColName(SArray** pColNameList);
|
|
static int32_t setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock,
|
|
const char* stbFullName, int64_t gid, bool newSubTableRule, const char* id);
|
|
static int32_t doCreateSinkTableInfo(const char* pDstTableName, STableSinkInfo** pInfo);
|
|
static int32_t doPutSinkTableInfoIntoCache(SSHashObj* pSinkTableMap, STableSinkInfo* pTableSinkInfo, uint64_t groupId,
|
|
const char* id);
|
|
static bool doGetSinkTableInfoFromCache(SSHashObj* pTableInfoMap, uint64_t groupId, STableSinkInfo** pInfo);
|
|
static int32_t doRemoveSinkTableInfoInCache(SSHashObj* pSinkTableMap, uint64_t groupId, const char* id);
|
|
static int32_t checkTagSchema(SStreamTask* pTask, SVnode* pVnode);
|
|
static void rebuildAndSendMultiResBlock(SStreamTask* pTask, const SArray* pBlocks, SVnode* pVnode, int64_t earlyTs);
|
|
static int32_t handleResultBlockMsg(SStreamTask* pTask, SSDataBlock* pDataBlock, int32_t index, SVnode* pVnode,
|
|
int64_t earlyTs);
|
|
static int32_t doWaitForDstTableDropped(SVnode* pVnode, SStreamTask* pTask, const char* dstTableName);
|
|
|
|
int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
|
|
const char* pIdStr, bool newSubTableRule) {
|
|
int32_t totalRows = pDataBlock->info.rows;
|
|
SColumnInfoData* pStartTsCol = taosArrayGet(pDataBlock->pDataBlock, START_TS_COLUMN_INDEX);
|
|
SColumnInfoData* pEndTsCol = taosArrayGet(pDataBlock->pDataBlock, END_TS_COLUMN_INDEX);
|
|
SColumnInfoData* pGidCol = taosArrayGet(pDataBlock->pDataBlock, GROUPID_COLUMN_INDEX);
|
|
SColumnInfoData* pTbNameCol = taosArrayGet(pDataBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
|
|
|
|
if (pStartTsCol == NULL || pEndTsCol == NULL || pGidCol == NULL || pTbNameCol == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
tqDebug("s-task:%s build %d rows delete msg for table:%s", pIdStr, totalRows, stbFullName);
|
|
|
|
for (int32_t row = 0; row < totalRows; row++) {
|
|
int64_t skey = *(int64_t*)colDataGetData(pStartTsCol, row);
|
|
int64_t ekey = *(int64_t*)colDataGetData(pEndTsCol, row);
|
|
int64_t groupId = *(int64_t*)colDataGetData(pGidCol, row);
|
|
|
|
char* name = NULL;
|
|
char* originName = NULL;
|
|
void* varTbName = NULL;
|
|
if (!colDataIsNull(pTbNameCol, totalRows, row, NULL)) {
|
|
varTbName = colDataGetVarData(pTbNameCol, row);
|
|
}
|
|
|
|
if (varTbName != NULL && varTbName != (void*)-1) {
|
|
size_t cap = TMAX(TSDB_TABLE_NAME_LEN, varDataLen(varTbName) + 1);
|
|
name = taosMemoryMalloc(cap);
|
|
if (name == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
memcpy(name, varDataVal(varTbName), varDataLen(varTbName));
|
|
name[varDataLen(varTbName)] = '\0';
|
|
if (newSubTableRule && !isAutoTableName(name) && !alreadyAddGroupId(name, groupId) && groupId != 0 &&
|
|
stbFullName) {
|
|
int32_t code = buildCtbNameAddGroupId(stbFullName, name, groupId, cap);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
return code;
|
|
}
|
|
}
|
|
} else if (stbFullName) {
|
|
int32_t code = buildCtbNameByGroupId(stbFullName, groupId, &name);
|
|
if (code) {
|
|
return code;
|
|
}
|
|
} else {
|
|
originName = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE);
|
|
if (originName == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
if (metaGetTableNameByUid(pTq->pVnode, groupId, originName) == 0) {
|
|
name = varDataVal(originName);
|
|
}
|
|
}
|
|
|
|
if (!name || *name == '\0') {
|
|
tqWarn("s-task:%s failed to build delete msg groupId:%" PRId64 ", skey:%" PRId64 " ekey:%" PRId64
|
|
" since invalid tbname:%s",
|
|
pIdStr, groupId, skey, ekey, name ? name : "NULL");
|
|
} else {
|
|
tqDebug("s-task:%s build delete msg groupId:%" PRId64 ", name:%s, skey:%" PRId64 " ekey:%" PRId64, pIdStr,
|
|
groupId, name, skey, ekey);
|
|
|
|
SSingleDeleteReq req = {.startTs = skey, .endTs = ekey};
|
|
tstrncpy(req.tbname, name, TSDB_TABLE_NAME_LEN);
|
|
void* p = taosArrayPush(deleteReq->deleteReqs, &req);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
}
|
|
|
|
if (originName) {
|
|
name = originName;
|
|
}
|
|
|
|
taosMemoryFreeClear(name);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int32_t encodeCreateChildTableForRPC(void* pReqs, int32_t vgId, void** pBuf, int32_t* contLen) {
|
|
int32_t ret = 0;
|
|
|
|
tEncodeSize(tEncodeSVCreateTbBatchReq, pReqs, *contLen, ret);
|
|
if (ret < 0) {
|
|
ret = -1;
|
|
goto end;
|
|
}
|
|
*contLen += sizeof(SMsgHead);
|
|
*pBuf = rpcMallocCont(*contLen);
|
|
if (NULL == *pBuf) {
|
|
ret = -1;
|
|
goto end;
|
|
}
|
|
((SMsgHead*)(*pBuf))->vgId = vgId;
|
|
((SMsgHead*)(*pBuf))->contLen = htonl(*contLen);
|
|
SEncoder coder = {0};
|
|
tEncoderInit(&coder, POINTER_SHIFT(*pBuf, sizeof(SMsgHead)), (*contLen) - sizeof(SMsgHead));
|
|
if (tEncodeSVCreateTbBatchReq(&coder, pReqs) < 0) {
|
|
rpcFreeCont(*pBuf);
|
|
*pBuf = NULL;
|
|
*contLen = 0;
|
|
tEncoderClear(&coder);
|
|
ret = -1;
|
|
goto end;
|
|
}
|
|
tEncoderClear(&coder);
|
|
|
|
end:
|
|
return ret;
|
|
}
|
|
|
|
static int32_t encodeDropChildTableForRPC(void* pReqs, int32_t vgId, void** ppBuf, int32_t *contLen) {
|
|
int32_t code = 0;
|
|
SEncoder ec = {0};
|
|
tEncodeSize(tEncodeSVDropTbBatchReq, pReqs, *contLen, code);
|
|
if (code < 0) {
|
|
code = TSDB_CODE_INVALID_MSG;
|
|
goto end;
|
|
}
|
|
*contLen += sizeof(SMsgHead);
|
|
*ppBuf = rpcMallocCont(*contLen);
|
|
|
|
if (!*ppBuf) {
|
|
code = terrno;
|
|
goto end;
|
|
}
|
|
|
|
((SMsgHead*)(*ppBuf))->vgId = vgId;
|
|
((SMsgHead*)(*ppBuf))->contLen = htonl(*contLen);
|
|
|
|
tEncoderInit(&ec, POINTER_SHIFT(*ppBuf, sizeof(SMsgHead)), (*contLen) - sizeof(SMsgHead));
|
|
code = tEncodeSVDropTbBatchReq(&ec, pReqs);
|
|
tEncoderClear(&ec);
|
|
if (code < 0) {
|
|
rpcFreeCont(*ppBuf);
|
|
*ppBuf = NULL;
|
|
*contLen = 0;
|
|
code = TSDB_CODE_INVALID_MSG;
|
|
goto end;
|
|
}
|
|
end:
|
|
return code;
|
|
}
|
|
|
|
static int32_t tqPutReqToQueue(SVnode* pVnode, void* pReqs, int32_t(*encoder)(void* pReqs, int32_t vgId, void** ppBuf, int32_t *contLen), tmsg_t msgType) {
|
|
void* buf = NULL;
|
|
int32_t tlen = 0;
|
|
|
|
int32_t code = encoder(pReqs, TD_VID(pVnode), &buf, &tlen);
|
|
if (code) {
|
|
tqError("vgId:%d failed to encode create table msg, create table failed, code:%s", TD_VID(pVnode), tstrerror(code));
|
|
return code;
|
|
}
|
|
|
|
SRpcMsg msg = {.msgType = msgType, .pCont = buf, .contLen = tlen};
|
|
code = tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg);
|
|
if (code) {
|
|
tqError("failed to put into write-queue since %s", terrstr());
|
|
}
|
|
|
|
return code;
|
|
}
|
|
|
|
int32_t initCreateTableMsg(SVCreateTbReq* pCreateTableReq, uint64_t suid, const char* stbFullName, int32_t numOfTags) {
|
|
pCreateTableReq->flags = 0;
|
|
pCreateTableReq->type = TSDB_CHILD_TABLE;
|
|
pCreateTableReq->ctb.suid = suid;
|
|
|
|
// set super table name
|
|
SName name = {0};
|
|
|
|
int32_t code = tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
|
|
if (code == 0) {
|
|
pCreateTableReq->ctb.stbName = taosStrdup((char*)tNameGetTableName(&name));
|
|
if (pCreateTableReq->ctb.stbName == NULL) { // ignore this error code
|
|
tqError("failed to duplicate the stb name:%s, failed to init create-table msg and create req table", stbFullName);
|
|
code = terrno;
|
|
}
|
|
}
|
|
|
|
pCreateTableReq->ctb.tagNum = numOfTags;
|
|
return code;
|
|
}
|
|
|
|
int32_t createDefaultTagColName(SArray** pColNameList) {
|
|
*pColNameList = NULL;
|
|
|
|
SArray* pTagColNameList = taosArrayInit(1, TSDB_COL_NAME_LEN);
|
|
if (pTagColNameList == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
char tagNameStr[TSDB_COL_NAME_LEN] = "group_id";
|
|
void* p = taosArrayPush(pTagColNameList, tagNameStr);
|
|
if (p == NULL) {
|
|
taosArrayDestroy(pTagColNameList);
|
|
return terrno;
|
|
}
|
|
|
|
*pColNameList = pTagColNameList;
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
int32_t setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName,
|
|
int64_t gid, bool newSubTableRule, const char* id) {
|
|
if (pDataBlock->info.parTbName[0]) {
|
|
if (newSubTableRule && !isAutoTableName(pDataBlock->info.parTbName) &&
|
|
!alreadyAddGroupId(pDataBlock->info.parTbName, gid) && gid != 0 && stbFullName) {
|
|
pCreateTableReq->name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
|
|
if (pCreateTableReq->name == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
tstrncpy(pCreateTableReq->name, pDataBlock->info.parTbName, TSDB_TABLE_NAME_LEN);
|
|
int32_t code = buildCtbNameAddGroupId(stbFullName, pCreateTableReq->name, gid, TSDB_TABLE_NAME_LEN);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
return code;
|
|
}
|
|
tqDebug("s-task:%s gen name from:%s blockdata", id, pDataBlock->info.parTbName);
|
|
} else {
|
|
pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName);
|
|
if (pCreateTableReq->name == NULL) {
|
|
return terrno;
|
|
}
|
|
tqDebug("s-task:%s copy name:%s from blockdata", id, pDataBlock->info.parTbName);
|
|
}
|
|
} else {
|
|
int32_t code = buildCtbNameByGroupId(stbFullName, gid, &pCreateTableReq->name);
|
|
tqDebug("s-task:%s no name in blockdata, auto-created table name:%s", id, pCreateTableReq->name);
|
|
return code;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int32_t doBuildAndSendCreateTableMsg(SVnode* pVnode, char* stbFullName, SSDataBlock* pDataBlock,
|
|
SStreamTask* pTask, int64_t suid) {
|
|
STSchema* pTSchema = pTask->outputInfo.tbSink.pTSchema;
|
|
int32_t rows = pDataBlock->info.rows;
|
|
SArray* tagArray = NULL;
|
|
const char* id = pTask->id.idStr;
|
|
int32_t vgId = pTask->pMeta->vgId;
|
|
int32_t code = 0;
|
|
STableSinkInfo* pInfo = NULL;
|
|
SVCreateTbBatchReq reqs = {0};
|
|
SArray* crTblArray = NULL;
|
|
|
|
tqDebug("s-task:%s build create %d table(s) msg", id, rows);
|
|
|
|
tagArray = taosArrayInit(4, sizeof(STagVal));
|
|
crTblArray = reqs.pArray = taosArrayInit(1, sizeof(SVCreateTbReq));
|
|
if ((NULL == reqs.pArray) || (tagArray == NULL)) {
|
|
tqError("s-task:%s failed to init create table msg, code:%s", id, tstrerror(terrno));
|
|
code = terrno;
|
|
goto _end;
|
|
}
|
|
|
|
for (int32_t rowId = 0; rowId < rows; rowId++) {
|
|
SVCreateTbReq* pCreateTbReq = &((SVCreateTbReq){0});
|
|
|
|
int32_t size = taosArrayGetSize(pDataBlock->pDataBlock);
|
|
int32_t numOfTags = TMAX(size - UD_TAG_COLUMN_INDEX, 1);
|
|
|
|
code = initCreateTableMsg(pCreateTbReq, suid, stbFullName, numOfTags);
|
|
if (code) {
|
|
tqError("s-task:%s vgId:%d failed to init create table msg", id, vgId);
|
|
continue;
|
|
}
|
|
|
|
taosArrayClear(tagArray);
|
|
|
|
if (size == 2) {
|
|
STagVal tagVal = {
|
|
.cid = pTSchema->numOfCols + 1, .type = TSDB_DATA_TYPE_UBIGINT, .i64 = pDataBlock->info.id.groupId};
|
|
|
|
void* p = taosArrayPush(tagArray, &tagVal);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
code = createDefaultTagColName(&pCreateTbReq->ctb.tagName);
|
|
if (code) {
|
|
return code;
|
|
}
|
|
} else {
|
|
for (int32_t tagId = UD_TAG_COLUMN_INDEX, step = 1; tagId < size; tagId++, step++) {
|
|
SColumnInfoData* pTagData = taosArrayGet(pDataBlock->pDataBlock, tagId);
|
|
if (pTagData == NULL) {
|
|
continue;
|
|
}
|
|
|
|
STagVal tagVal = {.cid = pTSchema->numOfCols + step, .type = pTagData->info.type};
|
|
void* pData = colDataGetData(pTagData, rowId);
|
|
if (colDataIsNull_s(pTagData, rowId)) {
|
|
continue;
|
|
} else if (IS_VAR_DATA_TYPE(pTagData->info.type)) {
|
|
tagVal.nData = varDataLen(pData);
|
|
tagVal.pData = (uint8_t*)varDataVal(pData);
|
|
} else {
|
|
memcpy(&tagVal.i64, pData, pTagData->info.bytes);
|
|
}
|
|
void* p = taosArrayPush(tagArray, &tagVal);
|
|
if (p == NULL) {
|
|
code = terrno;
|
|
goto _end;
|
|
}
|
|
}
|
|
}
|
|
|
|
code = tTagNew(tagArray, 1, false, (STag**)&pCreateTbReq->ctb.pTag);
|
|
taosArrayDestroy(tagArray);
|
|
tagArray = NULL;
|
|
|
|
if (pCreateTbReq->ctb.pTag == NULL || (code != 0)) {
|
|
tdDestroySVCreateTbReq(pCreateTbReq);
|
|
code = TSDB_CODE_OUT_OF_MEMORY;
|
|
goto _end;
|
|
}
|
|
|
|
uint64_t gid = pDataBlock->info.id.groupId;
|
|
if (taosArrayGetSize(pDataBlock->pDataBlock) > UD_GROUPID_COLUMN_INDEX) {
|
|
SColumnInfoData* pGpIdColInfo = taosArrayGet(pDataBlock->pDataBlock, UD_GROUPID_COLUMN_INDEX);
|
|
if (pGpIdColInfo == NULL) {
|
|
continue;
|
|
}
|
|
|
|
// todo remove this
|
|
void* pGpIdData = colDataGetData(pGpIdColInfo, rowId);
|
|
if (gid != *(int64_t*)pGpIdData) {
|
|
tqError("s-task:%s vgId:%d invalid groupId:%" PRId64 " actual:%" PRId64 " in sink task", id, vgId, gid,
|
|
*(int64_t*)pGpIdData);
|
|
}
|
|
}
|
|
|
|
code = setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, gid, IS_NEW_SUBTB_RULE(pTask),
|
|
pTask->id.idStr);
|
|
if (code) {
|
|
goto _end;
|
|
}
|
|
|
|
void* p = taosArrayPush(reqs.pArray, pCreateTbReq);
|
|
if (p == NULL) {
|
|
code = terrno;
|
|
goto _end;
|
|
}
|
|
|
|
bool alreadyCached = doGetSinkTableInfoFromCache(pTask->outputInfo.tbSink.pTbInfo, gid, &pInfo);
|
|
if (!alreadyCached) {
|
|
code = doCreateSinkTableInfo(pCreateTbReq->name, &pInfo);
|
|
if (code) {
|
|
tqError("vgId:%d failed to create sink tableInfo for table:%s, s-task:%s", vgId, pCreateTbReq->name, id);
|
|
continue;
|
|
}
|
|
|
|
code = doPutSinkTableInfoIntoCache(pTask->outputInfo.tbSink.pTbInfo, pInfo, gid, id);
|
|
if (code) {
|
|
tqError("vgId:%d failed to put sink tableInfo:%s into cache, s-task:%s", vgId, pCreateTbReq->name, id);
|
|
}
|
|
}
|
|
|
|
tqDebug("s-task:%s build create table:%s msg complete", id, pCreateTbReq->name);
|
|
}
|
|
|
|
reqs.nReqs = taosArrayGetSize(reqs.pArray);
|
|
code = tqPutReqToQueue(pVnode, &reqs, encodeCreateChildTableForRPC, TDMT_VND_CREATE_TABLE);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
tqError("s-task:%s failed to send create table msg, code:%s", id, tstrerror(code));
|
|
}
|
|
|
|
_end:
|
|
taosArrayDestroy(tagArray);
|
|
taosArrayDestroyEx(crTblArray, (FDelete)tdDestroySVCreateTbReq);
|
|
return code;
|
|
}
|
|
|
|
static int32_t doBuildAndSendDropTableMsg(SVnode* pVnode, char* pStbFullname, SSDataBlock* pDataBlock,
|
|
SStreamTask* pTask, int64_t suid) {
|
|
int32_t lino = 0;
|
|
int32_t code = 0;
|
|
int32_t rows = pDataBlock->info.rows;
|
|
const char* id = pTask->id.idStr;
|
|
SVDropTbBatchReq batchReq = {0};
|
|
SVDropTbReq req = {0};
|
|
|
|
if (rows <= 0 || rows > 1 || pTask->subtableWithoutMd5 == 0) return TSDB_CODE_SUCCESS;
|
|
|
|
batchReq.pArray = taosArrayInit(rows, sizeof(SVDropTbReq));
|
|
if (!batchReq.pArray) return terrno;
|
|
batchReq.nReqs = rows;
|
|
req.suid = suid;
|
|
req.igNotExists = true;
|
|
|
|
SColumnInfoData* pTbNameCol = taosArrayGet(pDataBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
|
|
char tbName[TSDB_TABLE_NAME_LEN + 1] = {0};
|
|
int32_t i = 0;
|
|
void* pData = colDataGetVarData(pTbNameCol, i);
|
|
memcpy(tbName, varDataVal(pData), varDataLen(pData));
|
|
tbName[varDataLen(pData) + 1] = 0;
|
|
req.name = tbName;
|
|
if (taosArrayPush(batchReq.pArray, &req) == NULL) {
|
|
TSDB_CHECK_CODE(terrno, lino, _exit);
|
|
}
|
|
|
|
SMetaReader mr = {0};
|
|
metaReaderDoInit(&mr, pVnode->pMeta, META_READER_LOCK);
|
|
|
|
code = metaGetTableEntryByName(&mr, tbName);
|
|
if (TSDB_CODE_SUCCESS == code && isValidDstChildTable(&mr, TD_VID(pVnode), tbName, pTask->outputInfo.tbSink.stbUid)) {
|
|
STableSinkInfo* pTableSinkInfo = NULL;
|
|
bool alreadyCached = doGetSinkTableInfoFromCache(pTask->outputInfo.tbSink.pTbInfo, pDataBlock->info.id.groupId, &pTableSinkInfo);
|
|
if (alreadyCached) {
|
|
pTableSinkInfo->uid = mr.me.uid;
|
|
}
|
|
}
|
|
metaReaderClear(&mr);
|
|
tqDebug("s-task:%s build drop %d table(s) msg", id, rows);
|
|
code = tqPutReqToQueue(pVnode, &batchReq, encodeDropChildTableForRPC, TDMT_VND_DROP_TABLE);
|
|
TSDB_CHECK_CODE(code, lino, _exit);
|
|
|
|
|
|
code = doWaitForDstTableDropped(pVnode, pTask, tbName);
|
|
TSDB_CHECK_CODE(code, lino, _exit);
|
|
|
|
_exit:
|
|
if (batchReq.pArray) {
|
|
taosArrayDestroy(batchReq.pArray);
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t doBuildAndSendSubmitMsg(SVnode* pVnode, SStreamTask* pTask, SSubmitReq2* pReq, int32_t numOfBlocks) {
|
|
const char* id = pTask->id.idStr;
|
|
int32_t vgId = TD_VID(pVnode);
|
|
int32_t len = 0;
|
|
void* pBuf = NULL;
|
|
int32_t numOfFinalBlocks = taosArrayGetSize(pReq->aSubmitTbData);
|
|
|
|
int32_t code = buildSubmitMsgImpl(pReq, vgId, &pBuf, &len);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
tqError("s-task:%s build submit msg failed, vgId:%d, code:%s", id, vgId, tstrerror(code));
|
|
return code;
|
|
}
|
|
|
|
SRpcMsg msg = {.msgType = TDMT_VND_SUBMIT, .pCont = pBuf, .contLen = len};
|
|
code = tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg);
|
|
if (code == TSDB_CODE_SUCCESS) {
|
|
tqDebug("s-task:%s vgId:%d comp %d blocks into %d and send to dstTable(s) completed", id, vgId, numOfBlocks,
|
|
numOfFinalBlocks);
|
|
} else {
|
|
tqError("s-task:%s failed to put into write-queue since %s", id, terrstr());
|
|
}
|
|
|
|
SSinkRecorder* pRec = &pTask->execInfo.sink;
|
|
|
|
pRec->numOfSubmit += 1;
|
|
if ((pRec->numOfSubmit % 1000) == 0) {
|
|
double el = (taosGetTimestampMs() - pTask->execInfo.readyTs) / 1000.0;
|
|
tqInfo("s-task:%s vgId:%d write %" PRId64 " blocks (%" PRId64 " rows) in %" PRId64
|
|
" submit into dst table, %.2fMiB duration:%.2f Sec.",
|
|
id, vgId, pRec->numOfBlocks, pRec->numOfRows, pRec->numOfSubmit, SIZE_IN_MiB(pRec->dataSize), el);
|
|
}
|
|
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
// merge the new submit table block with the existed blocks
|
|
// if ts in the new data block overlap with existed one, replace it
|
|
int32_t doMergeExistedRows(SSubmitTbData* pExisted, const SSubmitTbData* pNew, const char* id) {
|
|
int32_t oldLen = taosArrayGetSize(pExisted->aRowP);
|
|
int32_t newLen = taosArrayGetSize(pNew->aRowP);
|
|
int32_t numOfPk = 0;
|
|
|
|
int32_t j = 0, k = 0;
|
|
SArray* pFinal = taosArrayInit(oldLen + newLen, POINTER_BYTES);
|
|
if (pFinal == NULL) {
|
|
tqError("s-task:%s failed to prepare merge result datablock, code:%s", id, tstrerror(terrno));
|
|
return terrno;
|
|
}
|
|
|
|
while (j < newLen && k < oldLen) {
|
|
SRow* pNewRow = *(SRow**)TARRAY_GET_ELEM(pNew->aRowP, j);
|
|
SRow* pOldRow = *(SRow**)TARRAY_GET_ELEM(pExisted->aRowP, k);
|
|
|
|
if (pNewRow->ts < pOldRow->ts) {
|
|
void* p = taosArrayPush(pFinal, &pNewRow);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
j += 1;
|
|
} else if (pNewRow->ts > pOldRow->ts) {
|
|
void* p = taosArrayPush(pFinal, &pOldRow);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
k += 1;
|
|
} else {
|
|
// check for the existance of primary key
|
|
if (pNewRow->numOfPKs == 0) {
|
|
void* p = taosArrayPush(pFinal, &pNewRow);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
k += 1;
|
|
j += 1;
|
|
tRowDestroy(pOldRow);
|
|
} else {
|
|
numOfPk = pNewRow->numOfPKs;
|
|
|
|
SRowKey kNew, kOld;
|
|
tRowGetKey(pNewRow, &kNew);
|
|
tRowGetKey(pOldRow, &kOld);
|
|
|
|
int32_t ret = tRowKeyCompare(&kNew, &kOld);
|
|
if (ret <= 0) {
|
|
void* p = taosArrayPush(pFinal, &pNewRow);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
j += 1;
|
|
|
|
if (ret == 0) {
|
|
k += 1;
|
|
tRowDestroy(pOldRow);
|
|
}
|
|
} else {
|
|
void* p = taosArrayPush(pFinal, &pOldRow);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
k += 1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
while (j < newLen) {
|
|
SRow* pRow = *(SRow**)TARRAY_GET_ELEM(pNew->aRowP, j++);
|
|
void* p = taosArrayPush(pFinal, &pRow);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
}
|
|
|
|
while (k < oldLen) {
|
|
SRow* pRow = *(SRow**)TARRAY_GET_ELEM(pExisted->aRowP, k++);
|
|
void* p = taosArrayPush(pFinal, &pRow);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
}
|
|
|
|
taosArrayDestroy(pNew->aRowP);
|
|
taosArrayDestroy(pExisted->aRowP);
|
|
pExisted->aRowP = pFinal;
|
|
|
|
tqTrace("s-task:%s rows merged, final rows:%d, pk:%d uid:%" PRId64 ", existed auto-create table:%d, new-block:%d", id,
|
|
(int32_t)taosArrayGetSize(pFinal), numOfPk, pExisted->uid, (pExisted->pCreateTbReq != NULL),
|
|
(pNew->pCreateTbReq != NULL));
|
|
|
|
tdDestroySVCreateTbReq(pNew->pCreateTbReq);
|
|
taosMemoryFree(pNew->pCreateTbReq);
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
bool isValidDstChildTable(SMetaReader* pReader, int32_t vgId, const char* ctbName, int64_t suid) {
|
|
if (pReader->me.type != TSDB_CHILD_TABLE) {
|
|
tqError("vgId:%d, failed to write into %s, since table type:%d incorrect", vgId, ctbName, pReader->me.type);
|
|
terrno = TSDB_CODE_TDB_INVALID_TABLE_TYPE;
|
|
return false;
|
|
}
|
|
|
|
if (pReader->me.ctbEntry.suid != suid) {
|
|
tqError("vgId:%d, failed to write into %s, since suid mismatch, expect suid:%" PRId64 ", actual:%" PRId64, vgId,
|
|
ctbName, suid, pReader->me.ctbEntry.suid);
|
|
terrno = TSDB_CODE_TDB_TABLE_IN_OTHER_STABLE;
|
|
return false;
|
|
}
|
|
|
|
terrno = 0;
|
|
return true;
|
|
}
|
|
|
|
int32_t buildAutoCreateTableReq(const char* stbFullName, int64_t suid, int32_t numOfCols, SSDataBlock* pDataBlock,
|
|
SArray* pTagArray, bool newSubTableRule, SVCreateTbReq** pReq, const char* id) {
|
|
*pReq = NULL;
|
|
|
|
SVCreateTbReq* pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
|
|
if (pCreateTbReq == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
taosArrayClear(pTagArray);
|
|
int32_t code = initCreateTableMsg(pCreateTbReq, suid, stbFullName, 1);
|
|
if (code != 0) {
|
|
return code;
|
|
}
|
|
|
|
STagVal tagVal = {.cid = numOfCols, .type = TSDB_DATA_TYPE_UBIGINT, .i64 = pDataBlock->info.id.groupId};
|
|
void* p = taosArrayPush(pTagArray, &tagVal);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
code = tTagNew(pTagArray, 1, false, (STag**)&pCreateTbReq->ctb.pTag);
|
|
if (pCreateTbReq->ctb.pTag == NULL || (code != 0)) {
|
|
tdDestroySVCreateTbReq(pCreateTbReq);
|
|
taosMemoryFreeClear(pCreateTbReq);
|
|
return code;
|
|
}
|
|
|
|
code = createDefaultTagColName(&pCreateTbReq->ctb.tagName);
|
|
if (code) {
|
|
return code;
|
|
}
|
|
|
|
// set table name
|
|
code = setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, pDataBlock->info.id.groupId, newSubTableRule,
|
|
id);
|
|
if (code) {
|
|
return code;
|
|
}
|
|
|
|
*pReq = pCreateTbReq;
|
|
return code;
|
|
}
|
|
|
|
int32_t buildSubmitMsgImpl(SSubmitReq2* pSubmitReq, int32_t vgId, void** pMsg, int32_t* msgLen) {
|
|
int32_t code = 0;
|
|
void* pBuf = NULL;
|
|
*msgLen = 0;
|
|
|
|
// encode
|
|
int32_t len = 0;
|
|
tEncodeSize(tEncodeSubmitReq, pSubmitReq, len, code);
|
|
|
|
SEncoder encoder = {0};
|
|
len += sizeof(SSubmitReq2Msg);
|
|
|
|
pBuf = rpcMallocCont(len);
|
|
if (NULL == pBuf) {
|
|
tDestroySubmitReq(pSubmitReq, TSDB_MSG_FLG_ENCODE);
|
|
return terrno;
|
|
}
|
|
|
|
((SSubmitReq2Msg*)pBuf)->header.vgId = vgId;
|
|
((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len);
|
|
((SSubmitReq2Msg*)pBuf)->version = htobe64(1);
|
|
|
|
tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));
|
|
if (tEncodeSubmitReq(&encoder, pSubmitReq) < 0) {
|
|
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
|
tqError("failed to encode submit req, code:%s, ignore and continue", terrstr());
|
|
tEncoderClear(&encoder);
|
|
rpcFreeCont(pBuf);
|
|
tDestroySubmitReq(pSubmitReq, TSDB_MSG_FLG_ENCODE);
|
|
return code;
|
|
}
|
|
|
|
tEncoderClear(&encoder);
|
|
tDestroySubmitReq(pSubmitReq, TSDB_MSG_FLG_ENCODE);
|
|
|
|
*msgLen = len;
|
|
*pMsg = pBuf;
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
int32_t tsAscendingSortFn(const void* p1, const void* p2) {
|
|
SRow* pRow1 = *(SRow**)p1;
|
|
SRow* pRow2 = *(SRow**)p2;
|
|
|
|
if (pRow1->ts == pRow2->ts) {
|
|
return 0;
|
|
} else {
|
|
return pRow1->ts > pRow2->ts ? 1 : -1;
|
|
}
|
|
}
|
|
|
|
int32_t doConvertRows(SSubmitTbData* pTableData, const STSchema* pTSchema, SSDataBlock* pDataBlock, int64_t earlyTs,
|
|
const char* id) {
|
|
int32_t numOfRows = pDataBlock->info.rows;
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
|
|
SArray* pVals = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal));
|
|
pTableData->aRowP = taosArrayInit(numOfRows, sizeof(SRow*));
|
|
|
|
if (pTableData->aRowP == NULL || pVals == NULL) {
|
|
taosArrayDestroy(pTableData->aRowP);
|
|
pTableData->aRowP = NULL;
|
|
taosArrayDestroy(pVals);
|
|
code = terrno;
|
|
tqError("s-task:%s failed to prepare write stream res blocks, code:%s", id, tstrerror(code));
|
|
return code;
|
|
}
|
|
|
|
for (int32_t j = 0; j < numOfRows; j++) {
|
|
taosArrayClear(pVals);
|
|
|
|
int32_t dataIndex = 0;
|
|
int64_t ts = 0;
|
|
|
|
for (int32_t k = 0; k < pTSchema->numOfCols; k++) {
|
|
const STColumn* pCol = &pTSchema->columns[k];
|
|
|
|
// primary timestamp column, for debug purpose
|
|
if (k == 0) {
|
|
SColumnInfoData* pColData = taosArrayGet(pDataBlock->pDataBlock, dataIndex);
|
|
if (pColData == NULL) {
|
|
continue;
|
|
}
|
|
|
|
ts = *(int64_t*)colDataGetData(pColData, j);
|
|
tqTrace("s-task:%s sink row %d, col %d ts %" PRId64, id, j, k, ts);
|
|
|
|
if (ts < earlyTs) {
|
|
tqError("s-task:%s ts:%" PRId64 " of generated results out of valid time range %" PRId64 " , discarded", id,
|
|
ts, earlyTs);
|
|
taosArrayDestroy(pTableData->aRowP);
|
|
pTableData->aRowP = NULL;
|
|
taosArrayDestroy(pVals);
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
}
|
|
|
|
if (IS_SET_NULL(pCol)) {
|
|
if (pCol->flags & COL_IS_KEY) {
|
|
qError("ts:%" PRId64 " primary key column should not be null, colId:%" PRIi16 ", colType:%" PRIi8, ts,
|
|
pCol->colId, pCol->type);
|
|
break;
|
|
}
|
|
SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
|
|
void* p = taosArrayPush(pVals, &cv);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
} else {
|
|
SColumnInfoData* pColData = taosArrayGet(pDataBlock->pDataBlock, dataIndex);
|
|
if (pColData == NULL) {
|
|
continue;
|
|
}
|
|
|
|
if (colDataIsNull_s(pColData, j)) {
|
|
if (pCol->flags & COL_IS_KEY) {
|
|
qError("ts:%" PRId64 " primary key column should not be null, colId:%" PRIi16 ", colType:%" PRIi8,
|
|
ts, pCol->colId, pCol->type);
|
|
break;
|
|
}
|
|
|
|
SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
|
|
void* p = taosArrayPush(pVals, &cv);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
dataIndex++;
|
|
} else {
|
|
void* colData = colDataGetData(pColData, j);
|
|
if (IS_VAR_DATA_TYPE(pCol->type)) { // address copy, no value
|
|
SValue sv =
|
|
(SValue){.type = pCol->type, .nData = varDataLen(colData), .pData = (uint8_t*)varDataVal(colData)};
|
|
SColVal cv = COL_VAL_VALUE(pCol->colId, sv);
|
|
void* p = taosArrayPush(pVals, &cv);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
} else {
|
|
SValue sv = {.type = pCol->type};
|
|
valueSetDatum(&sv, pCol->type, colData, tDataTypes[pCol->type].bytes);
|
|
SColVal cv = COL_VAL_VALUE(pCol->colId, sv);
|
|
void* p = taosArrayPush(pVals, &cv);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
}
|
|
dataIndex++;
|
|
}
|
|
}
|
|
}
|
|
|
|
SRow* pRow = NULL;
|
|
code = tRowBuild(pVals, (STSchema*)pTSchema, &pRow);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
tDestroySubmitTbData(pTableData, TSDB_MSG_FLG_ENCODE);
|
|
taosArrayDestroy(pVals);
|
|
tqError("s-task:%s build rows for submit failed, ts:%"PRId64, id, ts);
|
|
return code;
|
|
}
|
|
|
|
void* p = taosArrayPush(pTableData->aRowP, &pRow);
|
|
if (p == NULL) {
|
|
return terrno;
|
|
}
|
|
}
|
|
|
|
taosArrayDestroy(pVals);
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
int32_t doWaitForDstTableCreated(SVnode* pVnode, SStreamTask* pTask, STableSinkInfo* pTableSinkInfo,
|
|
const char* dstTableName, int64_t* uid) {
|
|
int32_t vgId = TD_VID(pVnode);
|
|
int64_t suid = pTask->outputInfo.tbSink.stbUid;
|
|
const char* id = pTask->id.idStr;
|
|
int32_t timeout = 60; // 1min
|
|
int64_t start = taosGetTimestampSec();
|
|
|
|
while (pTableSinkInfo->uid == 0) {
|
|
if (streamTaskShouldStop(pTask)) {
|
|
tqDebug("s-task:%s task will stop, quit from waiting for table:%s create", id, dstTableName);
|
|
return TSDB_CODE_STREAM_EXEC_CANCELLED;
|
|
}
|
|
|
|
int64_t waitingDuration = taosGetTimestampSec() - start;
|
|
if (waitingDuration > timeout) {
|
|
tqError("s-task:%s wait for table-creating:%s more than %dsec, failed", id, dstTableName, timeout);
|
|
return TSDB_CODE_PAR_TABLE_NOT_EXIST;
|
|
}
|
|
|
|
// wait for the table to be created
|
|
SMetaReader mr = {0};
|
|
metaReaderDoInit(&mr, pVnode->pMeta, META_READER_LOCK);
|
|
|
|
int32_t code = metaGetTableEntryByName(&mr, dstTableName);
|
|
if (code == 0) { // table already exists, check its type and uid
|
|
bool isValid = isValidDstChildTable(&mr, vgId, dstTableName, suid);
|
|
if (isValid) { // not valid table, ignore it
|
|
tqDebug("s-task:%s set uid:%" PRIu64 " for dstTable:%s from meta", id, mr.me.uid, pTableSinkInfo->name.data);
|
|
// set the destination table uid
|
|
(*uid) = mr.me.uid;
|
|
pTableSinkInfo->uid = mr.me.uid;
|
|
}
|
|
|
|
metaReaderClear(&mr);
|
|
return terrno;
|
|
} else { // not exist, wait and retry
|
|
metaReaderClear(&mr);
|
|
taosMsleep(100);
|
|
tqDebug("s-task:%s wait 100ms for the table:%s ready before insert data", id, dstTableName);
|
|
}
|
|
}
|
|
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
static int32_t doWaitForDstTableDropped(SVnode* pVnode, SStreamTask* pTask, const char* dstTableName) {
|
|
int32_t vgId = TD_VID(pVnode);
|
|
int64_t suid = pTask->outputInfo.tbSink.stbUid;
|
|
const char* id = pTask->id.idStr;
|
|
|
|
while (1) {
|
|
if (streamTaskShouldStop(pTask)) {
|
|
tqDebug("s-task:%s task will stop, quit from waiting for table:%s drop", id, dstTableName);
|
|
return TSDB_CODE_STREAM_EXEC_CANCELLED;
|
|
}
|
|
SMetaReader mr = {0};
|
|
metaReaderDoInit(&mr, pVnode->pMeta, META_READER_LOCK);
|
|
int32_t code = metaGetTableEntryByName(&mr, dstTableName);
|
|
if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
|
|
metaReaderClear(&mr);
|
|
break;
|
|
} else if (TSDB_CODE_SUCCESS == code) {
|
|
if (isValidDstChildTable(&mr, vgId, dstTableName, suid)) {
|
|
metaReaderClear(&mr);
|
|
taosMsleep(100);
|
|
tqDebug("s-task:%s wait 100ms for table:%s drop", id, dstTableName);
|
|
} else {
|
|
metaReaderClear(&mr);
|
|
break;
|
|
}
|
|
} else {
|
|
tqError("s-task:%s failed to wait for table:%s drop", id, dstTableName);
|
|
metaReaderClear(&mr);
|
|
return terrno;
|
|
}
|
|
}
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
int32_t doCreateSinkTableInfo(const char* pDstTableName, STableSinkInfo** pInfo) {
|
|
int32_t nameLen = strlen(pDstTableName);
|
|
(*pInfo) = taosMemoryCalloc(1, sizeof(STableSinkInfo) + nameLen + 1);
|
|
if (*pInfo == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
(*pInfo)->name.len = nameLen;
|
|
memcpy((*pInfo)->name.data, pDstTableName, nameLen);
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDataBlock, char* stbFullName,
|
|
SSubmitTbData* pTableData) {
|
|
uint64_t groupId = pDataBlock->info.id.groupId;
|
|
char* dstTableName = pDataBlock->info.parTbName;
|
|
int32_t numOfRows = pDataBlock->info.rows;
|
|
const char* id = pTask->id.idStr;
|
|
int64_t suid = pTask->outputInfo.tbSink.stbUid;
|
|
STSchema* pTSchema = pTask->outputInfo.tbSink.pTSchema;
|
|
int32_t vgId = TD_VID(pVnode);
|
|
STableSinkInfo* pTableSinkInfo = NULL;
|
|
int32_t code = 0;
|
|
|
|
bool alreadyCached = doGetSinkTableInfoFromCache(pTask->outputInfo.tbSink.pTbInfo, groupId, &pTableSinkInfo);
|
|
|
|
if (alreadyCached) {
|
|
if (dstTableName[0] == 0) { // data block does not set the destination table name
|
|
tstrncpy(dstTableName, pTableSinkInfo->name.data, pTableSinkInfo->name.len + 1);
|
|
tqDebug("s-task:%s vgId:%d, gropuId:%" PRIu64 " datablock table name is null, set name:%s", id, vgId, groupId,
|
|
dstTableName);
|
|
} else {
|
|
tstrncpy(dstTableName, pTableSinkInfo->name.data, pTableSinkInfo->name.len + 1);
|
|
if (pTableSinkInfo->uid != 0) {
|
|
tqDebug("s-task:%s write %d rows into groupId:%" PRIu64 " dstTable:%s(uid:%" PRIu64 ")", id, numOfRows, groupId,
|
|
dstTableName, pTableSinkInfo->uid);
|
|
} else {
|
|
tqDebug("s-task:%s write %d rows into groupId:%" PRIu64 " dstTable:%s(not set uid yet for the secondary block)",
|
|
id, numOfRows, groupId, dstTableName);
|
|
}
|
|
}
|
|
} else { // this groupId has not been kept in cache yet
|
|
if (dstTableName[0] == 0) {
|
|
memset(dstTableName, 0, TSDB_TABLE_NAME_LEN);
|
|
code = buildCtbNameByGroupIdImpl(stbFullName, groupId, dstTableName);
|
|
if (code) {
|
|
tqDebug("s-task:%s failed to build auto create table-name:%s, groupId:0x%" PRId64, id, dstTableName, groupId);
|
|
return code;
|
|
} else {
|
|
tqDebug("s-task:%s no table name given, generated sub-table-name:%s, groupId:0x%" PRId64, id, dstTableName, groupId);
|
|
}
|
|
} else {
|
|
if (pTask->subtableWithoutMd5 != 1 && !isAutoTableName(dstTableName) &&
|
|
!alreadyAddGroupId(dstTableName, groupId) && groupId != 0) {
|
|
tqDebug("s-task:%s append groupId:%" PRId64 " for generated dstTable:%s", id, groupId, dstTableName);
|
|
if (pTask->ver == SSTREAM_TASK_SUBTABLE_CHANGED_VER) {
|
|
code = buildCtbNameAddGroupId(NULL, dstTableName, groupId, sizeof(pDataBlock->info.parTbName));
|
|
} else if (pTask->ver >= SSTREAM_TASK_APPEND_STABLE_NAME_VER && stbFullName) {
|
|
code = buildCtbNameAddGroupId(stbFullName, dstTableName, groupId, sizeof(pDataBlock->info.parTbName));
|
|
}
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
return code;
|
|
}
|
|
}
|
|
}
|
|
|
|
code = doCreateSinkTableInfo(dstTableName, &pTableSinkInfo);
|
|
if (code == 0) {
|
|
tqDebug("s-task:%s build new sinkTableInfo to add cache, dstTable:%s", id, dstTableName);
|
|
} else {
|
|
tqDebug("s-task:%s failed to build new sinkTableInfo, dstTable:%s", id, dstTableName);
|
|
return code;
|
|
}
|
|
}
|
|
|
|
if (alreadyCached) {
|
|
pTableData->uid = pTableSinkInfo->uid;
|
|
|
|
if (pTableData->uid == 0) {
|
|
tqTrace("s-task:%s cached tableInfo:%s uid is invalid, acquire it from meta", id, pTableSinkInfo->name.data);
|
|
return doWaitForDstTableCreated(pVnode, pTask, pTableSinkInfo, dstTableName, &pTableData->uid);
|
|
} else {
|
|
tqTrace("s-task:%s set the dstTable uid from cache:%" PRId64, id, pTableData->uid);
|
|
}
|
|
} else {
|
|
// The auto-create option will always set to be open for those submit messages, which arrives during the period
|
|
// the creating of the destination table, due to the absence of the user-specified table in TSDB. When scanning
|
|
// data from WAL, those submit messages, with auto-created table option, will be discarded expect the first, for
|
|
// those mismatched table uids. Only the FIRST table has the correct table uid, and those remain all have
|
|
// randomly generated, but false table uid in the WAL.
|
|
SMetaReader mr = {0};
|
|
metaReaderDoInit(&mr, pVnode->pMeta, META_READER_LOCK);
|
|
|
|
// table not in cache, let's try to extract it from tsdb meta
|
|
if (metaGetTableEntryByName(&mr, dstTableName) < 0) {
|
|
metaReaderClear(&mr);
|
|
|
|
if (pTask->outputInfo.tbSink.autoCreateCtb) {
|
|
tqDebug("s-task:%s stream write into table:%s, table auto created", id, dstTableName);
|
|
|
|
SArray* pTagArray = taosArrayInit(pTSchema->numOfCols + 1, sizeof(STagVal));
|
|
if (pTagArray == NULL) {
|
|
tqError("s-task:%s failed to build auto create submit msg in sink, vgId:%d, due to %s", id, vgId,
|
|
tstrerror(terrno));
|
|
return terrno;
|
|
}
|
|
|
|
pTableData->flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
|
|
code = buildAutoCreateTableReq(stbFullName, suid, pTSchema->numOfCols + 1, pDataBlock, pTagArray,
|
|
IS_NEW_SUBTB_RULE(pTask), &pTableData->pCreateTbReq, id);
|
|
taosArrayDestroy(pTagArray);
|
|
|
|
if (code) {
|
|
tqError("s-task:%s failed to build auto create dst-table req:%s, code:%s", id, dstTableName, tstrerror(code));
|
|
taosMemoryFree(pTableSinkInfo);
|
|
return code;
|
|
}
|
|
|
|
pTableSinkInfo->uid = 0;
|
|
code = doPutSinkTableInfoIntoCache(pTask->outputInfo.tbSink.pTbInfo, pTableSinkInfo, groupId, id);
|
|
} else {
|
|
tqError("s-task:%s vgId:%d dst-table:%s not auto-created, and not create in tsdb, discard data", id, vgId,
|
|
dstTableName);
|
|
return TSDB_CODE_TDB_TABLE_NOT_EXIST;
|
|
}
|
|
} else {
|
|
bool isValid = isValidDstChildTable(&mr, vgId, dstTableName, suid);
|
|
if (!isValid) {
|
|
metaReaderClear(&mr);
|
|
taosMemoryFree(pTableSinkInfo);
|
|
tqError("s-task:%s vgId:%d table:%s already exists, but not child table, stream results is discarded", id, vgId,
|
|
dstTableName);
|
|
return terrno;
|
|
} else {
|
|
pTableData->uid = mr.me.uid;
|
|
pTableSinkInfo->uid = mr.me.uid;
|
|
|
|
metaReaderClear(&mr);
|
|
code = doPutSinkTableInfoIntoCache(pTask->outputInfo.tbSink.pTbInfo, pTableSinkInfo, groupId, id);
|
|
}
|
|
}
|
|
}
|
|
|
|
return code;
|
|
}
|
|
|
|
int32_t tqSetDstTableDataPayload(uint64_t suid, const STSchema* pTSchema, int32_t blockIndex, SSDataBlock* pDataBlock,
|
|
SSubmitTbData* pTableData, int64_t earlyTs, const char* id) {
|
|
int32_t numOfRows = pDataBlock->info.rows;
|
|
char* dstTableName = pDataBlock->info.parTbName;
|
|
|
|
tqDebug("s-task:%s sink data pipeline, build submit msg from %dth resBlock, including %d rows, dst suid:%" PRId64, id,
|
|
blockIndex + 1, numOfRows, suid);
|
|
|
|
// convert all rows
|
|
int32_t code = doConvertRows(pTableData, pTSchema, pDataBlock, earlyTs, id);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
tqError("s-task:%s failed to convert rows from result block, code:%s", id, tstrerror(terrno));
|
|
return code;
|
|
}
|
|
|
|
if (pTableData->aRowP != NULL) {
|
|
taosArraySort(pTableData->aRowP, tsAscendingSortFn);
|
|
tqTrace("s-task:%s build submit msg for dstTable:%s, numOfRows:%d", id, dstTableName, numOfRows);
|
|
}
|
|
|
|
return code;
|
|
}
|
|
|
|
int32_t checkTagSchema(SStreamTask* pTask, SVnode* pVnode) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
const char* id = pTask->id.idStr;
|
|
STaskOutputInfo* pOutputInfo = &pTask->outputInfo;
|
|
int32_t vgId = pTask->pMeta->vgId;
|
|
|
|
if (pTask->outputInfo.tbSink.pTagSchema == NULL) {
|
|
SMetaReader mer1 = {0};
|
|
metaReaderDoInit(&mer1, pVnode->pMeta, META_READER_LOCK);
|
|
|
|
code = metaReaderGetTableEntryByUid(&mer1, pOutputInfo->tbSink.stbUid);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
tqError("s-task:%s vgId:%d failed to get the dst stable, failed to sink results", id, vgId);
|
|
metaReaderClear(&mer1);
|
|
return code;
|
|
}
|
|
|
|
pOutputInfo->tbSink.pTagSchema = tCloneSSchemaWrapper(&mer1.me.stbEntry.schemaTag);
|
|
metaReaderClear(&mer1);
|
|
|
|
if (pOutputInfo->tbSink.pTagSchema == NULL) {
|
|
tqError("s-task:%s failed to clone tag schema, code:%s, failed to sink results", id, tstrerror(terrno));
|
|
return terrno;
|
|
}
|
|
|
|
SSchemaWrapper* pTagSchema = pOutputInfo->tbSink.pTagSchema;
|
|
SSchema* pCol1 = &pTagSchema->pSchema[0];
|
|
if (pTagSchema->nCols == 1 && pCol1->type == TSDB_DATA_TYPE_UBIGINT && strcmp(pCol1->name, "group_id") == 0) {
|
|
pOutputInfo->tbSink.autoCreateCtb = true;
|
|
} else {
|
|
pOutputInfo->tbSink.autoCreateCtb = false;
|
|
}
|
|
}
|
|
|
|
return code;
|
|
}
|
|
|
|
void tqSinkDataIntoDstTable(SStreamTask* pTask, void* vnode, void* data) {
|
|
const SArray* pBlocks = (const SArray*)data;
|
|
SVnode* pVnode = (SVnode*)vnode;
|
|
int64_t suid = pTask->outputInfo.tbSink.stbUid;
|
|
char* stbFullName = pTask->outputInfo.tbSink.stbFullName;
|
|
STSchema* pTSchema = pTask->outputInfo.tbSink.pTSchema;
|
|
int32_t vgId = TD_VID(pVnode);
|
|
int32_t numOfBlocks = taosArrayGetSize(pBlocks);
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
const char* id = pTask->id.idStr;
|
|
int64_t earlyTs = tsdbGetEarliestTs(pVnode->pTsdb);
|
|
STaskOutputInfo* pOutputInfo = &pTask->outputInfo;
|
|
|
|
code = checkTagSchema(pTask, pVnode);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
return;
|
|
}
|
|
|
|
code = tqSendAllNotifyEvents(pBlocks, pTask, pVnode);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
tqError("vgId: %d, s-task:%s failed to send all event notifications", vgId, id);
|
|
// continue processing even if notification fails
|
|
}
|
|
|
|
bool onlySubmitData = hasOnlySubmitData(pBlocks, numOfBlocks);
|
|
if (!onlySubmitData || pTask->subtableWithoutMd5 == 1) {
|
|
tqDebug("vgId:%d, s-task:%s write %d stream resBlock(s) into table, has other type block, submit one-by-one", vgId,
|
|
id, numOfBlocks);
|
|
|
|
for (int32_t i = 0; i < numOfBlocks; ++i) {
|
|
if (streamTaskShouldStop(pTask)) {
|
|
return;
|
|
}
|
|
|
|
SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
|
|
if (pDataBlock == NULL) {
|
|
continue;
|
|
}
|
|
|
|
if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
|
|
code = doBuildAndSendDeleteMsg(pVnode, stbFullName, pDataBlock, pTask, suid);
|
|
} else if (pDataBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
|
|
code = doBuildAndSendCreateTableMsg(pVnode, stbFullName, pDataBlock, pTask, suid);
|
|
} else if (pDataBlock->info.type == STREAM_CHECKPOINT) {
|
|
continue;
|
|
} else if (pDataBlock->info.type == STREAM_DROP_CHILD_TABLE && pTask->subtableWithoutMd5) {
|
|
code = doBuildAndSendDropTableMsg(pVnode, stbFullName, pDataBlock, pTask, suid);
|
|
} else if (pDataBlock->info.type == STREAM_NOTIFY_EVENT) {
|
|
continue;
|
|
} else {
|
|
code = handleResultBlockMsg(pTask, pDataBlock, i, pVnode, earlyTs);
|
|
}
|
|
}
|
|
} else {
|
|
tqDebug("vgId:%d, s-task:%s write %d stream resBlock(s) into table, merge submit msg", vgId, id, numOfBlocks);
|
|
if (streamTaskShouldStop(pTask)) {
|
|
return;
|
|
}
|
|
|
|
rebuildAndSendMultiResBlock(pTask, pBlocks, pVnode, earlyTs);
|
|
}
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
|
bool hasOnlySubmitData(const SArray* pBlocks, int32_t numOfBlocks) {
|
|
for (int32_t i = 0; i < numOfBlocks; ++i) {
|
|
SSDataBlock* p = taosArrayGet(pBlocks, i);
|
|
if (p == NULL) {
|
|
continue;
|
|
}
|
|
|
|
if (p->info.type == STREAM_DELETE_RESULT || p->info.type == STREAM_CREATE_CHILD_TABLE) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
int32_t doPutSinkTableInfoIntoCache(SSHashObj* pSinkTableMap, STableSinkInfo* pTableSinkInfo, uint64_t groupId, const char* id) {
|
|
int32_t code = tSimpleHashPut(pSinkTableMap, &groupId, sizeof(uint64_t), &pTableSinkInfo, POINTER_BYTES);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
taosMemoryFreeClear(pTableSinkInfo);
|
|
} else {
|
|
tqDebug("s-task:%s new dst table:%s(uid:%" PRIu64 ") added into cache, total:%d", id, pTableSinkInfo->name.data,
|
|
pTableSinkInfo->uid, tSimpleHashGetSize(pSinkTableMap));
|
|
}
|
|
|
|
return code;
|
|
}
|
|
|
|
bool doGetSinkTableInfoFromCache(SSHashObj* pTableInfoMap, uint64_t groupId, STableSinkInfo** pInfo) {
|
|
void* pVal = tSimpleHashGet(pTableInfoMap, &groupId, sizeof(uint64_t));
|
|
if (pVal) {
|
|
*pInfo = *(STableSinkInfo**)pVal;
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
int32_t doRemoveSinkTableInfoInCache(SSHashObj* pSinkTableMap, uint64_t groupId, const char* id) {
|
|
if (tSimpleHashGetSize(pSinkTableMap) == 0) {
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
int32_t code = tSimpleHashRemove(pSinkTableMap, &groupId, sizeof(groupId));
|
|
if (code == 0) {
|
|
tqDebug("s-task:%s remove cached table meta for groupId:%" PRId64, id, groupId);
|
|
} else {
|
|
tqError("s-task:%s failed to remove table meta from hashmap, groupId:%" PRId64, id, groupId);
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t doBuildAndSendDeleteMsg(SVnode* pVnode, char* stbFullName, SSDataBlock* pDataBlock, SStreamTask* pTask,
|
|
int64_t suid) {
|
|
SBatchDeleteReq deleteReq = {.suid = suid, .deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq))};
|
|
if (deleteReq.deleteReqs == NULL) {
|
|
return terrno;
|
|
}
|
|
|
|
int32_t code =
|
|
tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, &deleteReq, pTask->id.idStr, IS_NEW_SUBTB_RULE(pTask));
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
return code;
|
|
}
|
|
|
|
if (taosArrayGetSize(deleteReq.deleteReqs) == 0) {
|
|
taosArrayDestroy(deleteReq.deleteReqs);
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
int32_t len;
|
|
tEncodeSize(tEncodeSBatchDeleteReq, &deleteReq, len, code);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("s-task:%s failed to encode delete request", pTask->id.idStr);
|
|
return code;
|
|
}
|
|
|
|
SEncoder encoder = {0};
|
|
void* serializedDeleteReq = rpcMallocCont(len + sizeof(SMsgHead));
|
|
void* abuf = POINTER_SHIFT(serializedDeleteReq, sizeof(SMsgHead));
|
|
tEncoderInit(&encoder, abuf, len);
|
|
code = tEncodeSBatchDeleteReq(&encoder, &deleteReq);
|
|
tEncoderClear(&encoder);
|
|
taosArrayDestroy(deleteReq.deleteReqs);
|
|
|
|
if (code) {
|
|
return code;
|
|
}
|
|
|
|
((SMsgHead*)serializedDeleteReq)->vgId = TD_VID(pVnode);
|
|
|
|
SRpcMsg msg = {.msgType = TDMT_VND_BATCH_DEL, .pCont = serializedDeleteReq, .contLen = len + sizeof(SMsgHead)};
|
|
if (tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) != 0) {
|
|
tqDebug("failed to put delete req into write-queue since %s", terrstr());
|
|
}
|
|
|
|
return TSDB_CODE_SUCCESS;
|
|
}
|
|
|
|
void rebuildAndSendMultiResBlock(SStreamTask* pTask, const SArray* pBlocks, SVnode* pVnode, int64_t earlyTs) {
|
|
int32_t code = 0;
|
|
const char* id = pTask->id.idStr;
|
|
int32_t vgId = pTask->pMeta->vgId;
|
|
int32_t numOfBlocks = taosArrayGetSize(pBlocks);
|
|
int64_t suid = pTask->outputInfo.tbSink.stbUid;
|
|
STSchema* pTSchema = pTask->outputInfo.tbSink.pTSchema;
|
|
char* stbFullName = pTask->outputInfo.tbSink.stbFullName;
|
|
|
|
SHashObj* pTableIndexMap =
|
|
taosHashInit(numOfBlocks, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
|
|
|
|
SSubmitReq2 submitReq = {.aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData))};
|
|
if (submitReq.aSubmitTbData == NULL) {
|
|
code = terrno;
|
|
tqError("s-task:%s vgId:%d failed to prepare submit msg in sink task, code:%s", id, vgId, tstrerror(code));
|
|
taosHashCleanup(pTableIndexMap);
|
|
return;
|
|
}
|
|
|
|
bool hasSubmit = false;
|
|
for (int32_t i = 0; i < numOfBlocks; i++) {
|
|
SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
|
|
if (pDataBlock == NULL) {
|
|
continue;
|
|
}
|
|
|
|
if (pDataBlock->info.type == STREAM_CHECKPOINT) {
|
|
continue;
|
|
}
|
|
|
|
if (pDataBlock->info.type == STREAM_NOTIFY_EVENT) {
|
|
continue;
|
|
}
|
|
|
|
hasSubmit = true;
|
|
pTask->execInfo.sink.numOfBlocks += 1;
|
|
uint64_t groupId = pDataBlock->info.id.groupId;
|
|
|
|
SSubmitTbData tbData = {.suid = suid, .uid = 0, .sver = pTSchema->version, .flags = TD_REQ_FROM_APP};
|
|
|
|
int32_t* index = taosHashGet(pTableIndexMap, &groupId, sizeof(groupId));
|
|
if (index == NULL) { // no data yet, append it
|
|
code = setDstTableDataUid(pVnode, pTask, pDataBlock, stbFullName, &tbData);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
tqError("vgId:%d dst-table gid:%" PRId64 " not exist, discard stream results", vgId, groupId);
|
|
continue;
|
|
}
|
|
|
|
code = tqSetDstTableDataPayload(suid, pTSchema, i, pDataBlock, &tbData, earlyTs, id);
|
|
if (code != TSDB_CODE_SUCCESS || tbData.aRowP == NULL) {
|
|
if (tbData.pCreateTbReq != NULL) {
|
|
tdDestroySVCreateTbReq(tbData.pCreateTbReq);
|
|
(void)doRemoveSinkTableInfoInCache(pTask->outputInfo.tbSink.pTbInfo, groupId, id);
|
|
tbData.pCreateTbReq = NULL;
|
|
}
|
|
continue;
|
|
}
|
|
|
|
void* p = taosArrayPush(submitReq.aSubmitTbData, &tbData);
|
|
if (p == NULL) {
|
|
tqError("vgId:%d, s-task:%s failed to build submit msg, data lost", vgId, id);
|
|
continue;
|
|
}
|
|
|
|
int32_t size = (int32_t)taosArrayGetSize(submitReq.aSubmitTbData) - 1;
|
|
code = taosHashPut(pTableIndexMap, &groupId, sizeof(groupId), &size, sizeof(size));
|
|
if (code) {
|
|
tqError("vgId:%d, s-task:%s failed to put group into index map, code:%s", vgId, id, tstrerror(code));
|
|
continue;
|
|
}
|
|
} else {
|
|
code = tqSetDstTableDataPayload(suid, pTSchema, i, pDataBlock, &tbData, earlyTs, id);
|
|
if (code != TSDB_CODE_SUCCESS || tbData.aRowP == NULL) {
|
|
if (tbData.pCreateTbReq != NULL) {
|
|
tdDestroySVCreateTbReq(tbData.pCreateTbReq);
|
|
tbData.pCreateTbReq = NULL;
|
|
}
|
|
continue;
|
|
}
|
|
|
|
SSubmitTbData* pExisted = taosArrayGet(submitReq.aSubmitTbData, *index);
|
|
if (pExisted == NULL) {
|
|
continue;
|
|
}
|
|
|
|
code = doMergeExistedRows(pExisted, &tbData, id);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
continue;
|
|
}
|
|
}
|
|
|
|
pTask->execInfo.sink.numOfRows += pDataBlock->info.rows;
|
|
}
|
|
|
|
taosHashCleanup(pTableIndexMap);
|
|
|
|
if (hasSubmit) {
|
|
code = doBuildAndSendSubmitMsg(pVnode, pTask, &submitReq, numOfBlocks);
|
|
if (code) { // failed and continue
|
|
tqError("vgId:%d failed to build and send submit msg", vgId);
|
|
}
|
|
} else {
|
|
tDestroySubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE);
|
|
tqDebug("vgId:%d, s-task:%s write results completed", vgId, id);
|
|
}
|
|
}
|
|
|
|
int32_t handleResultBlockMsg(SStreamTask* pTask, SSDataBlock* pDataBlock, int32_t index, SVnode* pVnode, int64_t earlyTs) {
|
|
int32_t code = 0;
|
|
STSchema* pTSchema = pTask->outputInfo.tbSink.pTSchema;
|
|
int64_t suid = pTask->outputInfo.tbSink.stbUid;
|
|
const char* id = pTask->id.idStr;
|
|
int32_t vgId = TD_VID(pVnode);
|
|
char* stbFullName = pTask->outputInfo.tbSink.stbFullName;
|
|
|
|
pTask->execInfo.sink.numOfBlocks += 1;
|
|
|
|
SSubmitReq2 submitReq = {.aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData))};
|
|
if (submitReq.aSubmitTbData == NULL) {
|
|
tqError("s-task:%s vgId:%d failed to prepare submit msg in sink task, code:%s", id, vgId, tstrerror(terrno));
|
|
return terrno;
|
|
}
|
|
|
|
SSubmitTbData tbData = {.suid = suid, .uid = 0, .sver = pTSchema->version, .flags = TD_REQ_FROM_APP};
|
|
code = setDstTableDataUid(pVnode, pTask, pDataBlock, stbFullName, &tbData);
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
tqError("vgId:%d s-task:%s dst-table not exist, stb:%s discard stream results", vgId, id, stbFullName);
|
|
tDestroySubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE);
|
|
return code;
|
|
}
|
|
|
|
code = tqSetDstTableDataPayload(suid, pTSchema, index, pDataBlock, &tbData, earlyTs, id);
|
|
if (code != TSDB_CODE_SUCCESS || tbData.aRowP == NULL) {
|
|
if (tbData.pCreateTbReq != NULL) {
|
|
tdDestroySVCreateTbReq(tbData.pCreateTbReq);
|
|
(void)doRemoveSinkTableInfoInCache(pTask->outputInfo.tbSink.pTbInfo, pDataBlock->info.id.groupId, id);
|
|
tbData.pCreateTbReq = NULL;
|
|
}
|
|
|
|
tDestroySubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE);
|
|
return code;
|
|
}
|
|
|
|
void* p = taosArrayPush(submitReq.aSubmitTbData, &tbData);
|
|
if (p == NULL) {
|
|
tqDebug("vgId:%d, s-task:%s failed to build submit msg, code:%s, data lost", vgId, id, tstrerror(terrno));
|
|
tDestroySubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE);
|
|
return terrno;
|
|
}
|
|
|
|
code = doBuildAndSendSubmitMsg(pVnode, pTask, &submitReq, 1);
|
|
if (code) { // failed and continue
|
|
tqDebug("vgId:%d, s-task:%s submit msg failed, code:%s data lost", vgId, id, tstrerror(code));
|
|
}
|
|
|
|
return code;
|
|
}
|