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

refactor: do some internal refactor.

This commit is contained in:
Haojun Liao 2023-05-04 16:15:14 +08:00
parent 11dccf12de
commit 09a04052ec
30 changed files with 305 additions and 276 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
include/common/tcommon.h
View File

@ -212,14 +212,6 @@ enum {
FETCH_TYPE__NONE, FETCH_TYPE__NONE,
}; };
typedef struct {
int8_t fetchType;
union {
SSDataBlock data;
void* meta;
};
} SFetchRet;
typedef struct SVarColAttr { typedef struct SVarColAttr {
int32_t* offset; // start position for each entry in the list int32_t* offset; // start position for each entry in the list
uint32_t length; // used buffer size that contain the valid data uint32_t length; // used buffer size that contain the valid data

10
include/common/tmsg.h
View File

@ -415,7 +415,7 @@ static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* p
return pSW; return pSW;
} }
static FORCE_INLINE void tDeleteSSchemaWrapper(SSchemaWrapper* pSchemaWrapper) { static FORCE_INLINE void tDeleteSchemaWrapper(SSchemaWrapper* pSchemaWrapper) {
if (pSchemaWrapper) { if (pSchemaWrapper) {
taosMemoryFree(pSchemaWrapper->pSchema); taosMemoryFree(pSchemaWrapper->pSchema);
taosMemoryFree(pSchemaWrapper); taosMemoryFree(pSchemaWrapper);
@ -3421,10 +3421,10 @@ typedef struct {
char data[]; // SSubmitReq2 char data[]; // SSubmitReq2
} SSubmitReq2Msg; } SSubmitReq2Msg;
int32_t tEncodeSSubmitReq2(SEncoder* pCoder, const SSubmitReq2* pReq); int32_t tEncodeSubmitReq(SEncoder* pCoder, const SSubmitReq2* pReq);
int32_t tDecodeSSubmitReq2(SDecoder* pCoder, SSubmitReq2* pReq); int32_t tDecodeSubmitReq(SDecoder* pCoder, SSubmitReq2* pReq);
void tDestroySSubmitTbData(SSubmitTbData* pTbData, int32_t flag); void tDestroySubmitTbData(SSubmitTbData* pTbData, int32_t flag);
void tDestroySSubmitReq(SSubmitReq2* pReq, int32_t flag); void tDestroySubmitReq(SSubmitReq2* pReq, int32_t flag);
typedef struct { typedef struct {
int32_t affectedRows; int32_t affectedRows;

2
include/libs/executor/executor.h
View File

@ -192,8 +192,6 @@ SArray* qGetQueriedTableListInfo(qTaskInfo_t tinfo);
int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType); int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType);
int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit);
void qStreamSetOpen(qTaskInfo_t tinfo); void qStreamSetOpen(qTaskInfo_t tinfo);
void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset); void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);

2
include/libs/wal/wal.h
View File

@ -147,8 +147,6 @@ typedef struct SWalReader {
int64_t curFileFirstVer; int64_t curFileFirstVer;
int64_t curVersion; int64_t curVersion;
int64_t capacity; int64_t capacity;
// int8_t curInvalid;
// int8_t curStopped;
TdThreadMutex mutex; TdThreadMutex mutex;
SWalFilterCond cond; SWalFilterCond cond;
// TODO remove it // TODO remove it

4
source/client/src/clientMain.c
View File

@ -191,7 +191,7 @@ void taos_free_result(TAOS_RES *res) {
taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree); taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
taosArrayDestroy(pRsp->rsp.blockDataLen); taosArrayDestroy(pRsp->rsp.blockDataLen);
taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree); taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper); taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSchemaWrapper);
// taosx // taosx
taosArrayDestroy(pRsp->rsp.createTableLen); taosArrayDestroy(pRsp->rsp.createTableLen);
taosArrayDestroyP(pRsp->rsp.createTableReq, taosMemoryFree); taosArrayDestroyP(pRsp->rsp.createTableReq, taosMemoryFree);
@ -204,7 +204,7 @@ void taos_free_result(TAOS_RES *res) {
taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree); taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
taosArrayDestroy(pRsp->rsp.blockDataLen); taosArrayDestroy(pRsp->rsp.blockDataLen);
taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree); taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper); taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSchemaWrapper);
pRsp->resInfo.pRspMsg = NULL; pRsp->resInfo.pRspMsg = NULL;
doFreeReqResultInfo(&pRsp->resInfo); doFreeReqResultInfo(&pRsp->resInfo);
taosMemoryFree(pRsp); taosMemoryFree(pRsp);

4
source/client/src/clientStmt.c
View File

@ -325,7 +325,7 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
taosHashCleanup(pStmt->exec.pBlockHash); taosHashCleanup(pStmt->exec.pBlockHash);
pStmt->exec.pBlockHash = NULL; pStmt->exec.pBlockHash = NULL;
tDestroySSubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
taosMemoryFreeClear(pStmt->exec.pCurrTbData); taosMemoryFreeClear(pStmt->exec.pCurrTbData);
STMT_ERR_RET(stmtCleanBindInfo(pStmt)); STMT_ERR_RET(stmtCleanBindInfo(pStmt));
@ -895,7 +895,7 @@ int stmtExec(TAOS_STMT* stmt) {
if (STMT_TYPE_QUERY == pStmt->sql.type) { if (STMT_TYPE_QUERY == pStmt->sql.type) {
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL); launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
} else { } else {
tDestroySSubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
taosMemoryFreeClear(pStmt->exec.pCurrTbData); taosMemoryFreeClear(pStmt->exec.pCurrTbData);
STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData)); STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));

4
source/client/src/clientTmq.c
View File

@ -864,7 +864,7 @@ static void* tmqFreeRspWrapper(SMqRspWrapper* rspWrapper) {
taosArrayDestroyP(pRsp->dataRsp.blockData, taosMemoryFree); taosArrayDestroyP(pRsp->dataRsp.blockData, taosMemoryFree);
taosArrayDestroy(pRsp->dataRsp.blockDataLen); taosArrayDestroy(pRsp->dataRsp.blockDataLen);
taosArrayDestroyP(pRsp->dataRsp.blockTbName, taosMemoryFree); taosArrayDestroyP(pRsp->dataRsp.blockTbName, taosMemoryFree);
taosArrayDestroyP(pRsp->dataRsp.blockSchema, (FDelete)tDeleteSSchemaWrapper); taosArrayDestroyP(pRsp->dataRsp.blockSchema, (FDelete)tDeleteSchemaWrapper);
} else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_META_RSP) { } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_META_RSP) {
SMqPollRspWrapper* pRsp = (SMqPollRspWrapper*)rspWrapper; SMqPollRspWrapper* pRsp = (SMqPollRspWrapper*)rspWrapper;
taosMemoryFreeClear(pRsp->pEpset); taosMemoryFreeClear(pRsp->pEpset);
@ -877,7 +877,7 @@ static void* tmqFreeRspWrapper(SMqRspWrapper* rspWrapper) {
taosArrayDestroyP(pRsp->taosxRsp.blockData, taosMemoryFree); taosArrayDestroyP(pRsp->taosxRsp.blockData, taosMemoryFree);
taosArrayDestroy(pRsp->taosxRsp.blockDataLen); taosArrayDestroy(pRsp->taosxRsp.blockDataLen);
taosArrayDestroyP(pRsp->taosxRsp.blockTbName, taosMemoryFree); taosArrayDestroyP(pRsp->taosxRsp.blockTbName, taosMemoryFree);
taosArrayDestroyP(pRsp->taosxRsp.blockSchema, (FDelete)tDeleteSSchemaWrapper); taosArrayDestroyP(pRsp->taosxRsp.blockSchema, (FDelete)tDeleteSchemaWrapper);
// taosx // taosx
taosArrayDestroy(pRsp->taosxRsp.createTableLen); taosArrayDestroy(pRsp->taosxRsp.createTableLen);
taosArrayDestroyP(pRsp->taosxRsp.createTableReq, taosMemoryFree); taosArrayDestroyP(pRsp->taosxRsp.createTableReq, taosMemoryFree);

10
source/client/test/clientTests.cpp
View File

@ -1053,9 +1053,9 @@ TEST(clientCase, sub_db_test) {
} }
TEST(clientCase, sub_tb_test) { TEST(clientCase, sub_tb_test) {
taos_options(TSDB_OPTION_CONFIGDIR, "/home/tests/dir/cfg/"); taos_options(TSDB_OPTION_CONFIGDIR, "~/first/cfg");
TAOS* pConn = taos_connect("vm116", "root", "taosdata", NULL, 0); TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr); ASSERT_NE(pConn, nullptr);
tmq_conf_t* conf = tmq_conf_new(); tmq_conf_t* conf = tmq_conf_new();
@ -1091,7 +1091,7 @@ TEST(clientCase, sub_tb_test) {
int32_t precision = 0; int32_t precision = 0;
int32_t totalRows = 0; int32_t totalRows = 0;
int32_t msgCnt = 0; int32_t msgCnt = 0;
int32_t timeout = 25000; int32_t timeout = 2500000;
int32_t count = 0; int32_t count = 0;
@ -1117,10 +1117,10 @@ TEST(clientCase, sub_tb_test) {
fields = taos_fetch_fields(pRes); fields = taos_fetch_fields(pRes);
numOfFields = taos_field_count(pRes); numOfFields = taos_field_count(pRes);
totalRows += 1; totalRows += 1;
if (totalRows % 100000 == 0) { // if (totalRows % 100000 == 0) {
taos_print_row(buf, row, fields, numOfFields); taos_print_row(buf, row, fields, numOfFields);
printf("row content: %s\n", buf); printf("row content: %s\n", buf);
} // }
} }
taos_free_result(pRes); taos_free_result(pRes);

4
source/common/src/tdatablock.c
View File

@ -2374,7 +2374,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
} }
SRow* pRow = NULL; SRow* pRow = NULL;
if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) { if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) {
tDestroySSubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end; goto _end;
} }
ASSERT(pRow); ASSERT(pRow);
@ -2388,7 +2388,7 @@ _end:
if (terrno != 0) { if (terrno != 0) {
*ppReq = NULL; *ppReq = NULL;
if (pReq) { if (pReq) {
tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFreeClear(pReq); taosMemoryFreeClear(pReq);
} }

4
source/common/src/tdataformat.c
View File

@ -1509,7 +1509,9 @@ void tTagSetCid(const STag *pTag, int16_t iTag, int16_t cid) {
// STSchema ======================================== // STSchema ========================================
STSchema *tBuildTSchema(SSchema *aSchema, int32_t numOfCols, int32_t version) { STSchema *tBuildTSchema(SSchema *aSchema, int32_t numOfCols, int32_t version) {
STSchema *pTSchema = taosMemoryCalloc(1, sizeof(STSchema) + sizeof(STColumn) * numOfCols); STSchema *pTSchema = taosMemoryCalloc(1, sizeof(STSchema) + sizeof(STColumn) * numOfCols);
if (pTSchema == NULL) return NULL; if (pTSchema == NULL) {
return NULL;
}
pTSchema->numOfCols = numOfCols; pTSchema->numOfCols = numOfCols;
pTSchema->version = version; pTSchema->version = version;

14
source/common/src/tmsg.c
View File

@ -7058,7 +7058,7 @@ void tDeleteSMqDataRsp(SMqDataRsp *pRsp) {
pRsp->blockDataLen = taosArrayDestroy(pRsp->blockDataLen); pRsp->blockDataLen = taosArrayDestroy(pRsp->blockDataLen);
taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree); taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree);
pRsp->blockData = NULL; pRsp->blockData = NULL;
taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSSchemaWrapper); taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSchemaWrapper);
pRsp->blockSchema = NULL; pRsp->blockSchema = NULL;
taosArrayDestroyP(pRsp->blockTbName, (FDelete)taosMemoryFree); taosArrayDestroyP(pRsp->blockTbName, (FDelete)taosMemoryFree);
pRsp->blockTbName = NULL; pRsp->blockTbName = NULL;
@ -7159,7 +7159,7 @@ void tDeleteSTaosxRsp(STaosxRsp *pRsp) {
pRsp->blockDataLen = NULL; pRsp->blockDataLen = NULL;
taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree); taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree);
pRsp->blockData = NULL; pRsp->blockData = NULL;
taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSSchemaWrapper); taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSchemaWrapper);
pRsp->blockSchema = NULL; pRsp->blockSchema = NULL;
taosArrayDestroyP(pRsp->blockTbName, (FDelete)taosMemoryFree); taosArrayDestroyP(pRsp->blockTbName, (FDelete)taosMemoryFree);
pRsp->blockTbName = NULL; pRsp->blockTbName = NULL;
@ -7332,7 +7332,7 @@ _exit:
return 0; return 0;
} }
int32_t tEncodeSSubmitReq2(SEncoder *pCoder, const SSubmitReq2 *pReq) { int32_t tEncodeSubmitReq(SEncoder *pCoder, const SSubmitReq2 *pReq) {
if (tStartEncode(pCoder) < 0) return -1; if (tStartEncode(pCoder) < 0) return -1;
if (tEncodeU64v(pCoder, taosArrayGetSize(pReq->aSubmitTbData)) < 0) return -1; if (tEncodeU64v(pCoder, taosArrayGetSize(pReq->aSubmitTbData)) < 0) return -1;
@ -7344,7 +7344,7 @@ int32_t tEncodeSSubmitReq2(SEncoder *pCoder, const SSubmitReq2 *pReq) {
return 0; return 0;
} }
int32_t tDecodeSSubmitReq2(SDecoder *pCoder, SSubmitReq2 *pReq) { int32_t tDecodeSubmitReq(SDecoder *pCoder, SSubmitReq2 *pReq) {
int32_t code = 0; int32_t code = 0;
memset(pReq, 0, sizeof(*pReq)); memset(pReq, 0, sizeof(*pReq));
@ -7387,7 +7387,7 @@ _exit:
return code; return code;
} }
void tDestroySSubmitTbData(SSubmitTbData *pTbData, int32_t flag) { void tDestroySubmitTbData(SSubmitTbData *pTbData, int32_t flag) {
if (NULL == pTbData) { if (NULL == pTbData) {
return; return;
} }
@ -7433,14 +7433,14 @@ void tDestroySSubmitTbData(SSubmitTbData *pTbData, int32_t flag) {
} }
} }
void tDestroySSubmitReq(SSubmitReq2 *pReq, int32_t flag) { void tDestroySubmitReq(SSubmitReq2 *pReq, int32_t flag) {
if (pReq->aSubmitTbData == NULL) return; if (pReq->aSubmitTbData == NULL) return;
int32_t nSubmitTbData = TARRAY_SIZE(pReq->aSubmitTbData); int32_t nSubmitTbData = TARRAY_SIZE(pReq->aSubmitTbData);
SSubmitTbData *aSubmitTbData = (SSubmitTbData *)TARRAY_DATA(pReq->aSubmitTbData); SSubmitTbData *aSubmitTbData = (SSubmitTbData *)TARRAY_DATA(pReq->aSubmitTbData);
for (int32_t i = 0; i < nSubmitTbData; i++) { for (int32_t i = 0; i < nSubmitTbData; i++) {
tDestroySSubmitTbData(&aSubmitTbData[i], flag); tDestroySubmitTbData(&aSubmitTbData[i], flag);
} }
taosArrayDestroy(pReq->aSubmitTbData); taosArrayDestroy(pReq->aSubmitTbData);
pReq->aSubmitTbData = NULL; pReq->aSubmitTbData = NULL;

5
source/dnode/vnode/inc/vnode.h
View File

@ -231,7 +231,7 @@ typedef struct SSnapContext {
} SSnapContext; } SSnapContext;
typedef struct STqReader { typedef struct STqReader {
SPackedData msg2; SPackedData msg;
SSubmitReq2 submit; SSubmitReq2 submit;
int32_t nextBlk; int32_t nextBlk;
int64_t lastBlkUid; int64_t lastBlkUid;
@ -242,7 +242,7 @@ typedef struct STqReader {
int32_t cachedSchemaVer; int32_t cachedSchemaVer;
int64_t cachedSchemaSuid; int64_t cachedSchemaSuid;
SSchemaWrapper *pSchemaWrapper; SSchemaWrapper *pSchemaWrapper;
STSchema *pSchema; SSDataBlock *pResBlock;
} STqReader; } STqReader;
STqReader *tqReaderOpen(SVnode *pVnode); STqReader *tqReaderOpen(SVnode *pVnode);
@ -255,6 +255,7 @@ int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList);
int32_t tqSeekVer(STqReader *pReader, int64_t ver, const char *id); int32_t tqSeekVer(STqReader *pReader, int64_t ver, const char *id);
int32_t tqNextBlock(STqReader *pReader, SSDataBlock* pBlock); int32_t tqNextBlock(STqReader *pReader, SSDataBlock* pBlock);
int32_t tqNextBlockInWal(STqReader* pReader);
int32_t extractSubmitMsgFromWal(SWalReader *pReader, SPackedData *pPackedData); int32_t extractSubmitMsgFromWal(SWalReader *pReader, SPackedData *pPackedData);
int32_t tqReaderSetSubmitMsg(STqReader *pReader, void *msgStr, int32_t msgLen, int64_t ver); int32_t tqReaderSetSubmitMsg(STqReader *pReader, void *msgStr, int32_t msgLen, int64_t ver);

2
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -193,7 +193,7 @@ STQ* tqOpen(const char* path, SVnode* pVnode);
void tqNotifyClose(STQ*); void tqNotifyClose(STQ*);
void tqClose(STQ*); void tqClose(STQ*);
int tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver); int tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver);
int tqRegisterPushEntry(STQ* pTq, void* handle, SRpcMsg* pMsg); int tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg);
int tqUnregisterPushHandle(STQ* pTq, void* pHandle); int tqUnregisterPushHandle(STQ* pTq, void* pHandle);
int tqStartStreamTasks(STQ* pTq); // restore all stream tasks after vnode launching completed. int tqStartStreamTasks(STQ* pTq); // restore all stream tasks after vnode launching completed.

1
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -639,7 +639,6 @@ tb_uid_t metaStbCursorNext(SMStbCursor *pStbCur) {
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, int lock) { STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, int lock) {
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
SSchemaWrapper *pSW = NULL; SSchemaWrapper *pSW = NULL;
SSchema *pSchema = NULL;
pSW = metaGetTableSchema(pMeta, uid, sver, lock); pSW = metaGetTableSchema(pMeta, uid, sver, lock);
if (!pSW) return NULL; if (!pSW) return NULL;

4
source/dnode/vnode/src/meta/metaSnapshot.c
View File

@ -217,8 +217,8 @@ typedef struct STableInfoForChildTable {
static void destroySTableInfoForChildTable(void* data) { static void destroySTableInfoForChildTable(void* data) {
STableInfoForChildTable* pData = (STableInfoForChildTable*)data; STableInfoForChildTable* pData = (STableInfoForChildTable*)data;
taosMemoryFree(pData->tableName); taosMemoryFree(pData->tableName);
tDeleteSSchemaWrapper(pData->schemaRow); tDeleteSchemaWrapper(pData->schemaRow);
tDeleteSSchemaWrapper(pData->tagRow); tDeleteSchemaWrapper(pData->tagRow);
} }
static void MoveToSnapShotVersion(SSnapContext* ctx) { static void MoveToSnapShotVersion(SSnapContext* ctx) {

4
source/dnode/vnode/src/meta/metaTable.c
View File

@ -673,8 +673,8 @@ int metaDropIndexFromSTable(SMeta *pMeta, int64_t version, SDropIndexReq *pReq)
metaUpdateUidIdx(pMeta, &nStbEntry); metaUpdateUidIdx(pMeta, &nStbEntry);
metaULock(pMeta); metaULock(pMeta);
tDeleteSSchemaWrapper(tag); tDeleteSchemaWrapper(tag);
tDeleteSSchemaWrapper(row); tDeleteSchemaWrapper(row);
if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf); if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf);
tDecoderClear(&dc); tDecoderClear(&dc);

4
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -684,7 +684,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
} }
if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) { if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) {
tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq); taosMemoryFree(pReq);
smaError("vgId:%d, process submit req for rsma suid:%" PRIu64 ", uid:%" PRIu64 " level %" PRIi8 smaError("vgId:%d, process submit req for rsma suid:%" PRIu64 ", uid:%" PRIu64 " level %" PRIi8
" failed since %s", " failed since %s",
@ -696,7 +696,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, output->info.version); SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, output->info.version);
if (pReq) { if (pReq) {
tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq); taosMemoryFree(pReq);
} }
} }

8
source/dnode/vnode/src/sma/smaTimeRange.c
View File

@ -299,7 +299,7 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *
} }
SRow *pRow = NULL; SRow *pRow = NULL;
if ((terrno = tRowBuild(pVals, (STSchema *)pTSchema, &pRow)) < 0) { if ((terrno = tRowBuild(pVals, (STSchema *)pTSchema, &pRow)) < 0) {
tDestroySSubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end; goto _end;
} }
taosArrayPush(tbData.aRowP, &pRow); taosArrayPush(tbData.aRowP, &pRow);
@ -309,7 +309,7 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *
} }
// encode // encode
tEncodeSize(tEncodeSSubmitReq2, pReq, len, terrno); tEncodeSize(tEncodeSubmitReq, pReq, len, terrno);
if (TSDB_CODE_SUCCESS == terrno) { if (TSDB_CODE_SUCCESS == terrno) {
SEncoder encoder; SEncoder encoder;
len += sizeof(SSubmitReq2Msg); len += sizeof(SSubmitReq2Msg);
@ -321,7 +321,7 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *
((SSubmitReq2Msg *)pBuf)->header.contLen = htonl(len); ((SSubmitReq2Msg *)pBuf)->header.contLen = htonl(len);
((SSubmitReq2Msg *)pBuf)->version = htobe64(1); ((SSubmitReq2Msg *)pBuf)->version = htobe64(1);
tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg)); tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));
if (tEncodeSSubmitReq2(&encoder, pReq) < 0) { if (tEncodeSubmitReq(&encoder, pReq) < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
/*vError("failed to encode submit req since %s", terrstr());*/ /*vError("failed to encode submit req since %s", terrstr());*/
} }
@ -332,7 +332,7 @@ _end:
taosArrayDestroy(tagArray); taosArrayDestroy(tagArray);
taosArrayDestroy(pVals); taosArrayDestroy(pVals);
if (pReq) { if (pReq) {
tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq); taosMemoryFree(pReq);
} }

2
source/dnode/vnode/src/tq/tqPush.c
View File

@ -287,7 +287,7 @@ int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t v
} }
int32_t tqRegisterPushEntry(STQ* pTq, void* handle, SRpcMsg* pMsg) { int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) {
int32_t vgId = TD_VID(pTq->pVnode); int32_t vgId = TD_VID(pTq->pVnode);
STqHandle* pHandle = (STqHandle*) handle; STqHandle* pHandle = (STqHandle*) handle;
if(pHandle->msg == NULL){ if(pHandle->msg == NULL){

391
source/dnode/vnode/src/tq/tqRead.c
View File

@ -265,9 +265,9 @@ STqReader* tqReaderOpen(SVnode* pVnode) {
pReader->pColIdList = NULL; pReader->pColIdList = NULL;
pReader->cachedSchemaVer = 0; pReader->cachedSchemaVer = 0;
pReader->cachedSchemaSuid = 0; pReader->cachedSchemaSuid = 0;
pReader->pSchema = NULL;
pReader->pSchemaWrapper = NULL; pReader->pSchemaWrapper = NULL;
pReader->tbIdHash = NULL; pReader->tbIdHash = NULL;
pReader->pResBlock = createDataBlock();
return pReader; return pReader;
} }
@ -276,19 +276,19 @@ void tqCloseReader(STqReader* pReader) {
if (pReader->pWalReader) { if (pReader->pWalReader) {
walCloseReader(pReader->pWalReader); walCloseReader(pReader->pWalReader);
} }
// free cached schema
if (pReader->pSchema) {
taosMemoryFree(pReader->pSchema);
}
if (pReader->pSchemaWrapper) { if (pReader->pSchemaWrapper) {
tDeleteSSchemaWrapper(pReader->pSchemaWrapper); tDeleteSchemaWrapper(pReader->pSchemaWrapper);
} }
if (pReader->pColIdList) { if (pReader->pColIdList) {
taosArrayDestroy(pReader->pColIdList); taosArrayDestroy(pReader->pColIdList);
} }
// free hash // free hash
blockDataDestroy(pReader->pResBlock);
taosHashCleanup(pReader->tbIdHash); taosHashCleanup(pReader->tbIdHash);
tDestroySSubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE); tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);
taosMemoryFree(pReader); taosMemoryFree(pReader);
} }
@ -322,9 +322,71 @@ int32_t extractSubmitMsgFromWal(SWalReader* pReader, SPackedData* pPackedData) {
return 0; return 0;
} }
// todo ignore the error in wal?
int32_t tqNextBlockInWal(STqReader* pReader) {
SWalReader* pWalReader = pReader->pWalReader;
while(1) {
SArray* pBlockList = pReader->submit.aSubmitTbData;
if (pBlockList == NULL || pReader->nextBlk >= taosArrayGetSize(pBlockList)) {
// try next message in wal file
if (walNextValidMsg(pWalReader) < 0) {
return FETCH_TYPE__NONE;
}
void* pBody = POINTER_SHIFT(pWalReader->pHead->head.body, sizeof(SSubmitReq2Msg));
int32_t bodyLen = pWalReader->pHead->head.bodyLen - sizeof(SSubmitReq2Msg);
int64_t ver = pWalReader->pHead->head.version;
SDecoder decoder = {0};
tDecoderInit(&decoder, pBody, bodyLen);
if (tDecodeSubmitReq(&decoder, &pReader->submit) < 0) {
tDecoderClear(&decoder);
tqError("decode wal file error, msgLen:%d, ver:%"PRId64, bodyLen, ver);
return FETCH_TYPE__NONE;
}
tDecoderClear(&decoder);
pReader->nextBlk = 0;
}
size_t numOfBlocks = taosArrayGetSize(pReader->submit.aSubmitTbData);
while (pReader->nextBlk < numOfBlocks) {
tqDebug("tq reader next data block %p, %d %" PRId64 " %d", pReader->msg.msgStr, pReader->msg.msgLen,
pReader->msg.ver, pReader->nextBlk);
SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
if (pReader->tbIdHash == NULL) {
int32_t code = tqRetrieveDataBlock(pReader->pResBlock, pReader, NULL);
if (code == TSDB_CODE_SUCCESS && pReader->pResBlock->info.rows > 0) {
return FETCH_TYPE__DATA;
}
}
void* ret = taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t));
if (ret != NULL) {
tqDebug("tq reader return submit block, uid:%"PRId64", ver:%"PRId64, pSubmitTbData->uid, pReader->msg.ver);
int32_t code = tqRetrieveDataBlock(pReader->pResBlock, pReader, NULL);
if (code == TSDB_CODE_SUCCESS && pReader->pResBlock->info.rows > 0) {
return FETCH_TYPE__DATA;
}
} else {
pReader->nextBlk += 1;
tqDebug("tq reader discard submit block, uid:%"PRId64", continue", pSubmitTbData->uid);
}
}
tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);
pReader->msg.msgStr = NULL;
}
}
int32_t tqNextBlock(STqReader* pReader, SSDataBlock* pBlock) { int32_t tqNextBlock(STqReader* pReader, SSDataBlock* pBlock) {
while (1) { while (1) {
if (pReader->msg2.msgStr == NULL) { if (pReader->msg.msgStr == NULL) {
if (walNextValidMsg(pReader->pWalReader) < 0) { if (walNextValidMsg(pReader->pWalReader) < 0) {
return FETCH_TYPE__NONE; return FETCH_TYPE__NONE;
} }
@ -337,8 +399,7 @@ int32_t tqNextBlock(STqReader* pReader, SSDataBlock* pBlock) {
} }
while (tqNextBlockImpl(pReader)) { while (tqNextBlockImpl(pReader)) {
memset(pBlock, 0, sizeof(SSDataBlock)); int32_t code = tqRetrieveDataBlock(pReader->pResBlock, pReader, NULL);
int32_t code = tqRetrieveDataBlock(pBlock, pReader, NULL);
if (code != TSDB_CODE_SUCCESS || pBlock->info.rows == 0) { if (code != TSDB_CODE_SUCCESS || pBlock->info.rows == 0) {
continue; continue;
} }
@ -349,31 +410,33 @@ int32_t tqNextBlock(STqReader* pReader, SSDataBlock* pBlock) {
} }
int32_t tqReaderSetSubmitMsg(STqReader* pReader, void* msgStr, int32_t msgLen, int64_t ver) { int32_t tqReaderSetSubmitMsg(STqReader* pReader, void* msgStr, int32_t msgLen, int64_t ver) {
pReader->msg2.msgStr = msgStr; pReader->msg.msgStr = msgStr;
pReader->msg2.msgLen = msgLen; pReader->msg.msgLen = msgLen;
pReader->msg2.ver = ver; pReader->msg.ver = ver;
tqDebug("tq reader set msg %p %d", msgStr, msgLen); tqDebug("tq reader set msg %p %d", msgStr, msgLen);
SDecoder decoder; SDecoder decoder;
tDecoderInit(&decoder, pReader->msg2.msgStr, pReader->msg2.msgLen);
if (tDecodeSSubmitReq2(&decoder, &pReader->submit) < 0) { tDecoderInit(&decoder, pReader->msg.msgStr, pReader->msg.msgLen);
if (tDecodeSubmitReq(&decoder, &pReader->submit) < 0) {
tDecoderClear(&decoder); tDecoderClear(&decoder);
tqError("DecodeSSubmitReq2 error, msgLen:%d, ver:%"PRId64, msgLen, ver); tqError("DecodeSSubmitReq2 error, msgLen:%d, ver:%"PRId64, msgLen, ver);
return -1; return -1;
} }
tDecoderClear(&decoder); tDecoderClear(&decoder);
return 0; return 0;
} }
bool tqNextBlockImpl(STqReader* pReader) { bool tqNextBlockImpl(STqReader* pReader) {
if (pReader->msg2.msgStr == NULL) { if (pReader->msg.msgStr == NULL) {
return false; return false;
} }
int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData); int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
while (pReader->nextBlk < blockSz) { while (pReader->nextBlk < blockSz) {
tqDebug("tq reader next data block %p, %d %" PRId64 " %d", pReader->msg2.msgStr, pReader->msg2.msgLen, tqDebug("tq reader next data block %p, %d %" PRId64 " %d", pReader->msg.msgStr, pReader->msg.msgLen,
pReader->msg2.ver, pReader->nextBlk); pReader->msg.ver, pReader->nextBlk);
SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
if (pReader->tbIdHash == NULL) { if (pReader->tbIdHash == NULL) {
@ -382,7 +445,7 @@ bool tqNextBlockImpl(STqReader* pReader) {
void* ret = taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t)); void* ret = taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t));
if (ret != NULL) { if (ret != NULL) {
tqDebug("tq reader block found, ver:%"PRId64", uid:%"PRId64, pReader->msg2.ver, pSubmitTbData->uid); tqDebug("tq reader block found, ver:%"PRId64", uid:%"PRId64, pReader->msg.ver, pSubmitTbData->uid);
return true; return true;
} else { } else {
tqDebug("tq reader discard submit block, uid:%"PRId64", continue", pSubmitTbData->uid); tqDebug("tq reader discard submit block, uid:%"PRId64", continue", pSubmitTbData->uid);
@ -391,15 +454,15 @@ bool tqNextBlockImpl(STqReader* pReader) {
pReader->nextBlk++; pReader->nextBlk++;
} }
tDestroySSubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE); tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);
pReader->nextBlk = 0; pReader->nextBlk = 0;
pReader->msg2.msgStr = NULL; pReader->msg.msgStr = NULL;
return false; return false;
} }
bool tqNextDataBlockFilterOut(STqReader* pReader, SHashObj* filterOutUids) { bool tqNextDataBlockFilterOut(STqReader* pReader, SHashObj* filterOutUids) {
if (pReader->msg2.msgStr == NULL) return false; if (pReader->msg.msgStr == NULL) return false;
int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData); int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
while (pReader->nextBlk < blockSz) { while (pReader->nextBlk < blockSz) {
@ -413,9 +476,9 @@ bool tqNextDataBlockFilterOut(STqReader* pReader, SHashObj* filterOutUids) {
pReader->nextBlk++; pReader->nextBlk++;
} }
tDestroySSubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE); tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);
pReader->nextBlk = 0; pReader->nextBlk = 0;
pReader->msg2.msgStr = NULL; pReader->msg.msgStr = NULL;
return false; return false;
} }
@ -450,10 +513,9 @@ int32_t tqMaskBlock(SSchemaWrapper* pDst, SSDataBlock* pBlock, const SSchemaWrap
} }
int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbData** pSubmitTbDataRet) { int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbData** pSubmitTbDataRet) {
tqDebug("tq reader retrieve data block %p, index:%d", pReader->msg2.msgStr, pReader->nextBlk); tqDebug("tq reader retrieve data block %p, index:%d", pReader->msg.msgStr, pReader->nextBlk);
SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
pReader->nextBlk++;
SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk++);
if (pSubmitTbDataRet) { if (pSubmitTbDataRet) {
*pSubmitTbDataRet = pSubmitTbData; *pSubmitTbDataRet = pSubmitTbData;
} }
@ -464,21 +526,11 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbDa
pReader->lastBlkUid = uid; pReader->lastBlkUid = uid;
pBlock->info.id.uid = uid; pBlock->info.id.uid = uid;
pBlock->info.version = pReader->msg2.ver; pBlock->info.version = pReader->msg.ver;
if (pReader->cachedSchemaSuid == 0 || pReader->cachedSchemaVer != sversion || pReader->cachedSchemaSuid != suid) { if (pReader->cachedSchemaSuid == 0 || pReader->cachedSchemaVer != sversion || pReader->cachedSchemaSuid != suid) {
taosMemoryFree(pReader->pSchema); tDeleteSchemaWrapper(pReader->pSchemaWrapper);
pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, uid, sversion, 1);
if (pReader->pSchema == NULL) {
tqWarn("vgId:%d, cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64
"), version %d, possibly dropped table",
pReader->pWalReader->pWal->cfg.vgId, uid, suid, sversion);
pReader->cachedSchemaSuid = 0;
terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
return -1;
}
tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, uid, sversion, 1); pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, uid, sversion, 1);
if (pReader->pSchemaWrapper == NULL) { if (pReader->pSchemaWrapper == NULL) {
tqWarn("vgId:%d, cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table", tqWarn("vgId:%d, cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table",
@ -488,93 +540,140 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbDa
return -1; return -1;
} }
STSchema* pTschema = pReader->pSchema; pReader->cachedSchemaSuid = suid;
pReader->cachedSchemaVer = sversion;
SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper; SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;
int32_t colNumNeed = taosArrayGetSize(pReader->pColIdList); int32_t numOfCols = taosArrayGetSize(pReader->pColIdList);
if (numOfCols == 0) { // all columns are required
if (colNumNeed == 0) { for (int32_t i = 0; i < pSchemaWrapper->nCols; ++i) {
int32_t colMeta = 0; SSchema* pColSchema = &pSchemaWrapper->pSchema[i];
while (colMeta < pSchemaWrapper->nCols) {
SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId); SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
int32_t code = blockDataAppendColInfo(pBlock, &colInfo);
int32_t code = blockDataAppendColInfo(pBlock, &colInfo);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto FAIL; blockDataFreeRes(pBlock);
return -1;
} }
colMeta++;
} }
} else { } else {
if (colNumNeed > pSchemaWrapper->nCols) { if (numOfCols > pSchemaWrapper->nCols) {
colNumNeed = pSchemaWrapper->nCols; numOfCols = pSchemaWrapper->nCols;
} }
int32_t colMeta = 0; int32_t i = 0;
int32_t colNeed = 0; int32_t j = 0;
while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) { while (i < pSchemaWrapper->nCols && j < numOfCols) {
SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta]; SSchema* pColSchema = &pSchemaWrapper->pSchema[i];
col_id_t colIdSchema = pColSchema->colId; col_id_t colIdSchema = pColSchema->colId;
col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pReader->pColIdList, colNeed);
col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pReader->pColIdList, j);
if (colIdSchema < colIdNeed) { if (colIdSchema < colIdNeed) {
colMeta++; i++;
} else if (colIdSchema > colIdNeed) { } else if (colIdSchema > colIdNeed) {
colNeed++; j++;
} else { } else {
SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId); SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
int32_t code = blockDataAppendColInfo(pBlock, &colInfo); int32_t code = blockDataAppendColInfo(pBlock, &colInfo);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto FAIL; goto FAIL;
} }
colMeta++; i++;
colNeed++; j++;
} }
} }
} }
}
int32_t numOfRows = 0; int32_t numOfRows = 0;
if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
SColData* pCol = taosArrayGet(pSubmitTbData->aCol, 0);
numOfRows = pCol->nVal;
} else {
numOfRows = taosArrayGetSize(pSubmitTbData->aRowP);
}
if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) { if (blockDataEnsureCapacity(pBlock, numOfRows) < 0) {
SArray* pCols = pSubmitTbData->aCol; terrno = TSDB_CODE_OUT_OF_MEMORY;
SColData* pCol = taosArrayGet(pCols, 0); goto FAIL;
numOfRows = pCol->nVal; }
} else {
SArray* pRows = pSubmitTbData->aRowP; pBlock->info.rows = numOfRows;
numOfRows = taosArrayGetSize(pRows);
int32_t colActual = blockDataGetNumOfCols(pBlock);
// convert and scan one block
if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
SArray* pCols = pSubmitTbData->aCol;
int32_t numOfCols = taosArrayGetSize(pCols);
int32_t targetIdx = 0;
int32_t sourceIdx = 0;
while (targetIdx < colActual) {
if (sourceIdx >= numOfCols) {
tqError("tqRetrieveDataBlock sourceIdx:%d >= numOfCols:%d", sourceIdx, numOfCols);
goto FAIL;
}
SColData* pCol = taosArrayGet(pCols, sourceIdx);
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx);
SColVal colVal;
if (pCol->nVal != numOfRows) {
tqError("tqRetrieveDataBlock pCol->nVal:%d != numOfRows:%d", pCol->nVal, numOfRows);
goto FAIL;
}
if (pCol->cid < pColData->info.colId) {
sourceIdx++;
} else if (pCol->cid == pColData->info.colId) {
for (int32_t i = 0; i < pCol->nVal; i++) {
tColDataGetValue(pCol, i, &colVal);
if (IS_STR_DATA_TYPE(colVal.type)) {
if (colVal.value.pData != NULL) {
char val[65535 + 2] = {0};
memcpy(varDataVal(val), colVal.value.pData, colVal.value.nData);
varDataSetLen(val, colVal.value.nData);
if (colDataAppend(pColData, i, val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
goto FAIL;
}
} else {
colDataSetNULL(pColData, i);
}
} else {
if (colDataAppend(pColData, i, (void*)&colVal.value.val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
goto FAIL;
}
}
}
sourceIdx++;
targetIdx++;
} else {
for (int32_t i = 0; i < pCol->nVal; i++) {
colDataSetNULL(pColData, i);
}
targetIdx++;
}
} }
} else {
SArray* pRows = pSubmitTbData->aRowP;
SSchemaWrapper* pWrapper = pReader->pSchemaWrapper;
STSchema* pTSchema = tBuildTSchema(pWrapper->pSchema, pWrapper->nCols, pWrapper->version);
if (blockDataEnsureCapacity(pBlock, numOfRows) < 0) { for (int32_t i = 0; i < numOfRows; i++) {
terrno = TSDB_CODE_OUT_OF_MEMORY; SRow* pRow = taosArrayGetP(pRows, i);
goto FAIL;
}
pBlock->info.rows = numOfRows;
int32_t colActual = blockDataGetNumOfCols(pBlock);
// convert and scan one block
if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
SArray* pCols = pSubmitTbData->aCol;
int32_t numOfCols = taosArrayGetSize(pCols);
int32_t targetIdx = 0;
int32_t sourceIdx = 0; int32_t sourceIdx = 0;
while (targetIdx < colActual) {
if(sourceIdx >= numOfCols){
tqError("tqRetrieveDataBlock sourceIdx:%d >= numOfCols:%d", sourceIdx, numOfCols);
goto FAIL;
}
SColData* pCol = taosArrayGet(pCols, sourceIdx);
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx);
SColVal colVal;
if(pCol->nVal != numOfRows){ for (int32_t j = 0; j < colActual; j++) {
tqError("tqRetrieveDataBlock pCol->nVal:%d != numOfRows:%d", pCol->nVal, numOfRows); SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, j);
goto FAIL; while (1) {
} SColVal colVal;
tRowGet(pRow, pTSchema, sourceIdx, &colVal);
if (pCol->cid < pColData->info.colId) { if (colVal.cid < pColData->info.colId) {
sourceIdx++; sourceIdx++;
} else if (pCol->cid == pColData->info.colId) { continue;
for (int32_t i = 0; i < pCol->nVal; i++) { } else if (colVal.cid == pColData->info.colId) {
tColDataGetValue(pCol, i, &colVal);
if (IS_STR_DATA_TYPE(colVal.type)) { if (IS_STR_DATA_TYPE(colVal.type)) {
if (colVal.value.pData != NULL) { if (colVal.value.pData != NULL) {
char val[65535 + 2] = {0}; char val[65535 + 2] = {0};
@ -591,59 +690,18 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbDa
goto FAIL; goto FAIL;
} }
} }
}
sourceIdx++; sourceIdx++;
targetIdx++; break;
} else { } else {
for (int32_t i = 0; i < pCol->nVal; i++) {
colDataSetNULL(pColData, i); colDataSetNULL(pColData, i);
} break;
targetIdx++;
}
}
} else {
SArray* pRows = pSubmitTbData->aRowP;
for (int32_t i = 0; i < numOfRows; i++) {
SRow* pRow = taosArrayGetP(pRows, i);
int32_t sourceIdx = 0;
for (int32_t j = 0; j < colActual; j++) {
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, j);
while (1) {
SColVal colVal;
tRowGet(pRow, pTschema, sourceIdx, &colVal);
if (colVal.cid < pColData->info.colId) {
sourceIdx++;
continue;
} else if (colVal.cid == pColData->info.colId) {
if (IS_STR_DATA_TYPE(colVal.type)) {
if (colVal.value.pData != NULL) {
char val[65535 + 2] = {0};
memcpy(varDataVal(val), colVal.value.pData, colVal.value.nData);
varDataSetLen(val, colVal.value.nData);
if (colDataAppend(pColData, i, val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
goto FAIL;
}
} else {
colDataSetNULL(pColData, i);
}
} else {
if (colDataAppend(pColData, i, (void*)&colVal.value.val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
goto FAIL;
}
}
sourceIdx++;
break;
} else {
colDataSetNULL(pColData, i);
break;
}
} }
} }
} }
} }
taosMemoryFreeClear(pTSchema);
} }
return 0; return 0;
@ -654,7 +712,7 @@ FAIL:
} }
int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas, SSubmitTbData** pSubmitTbDataRet) { int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas, SSubmitTbData** pSubmitTbDataRet) {
tqDebug("tq reader retrieve data block %p, %d", pReader->msg2.msgStr, pReader->nextBlk); tqDebug("tq reader retrieve data block %p, %d", pReader->msg.msgStr, pReader->nextBlk);
SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
pReader->nextBlk++; pReader->nextBlk++;
@ -665,18 +723,7 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
int64_t uid = pSubmitTbData->uid; int64_t uid = pSubmitTbData->uid;
pReader->lastBlkUid = uid; pReader->lastBlkUid = uid;
taosMemoryFree(pReader->pSchema); tDeleteSchemaWrapper(pReader->pSchemaWrapper);
pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, uid, sversion, 1);
if (pReader->pSchema == NULL) {
tqWarn("vgId:%d, cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64
"), version %d, possibly dropped table",
pReader->pWalReader->pWal->cfg.vgId, uid, suid, sversion);
pReader->cachedSchemaSuid = 0;
terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
return -1;
}
tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, uid, sversion, 1); pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, uid, sversion, 1);
if (pReader->pSchemaWrapper == NULL) { if (pReader->pSchemaWrapper == NULL) {
tqWarn("vgId:%d, cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table", tqWarn("vgId:%d, cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table",
@ -686,7 +733,6 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
return -1; return -1;
} }
STSchema* pTschema = pReader->pSchema;
SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper; SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;
int32_t numOfRows = 0; int32_t numOfRows = 0;
@ -743,18 +789,18 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
if (tqMaskBlock(pSW, &block, pSchemaWrapper, assigned) < 0) { if (tqMaskBlock(pSW, &block, pSchemaWrapper, assigned) < 0) {
blockDataFreeRes(&block); blockDataFreeRes(&block);
tDeleteSSchemaWrapper(pSW); tDeleteSchemaWrapper(pSW);
goto FAIL; goto FAIL;
} }
tqDebug("vgId:%d, build new block, col %d", pReader->pWalReader->pWal->cfg.vgId, tqDebug("vgId:%d, build new block, col %d", pReader->pWalReader->pWal->cfg.vgId,
(int32_t)taosArrayGetSize(block.pDataBlock)); (int32_t)taosArrayGetSize(block.pDataBlock));
block.info.id.uid = uid; block.info.id.uid = uid;
block.info.version = pReader->msg2.ver; block.info.version = pReader->msg.ver;
if (blockDataEnsureCapacity(&block, numOfRows - curRow) < 0) { if (blockDataEnsureCapacity(&block, numOfRows - curRow) < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
blockDataFreeRes(&block); blockDataFreeRes(&block);
tDeleteSSchemaWrapper(pSW); tDeleteSchemaWrapper(pSW);
goto FAIL; goto FAIL;
} }
taosArrayPush(blocks, &block); taosArrayPush(blocks, &block);
@ -803,14 +849,17 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
curRow++; curRow++;
} }
} else { } else {
SSchemaWrapper* pWrapper = pReader->pSchemaWrapper;
STSchema* pTSchema = tBuildTSchema(pWrapper->pSchema, pWrapper->nCols, pWrapper->version);
SArray* pRows = pSubmitTbData->aRowP; SArray* pRows = pSubmitTbData->aRowP;
for (int32_t i = 0; i < numOfRows; i++) { for (int32_t i = 0; i < numOfRows; i++) {
SRow* pRow = taosArrayGetP(pRows, i); SRow* pRow = taosArrayGetP(pRows, i);
bool buildNew = false; bool buildNew = false;
for (int32_t j = 0; j < pTschema->numOfCols; j++) { for (int32_t j = 0; j < pTSchema->numOfCols; j++) {
SColVal colVal; SColVal colVal;
tRowGet(pRow, pTschema, j, &colVal); tRowGet(pRow, pTSchema, j, &colVal);
if (curRow == 0) { if (curRow == 0) {
assigned[j] = !COL_VAL_IS_NONE(&colVal); assigned[j] = !COL_VAL_IS_NONE(&colVal);
buildNew = true; buildNew = true;
@ -839,18 +888,18 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
if (tqMaskBlock(pSW, &block, pSchemaWrapper, assigned) < 0) { if (tqMaskBlock(pSW, &block, pSchemaWrapper, assigned) < 0) {
blockDataFreeRes(&block); blockDataFreeRes(&block);
tDeleteSSchemaWrapper(pSW); tDeleteSchemaWrapper(pSW);
goto FAIL; goto FAIL;
} }
tqDebug("vgId:%d, build new block, col %d", pReader->pWalReader->pWal->cfg.vgId, tqDebug("vgId:%d, build new block, col %d", pReader->pWalReader->pWal->cfg.vgId,
(int32_t)taosArrayGetSize(block.pDataBlock)); (int32_t)taosArrayGetSize(block.pDataBlock));
block.info.id.uid = uid; block.info.id.uid = uid;
block.info.version = pReader->msg2.ver; block.info.version = pReader->msg.ver;
if (blockDataEnsureCapacity(&block, numOfRows - curRow) < 0) { if (blockDataEnsureCapacity(&block, numOfRows - curRow) < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
blockDataFreeRes(&block); blockDataFreeRes(&block);
tDeleteSSchemaWrapper(pSW); tDeleteSchemaWrapper(pSW);
goto FAIL; goto FAIL;
} }
taosArrayPush(blocks, &block); taosArrayPush(blocks, &block);
@ -868,7 +917,7 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
while (targetIdx < colActual) { while (targetIdx < colActual) {
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx); SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx);
SColVal colVal; SColVal colVal;
tRowGet(pRow, pTschema, sourceIdx, &colVal); tRowGet(pRow, pTSchema, sourceIdx, &colVal);
if (colVal.cid < pColData->info.colId) { if (colVal.cid < pColData->info.colId) {
sourceIdx++; sourceIdx++;
@ -895,6 +944,8 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
} }
curRow++; curRow++;
} }
taosMemoryFreeClear(pTSchema);
} }
SSDataBlock* pLastBlock = taosArrayGetLast(blocks); SSDataBlock* pLastBlock = taosArrayGetLast(blocks);

4
source/dnode/vnode/src/tq/tqScan.c
View File

@ -215,7 +215,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR
int64_t uid = pExec->pTqReader->lastBlkUid; int64_t uid = pExec->pTqReader->lastBlkUid;
if (tqAddTbNameToRsp(pTq, uid, pRsp, taosArrayGetSize(pBlocks)) < 0) { if (tqAddTbNameToRsp(pTq, uid, pRsp, taosArrayGetSize(pBlocks)) < 0) {
taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes); taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);
taosArrayDestroyP(pSchemas, (FDelete)tDeleteSSchemaWrapper); taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);
pBlocks = taosArrayInit(0, sizeof(SSDataBlock)); pBlocks = taosArrayInit(0, sizeof(SSDataBlock));
pSchemas = taosArrayInit(0, sizeof(void*)); pSchemas = taosArrayInit(0, sizeof(void*));
continue; continue;
@ -274,7 +274,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR
int64_t uid = pExec->pTqReader->lastBlkUid; int64_t uid = pExec->pTqReader->lastBlkUid;
if (tqAddTbNameToRsp(pTq, uid, pRsp, taosArrayGetSize(pBlocks)) < 0) { if (tqAddTbNameToRsp(pTq, uid, pRsp, taosArrayGetSize(pBlocks)) < 0) {
taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes); taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);
taosArrayDestroyP(pSchemas, (FDelete)tDeleteSSchemaWrapper); taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);
pBlocks = taosArrayInit(0, sizeof(SSDataBlock)); pBlocks = taosArrayInit(0, sizeof(SSDataBlock));
pSchemas = taosArrayInit(0, sizeof(void*)); pSchemas = taosArrayInit(0, sizeof(void*));
continue; continue;

14
source/dnode/vnode/src/tq/tqSink.c
View File

@ -672,7 +672,7 @@ void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void*
} }
SRow* pRow = NULL; SRow* pRow = NULL;
if ((terrno = tRowBuild(pVals, (STSchema*)pTSchema, &pRow)) < 0) { if ((terrno = tRowBuild(pVals, (STSchema*)pTSchema, &pRow)) < 0) {
tDestroySSubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end; goto _end;
} }
ASSERT(pRow); ASSERT(pRow);
@ -681,7 +681,7 @@ void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void*
SSubmitReq2 submitReq = {0}; SSubmitReq2 submitReq = {0};
if (!(submitReq.aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) { if (!(submitReq.aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
tDestroySSubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end; goto _end;
} }
@ -690,28 +690,28 @@ void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void*
// encode // encode
int32_t len; int32_t len;
int32_t code; int32_t code;
tEncodeSize(tEncodeSSubmitReq2, &submitReq, len, code); tEncodeSize(tEncodeSubmitReq, &submitReq, len, code);
SEncoder encoder; SEncoder encoder;
len += sizeof(SSubmitReq2Msg); len += sizeof(SSubmitReq2Msg);
pBuf = rpcMallocCont(len); pBuf = rpcMallocCont(len);
if (NULL == pBuf) { if (NULL == pBuf) {
tDestroySSubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE);
goto _end; goto _end;
} }
((SSubmitReq2Msg*)pBuf)->header.vgId = TD_VID(pVnode); ((SSubmitReq2Msg*)pBuf)->header.vgId = TD_VID(pVnode);
((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len); ((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len);
((SSubmitReq2Msg*)pBuf)->version = htobe64(1); ((SSubmitReq2Msg*)pBuf)->version = htobe64(1);
tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg)); tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));
if (tEncodeSSubmitReq2(&encoder, &submitReq) < 0) { if (tEncodeSubmitReq(&encoder, &submitReq) < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
tqError("failed to encode submit req since %s", terrstr()); tqError("failed to encode submit req since %s", terrstr());
tEncoderClear(&encoder); tEncoderClear(&encoder);
rpcFreeCont(pBuf); rpcFreeCont(pBuf);
tDestroySSubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE);
continue; continue;
} }
tEncoderClear(&encoder); tEncoderClear(&encoder);
tDestroySSubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE);
SRpcMsg msg = { SRpcMsg msg = {
.msgType = TDMT_VND_SUBMIT, .msgType = TDMT_VND_SUBMIT,

2
source/dnode/vnode/src/tq/tqUtil.c
View File

@ -180,7 +180,7 @@ static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle,
dataRsp.reqOffset.version == dataRsp.rspOffset.version && pHandle->consumerId == pRequest->consumerId) { dataRsp.reqOffset.version == dataRsp.rspOffset.version && pHandle->consumerId == pRequest->consumerId) {
// lock // lock
taosWLockLatch(&pTq->lock); taosWLockLatch(&pTq->lock);
code = tqRegisterPushEntry(pTq, pHandle, pMsg); code = tqRegisterPushHandle(pTq, pHandle, pMsg);
taosWUnLockLatch(&pTq->lock); taosWUnLockLatch(&pTq->lock);
tDeleteSMqDataRsp(&dataRsp); tDeleteSMqDataRsp(&dataRsp);
return code; return code;

12
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -1007,7 +1007,7 @@ static int32_t vnodeResetTableCxt(SMeta *pMeta, SSubmitReqConvertCxt *pCxt) {
} }
tdSTSRowIterInit(&pCxt->rowIter, pCxt->pTbSchema); tdSTSRowIterInit(&pCxt->rowIter, pCxt->pTbSchema);
tDestroySSubmitTbData(pCxt->pTbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(pCxt->pTbData, TSDB_MSG_FLG_ENCODE);
if (NULL == pCxt->pTbData) { if (NULL == pCxt->pTbData) {
pCxt->pTbData = taosMemoryCalloc(1, sizeof(SSubmitTbData)); pCxt->pTbData = taosMemoryCalloc(1, sizeof(SSubmitTbData));
if (NULL == pCxt->pTbData) { if (NULL == pCxt->pTbData) {
@ -1039,7 +1039,7 @@ static int32_t vnodeResetTableCxt(SMeta *pMeta, SSubmitReqConvertCxt *pCxt) {
static void vnodeDestroySubmitReqConvertCxt(SSubmitReqConvertCxt *pCxt) { static void vnodeDestroySubmitReqConvertCxt(SSubmitReqConvertCxt *pCxt) {
taosMemoryFreeClear(pCxt->pTbSchema); taosMemoryFreeClear(pCxt->pTbSchema);
tDestroySSubmitTbData(pCxt->pTbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(pCxt->pTbData, TSDB_MSG_FLG_ENCODE);
taosMemoryFreeClear(pCxt->pTbData); taosMemoryFreeClear(pCxt->pTbData);
taosArrayDestroy(pCxt->pColValues); taosArrayDestroy(pCxt->pColValues);
} }
@ -1149,7 +1149,7 @@ static int32_t vnodeRebuildSubmitReqMsg(SSubmitReq2 *pSubmitReq, void **ppMsg) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
char *pMsg = NULL; char *pMsg = NULL;
uint32_t msglen = 0; uint32_t msglen = 0;
tEncodeSize(tEncodeSSubmitReq2, pSubmitReq, msglen, code); tEncodeSize(tEncodeSubmitReq, pSubmitReq, msglen, code);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
pMsg = taosMemoryMalloc(msglen); pMsg = taosMemoryMalloc(msglen);
if (NULL == pMsg) { if (NULL == pMsg) {
@ -1159,7 +1159,7 @@ static int32_t vnodeRebuildSubmitReqMsg(SSubmitReq2 *pSubmitReq, void **ppMsg) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
SEncoder encoder; SEncoder encoder;
tEncoderInit(&encoder, pMsg, msglen); tEncoderInit(&encoder, pMsg, msglen);
code = tEncodeSSubmitReq2(&encoder, pSubmitReq); code = tEncodeSubmitReq(&encoder, pSubmitReq);
tEncoderClear(&encoder); tEncoderClear(&encoder);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
@ -1199,7 +1199,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
len -= sizeof(SSubmitReq2Msg); len -= sizeof(SSubmitReq2Msg);
SDecoder dc = {0}; SDecoder dc = {0};
tDecoderInit(&dc, pReq, len); tDecoderInit(&dc, pReq, len);
if (tDecodeSSubmitReq2(&dc, pSubmitReq) < 0) { if (tDecodeSubmitReq(&dc, pSubmitReq) < 0) {
code = TSDB_CODE_INVALID_MSG; code = TSDB_CODE_INVALID_MSG;
goto _exit; goto _exit;
} }
@ -1388,7 +1388,7 @@ _exit:
// clear // clear
taosArrayDestroy(newTbUids); taosArrayDestroy(newTbUids);
tDestroySSubmitReq(pSubmitReq, 0 == pMsg->version ? TSDB_MSG_FLG_CMPT : TSDB_MSG_FLG_DECODE); tDestroySubmitReq(pSubmitReq, 0 == pMsg->version ? TSDB_MSG_FLG_CMPT : TSDB_MSG_FLG_DECODE);
tDestroySSubmitRsp2(pSubmitRsp, TSDB_MSG_FLG_ENCODE); tDestroySSubmitRsp2(pSubmitRsp, TSDB_MSG_FLG_ENCODE);
if (code) terrno = code; if (code) terrno = code;

12
source/libs/executor/src/dataInserter.c
View File

@ -126,7 +126,7 @@ static int32_t submitReqToMsg(int32_t vgId, SSubmitReq2* pReq, void** pData, int
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t len = 0; int32_t len = 0;
void* pBuf = NULL; void* pBuf = NULL;
tEncodeSize(tEncodeSSubmitReq2, pReq, len, code); tEncodeSize(tEncodeSubmitReq, pReq, len, code);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
SEncoder encoder; SEncoder encoder;
len += sizeof(SSubmitReq2Msg); len += sizeof(SSubmitReq2Msg);
@ -138,7 +138,7 @@ static int32_t submitReqToMsg(int32_t vgId, SSubmitReq2* pReq, void** pData, int
((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len); ((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len);
((SSubmitReq2Msg*)pBuf)->version = htobe64(1); ((SSubmitReq2Msg*)pBuf)->version = htobe64(1);
tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg)); tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));
code = tEncodeSSubmitReq2(&encoder, pReq); code = tEncodeSubmitReq(&encoder, pReq);
tEncoderClear(&encoder); tEncoderClear(&encoder);
} }
@ -281,7 +281,7 @@ int32_t buildSubmitReqFromBlock(SDataInserterHandle* pInserter, SSubmitReq2** pp
SRow* pRow = NULL; SRow* pRow = NULL;
if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) { if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) {
tDestroySSubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
goto _end; goto _end;
} }
taosArrayPush(tbData.aRowP, &pRow); taosArrayPush(tbData.aRowP, &pRow);
@ -301,7 +301,7 @@ _end:
if (terrno != 0) { if (terrno != 0) {
*ppReq = NULL; *ppReq = NULL;
if (pReq) { if (pReq) {
tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq); taosMemoryFree(pReq);
} }
return terrno; return terrno;
@ -326,7 +326,7 @@ int32_t dataBlocksToSubmitReq(SDataInserterHandle* pInserter, void** pMsg, int32
code = buildSubmitReqFromBlock(pInserter, &pReq, pDataBlock, pTSchema, uid, vgId, suid); code = buildSubmitReqFromBlock(pInserter, &pReq, pDataBlock, pTSchema, uid, vgId, suid);
if (code) { if (code) {
if (pReq) { if (pReq) {
tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq); taosMemoryFree(pReq);
} }
@ -335,7 +335,7 @@ int32_t dataBlocksToSubmitReq(SDataInserterHandle* pInserter, void** pMsg, int32
} }
code = submitReqToMsg(vgId, pReq, pMsg, msgLen); code = submitReqToMsg(vgId, pReq, pMsg, msgLen);
tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq); taosMemoryFree(pReq);
return code; return code;

16
source/libs/executor/src/executor.c
View File

@ -1052,19 +1052,6 @@ int32_t initQueryTableDataCondForTmq(SQueryTableDataCond* pCond, SSnapContext* s
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
if ((pTaskInfo->execModel != OPTR_EXEC_MODEL_QUEUE) || (pTaskInfo->streamInfo.submit.msgStr != NULL)) {
qError("qStreamSetScanMemData err:%d,%p", pTaskInfo->execModel, pTaskInfo->streamInfo.submit.msgStr);
terrno = TSDB_CODE_PAR_INTERNAL_ERROR;
return -1;
}
qDebug("set the submit block for future scan");
pTaskInfo->streamInfo.submit = submit;
return 0;
}
void qStreamSetOpen(qTaskInfo_t tinfo) { void qStreamSetOpen(qTaskInfo_t tinfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
SOperatorInfo* pOperator = pTaskInfo->pRoot; SOperatorInfo* pOperator = pTaskInfo->pRoot;
@ -1086,6 +1073,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
if (pOperator == NULL) { if (pOperator == NULL) {
return -1; return -1;
} }
SStreamScanInfo* pInfo = pOperator->info; SStreamScanInfo* pInfo = pOperator->info;
STableScanInfo* pScanInfo = pInfo->pTableScanOp->info; STableScanInfo* pScanInfo = pInfo->pTableScanOp->info;
STableScanBase* pScanBaseInfo = &pScanInfo->base; STableScanBase* pScanBaseInfo = &pScanInfo->base;
@ -1221,7 +1209,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond); cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);
strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName); strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName);
tDeleteSSchemaWrapper(pTaskInfo->streamInfo.schema); tDeleteSchemaWrapper(pTaskInfo->streamInfo.schema);
pTaskInfo->streamInfo.schema = mtInfo.schema; pTaskInfo->streamInfo.schema = mtInfo.schema;
qDebug("tmqsnap qStreamPrepareScan snapshot data uid:%" PRId64 " ts %" PRId64 " %s", mtInfo.uid, pOffset->ts, id); qDebug("tmqsnap qStreamPrepareScan snapshot data uid:%" PRId64 " ts %" PRId64 " %s", mtInfo.uid, pOffset->ts, id);

6
source/libs/executor/src/querytask.c
View File

@ -109,8 +109,8 @@ int32_t createExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHand
void cleanupQueriedTableScanInfo(SSchemaInfo* pSchemaInfo) { void cleanupQueriedTableScanInfo(SSchemaInfo* pSchemaInfo) {
taosMemoryFreeClear(pSchemaInfo->dbname); taosMemoryFreeClear(pSchemaInfo->dbname);
taosMemoryFreeClear(pSchemaInfo->tablename); taosMemoryFreeClear(pSchemaInfo->tablename);
tDeleteSSchemaWrapper(pSchemaInfo->sw); tDeleteSchemaWrapper(pSchemaInfo->sw);
tDeleteSSchemaWrapper(pSchemaInfo->qsw); tDeleteSchemaWrapper(pSchemaInfo->qsw);
} }
int32_t initQueriedTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, const char* dbName, SExecTaskInfo* pTaskInfo) { int32_t initQueriedTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, const char* dbName, SExecTaskInfo* pTaskInfo) {
@ -197,7 +197,7 @@ SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode) {
return pqSw; return pqSw;
} }
static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) { tDeleteSSchemaWrapper(pStreamInfo->schema); } static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) { tDeleteSchemaWrapper(pStreamInfo->schema); }
static void freeBlock(void* pParam) { static void freeBlock(void* pParam) {
SSDataBlock* pBlock = *(SSDataBlock**)pParam; SSDataBlock* pBlock = *(SSDataBlock**)pParam;

18
source/libs/executor/src/scanoperator.c
View File

@ -1623,7 +1623,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
pInfo->pRes->info.dataLoad = 1; pInfo->pRes->info.dataLoad = 1;
blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex); blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
blockDataFreeRes((SSDataBlock*)pBlock); // blockDataFreeRes((SSDataBlock*)pBlock);
calBlockTbName(pInfo, pInfo->pRes); calBlockTbName(pInfo, pInfo->pRes);
return 0; return 0;
@ -1637,7 +1637,7 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
qDebug("start to exec queue scan, %s", id); qDebug("start to exec queue scan, %s", id);
if (pTaskInfo->streamInfo.submit.msgStr != NULL) { if (pTaskInfo->streamInfo.submit.msgStr != NULL) {
if (pInfo->tqReader->msg2.msgStr == NULL) { if (pInfo->tqReader->msg.msgStr == NULL) {
SPackedData submit = pTaskInfo->streamInfo.submit; SPackedData submit = pTaskInfo->streamInfo.submit;
if (tqReaderSetSubmitMsg(pInfo->tqReader, submit.msgStr, submit.msgLen, submit.ver) < 0) { if (tqReaderSetSubmitMsg(pInfo->tqReader, submit.msgStr, submit.msgLen, submit.ver) < 0) {
qError("submit msg messed up when initing stream submit block %p", submit.msgStr); qError("submit msg messed up when initing stream submit block %p", submit.msgStr);
@ -1663,7 +1663,7 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
} }
} }
pInfo->tqReader->msg2 = (SPackedData){0}; pInfo->tqReader->msg = (SPackedData){0};
pTaskInfo->streamInfo.submit = (SPackedData){0}; pTaskInfo->streamInfo.submit = (SPackedData){0};
return NULL; return NULL;
} }
@ -1689,17 +1689,17 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__LOG) { if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__LOG) {
while (1) { while (1) {
SSDataBlock block = {0}; int32_t type = tqNextBlockInWal(pInfo->tqReader);
int32_t type = tqNextBlock(pInfo->tqReader, &block); SSDataBlock* pRes = pInfo->tqReader->pResBlock;
// curVersion move to next, so currentOffset = curVersion - 1 // curVersion move to next, so currentOffset = curVersion - 1
tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pInfo->tqReader->pWalReader->curVersion - 1); tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pInfo->tqReader->pWalReader->curVersion - 1);
if (type == FETCH_TYPE__DATA) { if (type == FETCH_TYPE__DATA) {
qDebug("doQueueScan get data from log %" PRId64 " rows, version:%" PRId64, block.info.rows, qDebug("doQueueScan get data from log %" PRId64 " rows, version:%" PRId64, pRes->info.rows,
pTaskInfo->streamInfo.currentOffset.version); pTaskInfo->streamInfo.currentOffset.version);
blockDataCleanup(pInfo->pRes); blockDataCleanup(pInfo->pRes);
setBlockIntoRes(pInfo, &block, true); setBlockIntoRes(pInfo, pRes, true);
if (pInfo->pRes->info.rows > 0) { if (pInfo->pRes->info.rows > 0) {
qDebug("doQueueScan get data from log %" PRId64 " rows, return, version:%" PRId64, pInfo->pRes->info.rows, qDebug("doQueueScan get data from log %" PRId64 " rows, return, version:%" PRId64, pInfo->pRes->info.rows,
pTaskInfo->streamInfo.currentOffset.version); pTaskInfo->streamInfo.currentOffset.version);
@ -2055,7 +2055,7 @@ FETCH_NEXT_BLOCK:
NEXT_SUBMIT_BLK: NEXT_SUBMIT_BLK:
while (1) { while (1) {
if (pInfo->tqReader->msg2.msgStr == NULL) { if (pInfo->tqReader->msg.msgStr == NULL) {
if (pInfo->validBlockIndex >= totBlockNum) { if (pInfo->validBlockIndex >= totBlockNum) {
updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo); updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
doClearBufferedBlocks(pInfo); doClearBufferedBlocks(pInfo);
@ -2191,7 +2191,7 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
qDebug("tmqsnap change get data uid:%" PRId64 "", mtInfo.uid); qDebug("tmqsnap change get data uid:%" PRId64 "", mtInfo.uid);
} }
qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType); qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);
tDeleteSSchemaWrapper(mtInfo.schema); tDeleteSchemaWrapper(mtInfo.schema);
return NULL; return NULL;
} else if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_META) { } else if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_META) {
SSnapContext* sContext = pInfo->sContext; SSnapContext* sContext = pInfo->sContext;

8
source/libs/parser/src/parInsertUtil.c
View File

@ -313,7 +313,7 @@ void insDestroyTableDataCxt(STableDataCxt* pTableCxt) {
insDestroyBoundColInfo(&pTableCxt->boundColsInfo); insDestroyBoundColInfo(&pTableCxt->boundColsInfo);
taosArrayDestroyEx(pTableCxt->pValues, destroyColVal); taosArrayDestroyEx(pTableCxt->pValues, destroyColVal);
if (pTableCxt->pData) { if (pTableCxt->pData) {
tDestroySSubmitTbData(pTableCxt->pData, TSDB_MSG_FLG_ENCODE); tDestroySubmitTbData(pTableCxt->pData, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pTableCxt->pData); taosMemoryFree(pTableCxt->pData);
} }
taosMemoryFree(pTableCxt); taosMemoryFree(pTableCxt);
@ -324,7 +324,7 @@ void insDestroyVgroupDataCxt(SVgroupDataCxt* pVgCxt) {
return; return;
} }
tDestroySSubmitReq(pVgCxt->pData, TSDB_MSG_FLG_ENCODE); tDestroySubmitReq(pVgCxt->pData, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pVgCxt->pData); taosMemoryFree(pVgCxt->pData);
taosMemoryFree(pVgCxt); taosMemoryFree(pVgCxt);
} }
@ -499,7 +499,7 @@ static int32_t buildSubmitReq(int32_t vgId, SSubmitReq2* pReq, void** pData, uin
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
uint32_t len = 0; uint32_t len = 0;
void* pBuf = NULL; void* pBuf = NULL;
tEncodeSize(tEncodeSSubmitReq2, pReq, len, code); tEncodeSize(tEncodeSubmitReq, pReq, len, code);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
SEncoder encoder; SEncoder encoder;
len += sizeof(SSubmitReq2Msg); len += sizeof(SSubmitReq2Msg);
@ -511,7 +511,7 @@ static int32_t buildSubmitReq(int32_t vgId, SSubmitReq2* pReq, void** pData, uin
((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len); ((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len);
((SSubmitReq2Msg*)pBuf)->version = htobe64(1); ((SSubmitReq2Msg*)pBuf)->version = htobe64(1);
tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg)); tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));
code = tEncodeSSubmitReq2(&encoder, pReq); code = tEncodeSubmitReq(&encoder, pReq);
tEncoderClear(&encoder); tEncoderClear(&encoder);
} }

2
source/libs/stream/src/streamTask.c
View File

@ -193,7 +193,7 @@ void tFreeStreamTask(SStreamTask* pTask) {
taosArrayDestroyP(pTask->childEpInfo, taosMemoryFree); taosArrayDestroyP(pTask->childEpInfo, taosMemoryFree);
if (pTask->outputType == TASK_OUTPUT__TABLE) { if (pTask->outputType == TASK_OUTPUT__TABLE) {
tDeleteSSchemaWrapper(pTask->tbSink.pSchemaWrapper); tDeleteSchemaWrapper(pTask->tbSink.pSchemaWrapper);
taosMemoryFree(pTask->tbSink.pTSchema); taosMemoryFree(pTask->tbSink.pTSchema);
} }