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

more code

This commit is contained in:
Hongze Cheng 2022-11-28 13:41:06 +08:00
parent 167c8051ba
commit 0f8016995c
9 changed files with 162 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -103,6 +103,7 @@ int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SRow **ppRow) {
ASSERT(((SColVal *)aColVal->pData)[0].type == TSDB_DATA_TYPE_TIMESTAMP); ASSERT(((SColVal *)aColVal->pData)[0].type == TSDB_DATA_TYPE_TIMESTAMP);
// scan --------------- // scan ---------------
SRow *pRow = NULL;
uint8_t flag = 0; uint8_t flag = 0;
int32_t iColVal = 1; int32_t iColVal = 1;
const int32_t nColVal = taosArrayGetSize(aColVal); const int32_t nColVal = taosArrayGetSize(aColVal);
@ -196,9 +197,11 @@ int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SRow **ppRow) {
} }
// alloc -------------- // alloc --------------
code = tRealloc((uint8_t **)ppRow, nRow); pRow = taosMemoryMalloc(nRow);
if (code) return code; if (NULL == pRow) {
SRow *pRow = *ppRow; code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
// build -------------- // build --------------
pColVal = (SColVal *)taosArrayGet(aColVal, 0); pColVal = (SColVal *)taosArrayGet(aColVal, 0);
@ -349,6 +352,12 @@ int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SRow **ppRow) {
} }
_exit: _exit:
if (code) {
*ppRow = NULL;
tRowDestroy(pRow);
} else {
*ppRow = pRow;
}
return code; return code;
} }
@ -490,7 +499,9 @@ void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
} }
} }
void tRowDestroy(SRow *pRow) { tFree((uint8_t *)pRow); } void tRowDestroy(SRow *pRow) {
if (pRow) taosMemoryFree(pRow);
}
static int32_t tRowPCmprFn(const void *p1, const void *p2) { static int32_t tRowPCmprFn(const void *p1, const void *p2) {
if ((*(SRow **)p1)->ts < (*(SRow **)p2)->ts) { if ((*(SRow **)p1)->ts < (*(SRow **)p2)->ts) {

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

@ -6665,7 +6665,7 @@ static int32_t tEncodeSSubmitTbData(SEncoder *pCoder, const SSubmitTbData *pSubm
if (tEncodeI64v(pCoder, taosArrayGetSize(pSubmitTbData->aRowP)) < 0) return -1; if (tEncodeI64v(pCoder, taosArrayGetSize(pSubmitTbData->aRowP)) < 0) return -1;
for (int32_t i = 0; i < taosArrayGetSize(pSubmitTbData->aRowP); i++) { for (int32_t i = 0; i < taosArrayGetSize(pSubmitTbData->aRowP); i++) {
SRow *pRow = taosArrayGetP(pSubmitTbData->aRowP, i); SRow *pRow = taosArrayGetP(pSubmitTbData->aRowP, i);
memcpy(pCoder->data + pCoder->pos, pRow, pRow->len); if (pCoder->data) memcpy(pCoder->data + pCoder->pos, pRow, pRow->len);
pCoder->pos += pRow->len; pCoder->pos += pRow->len;
} }
} }
@ -6713,10 +6713,8 @@ static int32_t tDecodeSSubmitTbData(SDecoder *pCoder, SSubmitTbData *pSubmitTbDa
for (int32_t i = 0; i < nRows; i++) { for (int32_t i = 0; i < nRows; i++) {
SRow **ppRow = taosArrayReserve(pSubmitTbData->aRowP, 1); SRow **ppRow = taosArrayReserve(pSubmitTbData->aRowP, 1);
if (tDecodeBinary(pCoder, (uint8_t **)ppRow, NULL) < 0) { *ppRow = (SRow *)(pCoder->data + pCoder->pos);
code = TSDB_CODE_INVALID_MSG; pCoder->pos += (*ppRow)->len;
goto _exit;
}
} }
} }
@ -6822,13 +6820,18 @@ _exit:
if (code) { if (code) {
*ppReq = NULL; *ppReq = NULL;
if (pReq) { if (pReq) {
// todo: do other clear if (pReq->aCreateTbReq) {
taosArrayDestroy(pReq->aCreateTbReq);
}
if (pReq->aSubmitTbData) {
taosArrayDestroy(pReq->aSubmitTbData);
}
taosMemoryFree(pReq); taosMemoryFree(pReq);
} }
} else { } else {
*ppReq = pReq; *ppReq = pReq;
} }
return 0; return code;
} }
void tDestroySSubmitTbData(SSubmitTbData *pTbData) { void tDestroySSubmitTbData(SSubmitTbData *pTbData) {

12
source/dnode/vnode/src/inc/tsdb.h
View File

@ -121,7 +121,7 @@ int32_t tRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema);
int32_t tRowMergerInit(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema); int32_t tRowMergerInit(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema);
void tRowMergerClear(SRowMerger *pMerger); void tRowMergerClear(SRowMerger *pMerger);
int32_t tRowMerge(SRowMerger *pMerger, TSDBROW *pRow); int32_t tRowMerge(SRowMerger *pMerger, TSDBROW *pRow);
int32_t tRowMergerGetRow(SRowMerger *pMerger, STSRow **ppRow); int32_t tRowMergerGetRow(SRowMerger *pMerger, SRow **ppRow);
// TABLEID // TABLEID
int32_t tTABLEIDCmprFn(const void *p1, const void *p2); int32_t tTABLEIDCmprFn(const void *p1, const void *p2);
// TSDBKEY // TSDBKEY
@ -336,7 +336,7 @@ typedef struct SMemSkipListNode SMemSkipListNode;
struct SMemSkipListNode { struct SMemSkipListNode {
int8_t level; int8_t level;
int64_t version; int64_t version;
STSRow *pTSRow; SRow *pTSRow;
SMemSkipListNode *forwards[0]; SMemSkipListNode *forwards[0];
}; };
typedef struct SMemSkipList { typedef struct SMemSkipList {
@ -380,7 +380,7 @@ struct TSDBROW {
union { union {
struct { struct {
int64_t version; int64_t version;
STSRow *pTSRow; SRow *pTSRow;
}; };
struct { struct {
SBlockData *pBlockData; SBlockData *pBlockData;
@ -739,8 +739,8 @@ typedef struct {
int32_t tsdbOpenCache(STsdb *pTsdb); int32_t tsdbOpenCache(STsdb *pTsdb);
void tsdbCloseCache(STsdb *pTsdb); void tsdbCloseCache(STsdb *pTsdb);
int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb *pTsdb); int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, SRow *row, STsdb *pTsdb);
int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, STSRow *row, bool dup); int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, SRow *row, bool dup);
int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, SCacheRowsReader *pr, LRUHandle **h); int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, SCacheRowsReader *pr, LRUHandle **h);
int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, SCacheRowsReader *pr, LRUHandle **h); int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, SCacheRowsReader *pr, LRUHandle **h);
int32_t tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h); int32_t tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h);
@ -752,7 +752,7 @@ int32_t tsdbCacheDelete(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey);
void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity); void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity);
size_t tsdbCacheGetCapacity(SVnode *pVnode); size_t tsdbCacheGetCapacity(SVnode *pVnode);
int32_t tsdbCacheLastArray2Row(SArray *pLastArray, STSRow **ppRow, STSchema *pSchema); // int32_t tsdbCacheLastArray2Row(SArray *pLastArray, STSRow **ppRow, STSchema *pSchema);
// ========== inline functions ========== // ========== inline functions ==========
static FORCE_INLINE int32_t tsdbKeyCmprFn(const void *p1, const void *p2) { static FORCE_INLINE int32_t tsdbKeyCmprFn(const void *p1, const void *p2) {

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

@ -160,8 +160,7 @@ int32_t tsdbRollbackCommit(STsdb* pTsdb);
int32_t tsdbDoRetention(STsdb* pTsdb, int64_t now); int32_t tsdbDoRetention(STsdb* pTsdb, int64_t now);
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq* pMsg); int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq* pMsg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp); int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitTbData* pSubmitTbData, SSubmitBlkRsp* pRsp);
SSubmitBlkRsp* pRsp);
int32_t tsdbDeleteTableData(STsdb* pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey); int32_t tsdbDeleteTableData(STsdb* pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey);
int32_t tsdbSetKeepCfg(STsdb* pTsdb, STsdbCfg* pCfg); int32_t tsdbSetKeepCfg(STsdb* pTsdb, STsdbCfg* pCfg);

42
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -190,7 +190,7 @@ int32_t tsdbCacheDelete(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey) {
return code; return code;
} }
int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, STSRow *row, bool dup) { int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, SRow *row, bool dup) {
int32_t code = 0; int32_t code = 0;
STSRow *cacheRow = NULL; STSRow *cacheRow = NULL;
char key[32] = {0}; char key[32] = {0};
@ -222,7 +222,7 @@ int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, ST
SColVal *tColVal = &tTsVal1->colVal; SColVal *tColVal = &tTsVal1->colVal;
SColVal colVal = {0}; SColVal colVal = {0};
tTSRowGetVal(row, pTSchema, iCol, &colVal); tRowGet(row, pTSchema, iCol, &colVal);
if (!COL_VAL_IS_NONE(&colVal)) { if (!COL_VAL_IS_NONE(&colVal)) {
if (keyTs == tTsVal1->ts && !COL_VAL_IS_NONE(tColVal)) { if (keyTs == tTsVal1->ts && !COL_VAL_IS_NONE(tColVal)) {
invalidate = true; invalidate = true;
@ -316,7 +316,7 @@ int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, ST
return code; return code;
} }
int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb *pTsdb) { int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, SRow *row, STsdb *pTsdb) {
int32_t code = 0; int32_t code = 0;
STSRow *cacheRow = NULL; STSRow *cacheRow = NULL;
char key[32] = {0}; char key[32] = {0};
@ -348,7 +348,7 @@ int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb
SColVal *tColVal = &tTsVal1->colVal; SColVal *tColVal = &tTsVal1->colVal;
SColVal colVal = {0}; SColVal colVal = {0};
tTSRowGetVal(row, pTSchema, iCol, &colVal); tRowGet(row, pTSchema, iCol, &colVal);
if (!COL_VAL_IS_VALUE(&colVal)) { if (!COL_VAL_IS_VALUE(&colVal)) {
if (keyTs == tTsVal1->ts && COL_VAL_IS_VALUE(tColVal)) { if (keyTs == tTsVal1->ts && COL_VAL_IS_VALUE(tColVal)) {
invalidate = true; invalidate = true;
@ -1456,29 +1456,29 @@ int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, SCacheRowsReader *
return code; return code;
} }
int32_t tsdbCacheLastArray2Row(SArray *pLastArray, STSRow **ppRow, STSchema *pTSchema) { // int32_t tsdbCacheLastArray2Row(SArray *pLastArray, STSRow **ppRow, STSchema *pTSchema) {
int32_t code = 0; // int32_t code = 0;
int16_t nCol = taosArrayGetSize(pLastArray); // int16_t nCol = taosArrayGetSize(pLastArray);
SArray *pColArray = taosArrayInit(nCol, sizeof(SColVal)); // SArray *pColArray = taosArrayInit(nCol, sizeof(SColVal));
for (int16_t iCol = 0; iCol < nCol; ++iCol) { // for (int16_t iCol = 0; iCol < nCol; ++iCol) {
SLastCol *tTsVal = (SLastCol *)taosArrayGet(pLastArray, iCol); // SLastCol *tTsVal = (SLastCol *)taosArrayGet(pLastArray, iCol);
SColVal *tColVal = &tTsVal->colVal; // SColVal *tColVal = &tTsVal->colVal;
taosArrayPush(pColArray, tColVal); // taosArrayPush(pColArray, tColVal);
} // }
code = tdSTSRowNew(pColArray, pTSchema, ppRow); // code = tdSTSRowNew(pColArray, pTSchema, ppRow);
if (code) goto _err; // if (code) goto _err;
taosArrayDestroy(pColArray); // taosArrayDestroy(pColArray);
return code; // return code;
_err: // _err:
taosArrayDestroy(pColArray); // taosArrayDestroy(pColArray);
return code; // return code;
} // }
int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, SCacheRowsReader *pr, LRUHandle **handle) { int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, SCacheRowsReader *pr, LRUHandle **handle) {
int32_t code = 0; int32_t code = 0;

57
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -29,7 +29,7 @@
static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags); static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags);
static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData); static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData);
static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version, static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version,
SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp); SSubmitTbData *pSubmitTbData, SSubmitBlkRsp *pRsp);
int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) { int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) {
int32_t code = 0; int32_t code = 0;
@ -95,13 +95,12 @@ STbData *tsdbGetTbDataFromMemTable(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t
return pTbData; return pTbData;
} }
int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitTbData *pSubmitTbData, SSubmitBlkRsp *pRsp) {
SSubmitBlkRsp *pRsp) {
int32_t code = 0; int32_t code = 0;
SMemTable *pMemTable = pTsdb->mem; SMemTable *pMemTable = pTsdb->mem;
STbData *pTbData = NULL; STbData *pTbData = NULL;
tb_uid_t suid = pMsgIter->suid; tb_uid_t suid = pSubmitTbData->suid;
tb_uid_t uid = pMsgIter->uid; tb_uid_t uid = pSubmitTbData->uid;
SMetaInfo info; SMetaInfo info;
code = metaGetInfo(pTsdb->pVnode->pMeta, uid, &info, NULL); code = metaGetInfo(pTsdb->pVnode->pMeta, uid, &info, NULL);
@ -116,14 +115,14 @@ int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitMsgIter *pMsgI
if (info.suid) { if (info.suid) {
metaGetInfo(pTsdb->pVnode->pMeta, info.suid, &info, NULL); metaGetInfo(pTsdb->pVnode->pMeta, info.suid, &info, NULL);
} }
if (pMsgIter->sversion != info.skmVer) { if (pSubmitTbData->sver != info.skmVer) {
tsdbError("vgId:%d, req sver:%d, skmVer:%d suid:%" PRId64 " uid:%" PRId64, TD_VID(pTsdb->pVnode), tsdbError("vgId:%d, req sver:%d, skmVer:%d suid:%" PRId64 " uid:%" PRId64, TD_VID(pTsdb->pVnode),
pMsgIter->sversion, info.skmVer, suid, uid); pSubmitTbData->sver, info.skmVer, suid, uid);
code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER; code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
goto _err; goto _err;
} }
pRsp->sver = info.skmVer; if (pRsp) pRsp->sver = info.skmVer;
// create/get STbData to op // create/get STbData to op
code = tsdbGetOrCreateTbData(pMemTable, suid, uid, &pTbData); code = tsdbGetOrCreateTbData(pMemTable, suid, uid, &pTbData);
@ -132,7 +131,7 @@ int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitMsgIter *pMsgI
} }
// do insert impl // do insert impl
code = tsdbInsertTableDataImpl(pMemTable, pTbData, version, pMsgIter, pBlock, pRsp); code = tsdbInsertTableDataImpl(pMemTable, pTbData, version, pSubmitTbData, pRsp);
if (code) { if (code) {
goto _err; goto _err;
} }
@ -468,8 +467,8 @@ static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
return level; return level;
} }
static int32_t tbDataDoPut(SMemTable *pMemTable, STbData *pTbData, SMemSkipListNode **pos, int64_t version, static int32_t tbDataDoPut(SMemTable *pMemTable, STbData *pTbData, SMemSkipListNode **pos, int64_t version, SRow *pRow,
STSRow *pRow, int8_t forward) { int8_t forward) {
int32_t code = 0; int32_t code = 0;
int8_t level; int8_t level;
SMemSkipListNode *pNode; SMemSkipListNode *pNode;
@ -538,23 +537,21 @@ _exit:
} }
static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version, static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version,
SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp) { SSubmitTbData *pSubmitTbData, SSubmitBlkRsp *pRsp) {
int32_t code = 0; int32_t code = 0;
SSubmitBlkIter blkIter = {0}; // SSubmitBlkIter blkIter = {0};
TSDBKEY key = {.version = version}; TSDBKEY key = {.version = version};
SMemSkipListNode *pos[SL_MAX_LEVEL]; SMemSkipListNode *pos[SL_MAX_LEVEL];
TSDBROW row = tsdbRowFromTSRow(version, NULL); TSDBROW row = tsdbRowFromTSRow(version, NULL);
int32_t nRow = 0; int32_t nRow = taosArrayGetSize(pSubmitTbData->aRowP);
STSRow *pLastRow = NULL; int32_t iRow = 0;
SRow *pLastRow = NULL;
tInitSubmitBlkIter(pMsgIter, pBlock, &blkIter);
// backward put first data // backward put first data
row.pTSRow = tGetSubmitBlkNext(&blkIter); row.pTSRow = taosArrayGetP(pSubmitTbData->aRowP, iRow);
if (row.pTSRow == NULL) return code;
key.ts = row.pTSRow->ts; key.ts = row.pTSRow->ts;
nRow++; iRow++;
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_BACKWARD); tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_BACKWARD);
code = tbDataDoPut(pMemTable, pTbData, pos, version, row.pTSRow, 0); code = tbDataDoPut(pMemTable, pTbData, pos, version, row.pTSRow, 0);
if (code) { if (code) {
@ -566,17 +563,19 @@ static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, i
pLastRow = row.pTSRow; pLastRow = row.pTSRow;
// forward put rest data // forward put rest data
row.pTSRow = tGetSubmitBlkNext(&blkIter); if (iRow < nRow) {
if (row.pTSRow) {
for (int8_t iLevel = pos[0]->level; iLevel < pTbData->sl.maxLevel; iLevel++) { for (int8_t iLevel = pos[0]->level; iLevel < pTbData->sl.maxLevel; iLevel++) {
pos[iLevel] = SL_NODE_BACKWARD(pos[iLevel], iLevel); pos[iLevel] = SL_NODE_BACKWARD(pos[iLevel], iLevel);
} }
do {
while (iRow < nRow) {
row.pTSRow = taosArrayGetP(pSubmitTbData->aRowP, iRow);
key.ts = row.pTSRow->ts; key.ts = row.pTSRow->ts;
nRow++;
if (SL_NODE_FORWARD(pos[0], 0) != pTbData->sl.pTail) { if (SL_NODE_FORWARD(pos[0], 0) != pTbData->sl.pTail) {
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS); tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS);
} }
code = tbDataDoPut(pMemTable, pTbData, pos, version, row.pTSRow, 1); code = tbDataDoPut(pMemTable, pTbData, pos, version, row.pTSRow, 1);
if (code) { if (code) {
goto _err; goto _err;
@ -584,8 +583,8 @@ static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, i
pLastRow = row.pTSRow; pLastRow = row.pTSRow;
row.pTSRow = tGetSubmitBlkNext(&blkIter); iRow++;
} while (row.pTSRow); }
} }
if (key.ts >= pTbData->maxKey) { if (key.ts >= pTbData->maxKey) {
@ -607,8 +606,8 @@ static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, i
pMemTable->maxKey = TMAX(pMemTable->maxKey, pTbData->maxKey); pMemTable->maxKey = TMAX(pMemTable->maxKey, pTbData->maxKey);
pMemTable->nRow += nRow; pMemTable->nRow += nRow;
pRsp->numOfRows = nRow; if (pRsp) pRsp->numOfRows = nRow;
pRsp->affectedRows = nRow; if (pRsp) pRsp->affectedRows = nRow;
return code; return code;

58
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -186,7 +186,7 @@ static int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STabl
SRowMerger* pMerger, SVersionRange* pVerRange); SRowMerger* pMerger, SVersionRange* pVerRange);
static int32_t doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger, static int32_t doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger,
STsdbReader* pReader); STsdbReader* pReader);
static int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, static int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, SRow* pTSRow,
STableBlockScanInfo* pInfo); STableBlockScanInfo* pInfo);
static int32_t doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData, static int32_t doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData,
int32_t rowIndex); int32_t rowIndex);
@ -194,10 +194,10 @@ static void setComposedBlockFlag(STsdbReader* pReader, bool composed);
static bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order, static bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order,
SVersionRange* pVerRange); SVersionRange* pVerRange);
static int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, static int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, SRow** pTSRow,
STSRow** pTSRow, STsdbReader* pReader, bool* freeTSRow); STsdbReader* pReader, bool* freeTSRow);
static int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo, static int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo,
STsdbReader* pReader, STSRow** pTSRow); STsdbReader* pReader, SRow** pTSRow);
static int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key, static int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
STsdbReader* pReader); STsdbReader* pReader);
@ -242,7 +242,7 @@ static int32_t setColumnIdSlotList(SBlockLoadSuppInfo* pSupInfo, SSDataBlock* pB
static void updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInfo) { static void updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInfo) {
int32_t i = 0, j = 0; int32_t i = 0, j = 0;
while(i < pSchema->numOfCols && j < pSupInfo->numOfCols) { while (i < pSchema->numOfCols && j < pSupInfo->numOfCols) {
STColumn* pTCol = &pSchema->columns[i]; STColumn* pTCol = &pSchema->columns[i];
if (pTCol->colId == pSupInfo->colIds[j]) { if (pTCol->colId == pSupInfo->colIds[j]) {
if (!IS_BSMA_ON(pTCol)) { if (!IS_BSMA_ON(pTCol)) {
@ -305,7 +305,8 @@ static void* getPosInBlockInfoBuf(SBlockInfoBuf* pBuf, int32_t index) {
} }
// NOTE: speedup the whole processing by preparing the buffer for STableBlockScanInfo in batch model // NOTE: speedup the whole processing by preparing the buffer for STableBlockScanInfo in batch model
static SHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, SBlockInfoBuf* pBuf, const STableKeyInfo* idList, int32_t numOfTables) { static SHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, SBlockInfoBuf* pBuf, const STableKeyInfo* idList,
int32_t numOfTables) {
// allocate buffer in order to load data blocks from file // allocate buffer in order to load data blocks from file
// todo use simple hash instead, optimize the memory consumption // todo use simple hash instead, optimize the memory consumption
SHashObj* pTableMap = SHashObj* pTableMap =
@ -754,7 +755,6 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockN
numOfTables, pBlockNum->numOfBlocks, numOfQTable, pBlockNum->numOfLastFiles, sizeInDisk / 1000.0, el, numOfTables, pBlockNum->numOfBlocks, numOfQTable, pBlockNum->numOfLastFiles, sizeInDisk / 1000.0, el,
pReader->idStr); pReader->idStr);
pReader->cost.numOfBlocks += total; pReader->cost.numOfBlocks += total;
pReader->cost.headFileLoadTime += el; pReader->cost.headFileLoadTime += el;
@ -951,7 +951,7 @@ static void copyPrimaryTsCol(const SBlockData* pBlockData, SFileBlockDumpInfo* p
// a faster version of copy procedure. // a faster version of copy procedure.
static void copyNumericCols(const SColData* pData, SFileBlockDumpInfo* pDumpInfo, SColumnInfoData* pColData, static void copyNumericCols(const SColData* pData, SFileBlockDumpInfo* pDumpInfo, SColumnInfoData* pColData,
int32_t dumpedRows, bool asc) { int32_t dumpedRows, bool asc) {
uint8_t* p = NULL; uint8_t* p = NULL;
if (asc) { if (asc) {
p = pData->pData + tDataTypes[pData->type].bytes * pDumpInfo->rowIndex; p = pData->pData + tDataTypes[pData->type].bytes * pDumpInfo->rowIndex;
@ -960,7 +960,7 @@ static void copyNumericCols(const SColData* pData, SFileBlockDumpInfo* pDumpInfo
p = pData->pData + tDataTypes[pData->type].bytes * startIndex; p = pData->pData + tDataTypes[pData->type].bytes * startIndex;
} }
int32_t step = asc? 1:-1; int32_t step = asc ? 1 : -1;
// make sure it is aligned to 8bit // make sure it is aligned to 8bit
ASSERT((((uint64_t)pColData->pData) & (0x8 - 1)) == 0); ASSERT((((uint64_t)pColData->pData) & (0x8 - 1)) == 0);
@ -970,12 +970,11 @@ static void copyNumericCols(const SColData* pData, SFileBlockDumpInfo* pDumpInfo
// 2. reverse the array list in case of descending order scan data block // 2. reverse the array list in case of descending order scan data block
if (!asc) { if (!asc) {
switch(pColData->info.type) { switch (pColData->info.type) {
case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_TIMESTAMP:
case TSDB_DATA_TYPE_DOUBLE: case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_BIGINT: case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_UBIGINT: case TSDB_DATA_TYPE_UBIGINT: {
{
int32_t mid = dumpedRows >> 1u; int32_t mid = dumpedRows >> 1u;
int64_t* pts = (int64_t*)pColData->pData; int64_t* pts = (int64_t*)pColData->pData;
for (int32_t j = 0; j < mid; ++j) { for (int32_t j = 0; j < mid; ++j) {
@ -989,7 +988,7 @@ static void copyNumericCols(const SColData* pData, SFileBlockDumpInfo* pDumpInfo
case TSDB_DATA_TYPE_BOOL: case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT: case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_UTINYINT: { case TSDB_DATA_TYPE_UTINYINT: {
int32_t mid = dumpedRows >> 1u; int32_t mid = dumpedRows >> 1u;
int8_t* pts = (int8_t*)pColData->pData; int8_t* pts = (int8_t*)pColData->pData;
for (int32_t j = 0; j < mid; ++j) { for (int32_t j = 0; j < mid; ++j) {
int8_t t = pts[j]; int8_t t = pts[j];
@ -1730,7 +1729,7 @@ static FORCE_INLINE STSchema* doGetSchemaForTSRow(int32_t sversion, STsdbReader*
static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, TSDBROW* pRow, static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, TSDBROW* pRow,
SIterInfo* pIter, int64_t key, SLastBlockReader* pLastBlockReader) { SIterInfo* pIter, int64_t key, SLastBlockReader* pLastBlockReader) {
SRowMerger merge = {0}; SRowMerger merge = {0};
STSRow* pTSRow = NULL; SRow* pTSRow = NULL;
SBlockData* pBlockData = &pReader->status.fileBlockData; SBlockData* pBlockData = &pReader->status.fileBlockData;
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
@ -1876,7 +1875,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader,
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
int64_t tsLastBlock = getCurrentKeyInLastBlock(pLastBlockReader); int64_t tsLastBlock = getCurrentKeyInLastBlock(pLastBlockReader);
STSRow* pTSRow = NULL; SRow* pTSRow = NULL;
SRowMerger merge = {0}; SRowMerger merge = {0};
TSDBROW fRow = tMergeTreeGetRow(&pLastBlockReader->mergeTree); TSDBROW fRow = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
tsdbTrace("fRow ptr:%p, %d, uid:%" PRIu64 ", %s", fRow.pBlockData, fRow.iRow, pLastBlockReader->uid, pReader->idStr); tsdbTrace("fRow ptr:%p, %d, uid:%" PRIu64 ", %s", fRow.pBlockData, fRow.iRow, pLastBlockReader->uid, pReader->idStr);
@ -1953,7 +1952,7 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader
if (key < ts) { // imem, mem are all empty, file blocks (data blocks and last block) exist if (key < ts) { // imem, mem are all empty, file blocks (data blocks and last block) exist
return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader); return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
} else if (key == ts) { } else if (key == ts) {
STSRow* pTSRow = NULL; SRow* pTSRow = NULL;
SRowMerger merge = {0}; SRowMerger merge = {0};
int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema); int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
@ -1993,7 +1992,7 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader
static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData, static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData,
SLastBlockReader* pLastBlockReader) { SLastBlockReader* pLastBlockReader) {
SRowMerger merge = {0}; SRowMerger merge = {0};
STSRow* pTSRow = NULL; SRow* pTSRow = NULL;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
SArray* pDelList = pBlockScanInfo->delSkyline; SArray* pDelList = pBlockScanInfo->delSkyline;
@ -2354,7 +2353,7 @@ int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBloc
} else { } else {
TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex); TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
STSRow* pTSRow = NULL; SRow* pTSRow = NULL;
SRowMerger merge = {0}; SRowMerger merge = {0};
int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema); int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema);
@ -2436,7 +2435,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
TSDBKEY keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader); TSDBKEY keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader);
// it is a clean block, load it directly // it is a clean block, load it directly
if (isCleanFileDataBlock(pReader, pBlockInfo, pBlock, pBlockScanInfo, keyInBuf, pLastBlockReader) && if (isCleanFileDataBlock(pReader, pBlockInfo, pBlock, pBlockScanInfo, keyInBuf, pLastBlockReader) &&
pBlock->nRow <= pReader->capacity) { pBlock->nRow <= pReader->capacity) {
if (asc || ((!asc) && (!hasDataInLastBlock(pLastBlockReader)))) { if (asc || ((!asc) && (!hasDataInLastBlock(pLastBlockReader)))) {
copyBlockDataToSDataBlock(pReader, pBlockScanInfo); copyBlockDataToSDataBlock(pReader, pBlockScanInfo);
@ -3354,7 +3353,7 @@ int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockSc
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, STSRow** pTSRow, int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, SRow** pTSRow,
STsdbReader* pReader, bool* freeTSRow) { STsdbReader* pReader, bool* freeTSRow) {
TSDBROW* pNextRow = NULL; TSDBROW* pNextRow = NULL;
TSDBROW current = *pRow; TSDBROW current = *pRow;
@ -3423,7 +3422,7 @@ int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter,
} }
int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader,
STSRow** pTSRow) { SRow** pTSRow) {
SRowMerger merge = {0}; SRowMerger merge = {0};
TSDBKEY k = TSDBROW_KEY(pRow); TSDBKEY k = TSDBROW_KEY(pRow);
@ -3476,7 +3475,7 @@ int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* p
return code; return code;
} }
int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STSRow** pTSRow, int64_t endKey, int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, SRow** pTSRow, int64_t endKey,
bool* freeTSRow) { bool* freeTSRow) {
TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader); TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pBlockScanInfo->delSkyline, pReader); TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pBlockScanInfo->delSkyline, pReader);
@ -3533,8 +3532,7 @@ int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pR
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, SRow* pTSRow, STableBlockScanInfo* pScanInfo) {
STableBlockScanInfo* pScanInfo) {
int32_t numOfRows = pBlock->info.rows; int32_t numOfRows = pBlock->info.rows;
int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock); int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock);
int64_t uid = pScanInfo->uid; int64_t uid = pScanInfo->uid;
@ -3556,7 +3554,7 @@ int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow*
col_id_t colId = pColInfoData->info.colId; col_id_t colId = pColInfoData->info.colId;
if (colId == pSchema->columns[j].colId) { if (colId == pSchema->columns[j].colId) {
tTSRowGetVal(pTSRow, pSchema, j, &colVal); tRowGet(pTSRow, pSchema, j, &colVal);
doCopyColVal(pColInfoData, numOfRows, i, &colVal, pSupInfo); doCopyColVal(pColInfoData, numOfRows, i, &colVal, pSupInfo);
i += 1; i += 1;
j += 1; j += 1;
@ -3633,8 +3631,8 @@ int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t e
SSDataBlock* pBlock = pReader->pResBlock; SSDataBlock* pBlock = pReader->pResBlock;
do { do {
STSRow* pTSRow = NULL; SRow* pTSRow = NULL;
bool freeTSRow = false; bool freeTSRow = false;
tsdbGetNextRowInMem(pBlockScanInfo, pReader, &pTSRow, endKey, &freeTSRow); tsdbGetNextRowInMem(pBlockScanInfo, pReader, &pTSRow, endKey, &freeTSRow);
if (pTSRow == NULL) { if (pTSRow == NULL) {
break; break;
@ -3789,7 +3787,7 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableL
updateBlockSMAInfo(pReader->pSchema, &pReader->suppInfo); updateBlockSMAInfo(pReader->pSchema, &pReader->suppInfo);
} }
STsdbReader* p = (pReader->innerReader[0] != NULL)? pReader->innerReader[0]:pReader; STsdbReader* p = (pReader->innerReader[0] != NULL) ? pReader->innerReader[0] : pReader;
pReader->status.pTableMap = createDataBlockScanInfo(p, &pReader->blockInfoBuf, pTableList, numOfTables); pReader->status.pTableMap = createDataBlockScanInfo(p, &pReader->blockInfoBuf, pTableList, numOfTables);
if (pReader->status.pTableMap == NULL) { if (pReader->status.pTableMap == NULL) {
tsdbReaderClose(p); tsdbReaderClose(p);
@ -4123,7 +4121,7 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS
if (pSup->colIds[j] == PRIMARYKEY_TIMESTAMP_COL_ID) { if (pSup->colIds[j] == PRIMARYKEY_TIMESTAMP_COL_ID) {
taosArrayPush(pNewAggList, &pSup->tsColAgg); taosArrayPush(pNewAggList, &pSup->tsColAgg);
} else { } else {
// all date in this block are null // all date in this block are null
SColumnDataAgg nullColAgg = {.colId = pSup->colIds[j], .numOfNull = pBlock->nRow}; SColumnDataAgg nullColAgg = {.colId = pSup->colIds[j], .numOfNull = pBlock->nRow};
taosArrayPush(pNewAggList, &nullColAgg); taosArrayPush(pNewAggList, &nullColAgg);
} }
@ -4135,7 +4133,7 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS
taosArrayAddAll(pSup->pColAgg, pNewAggList); taosArrayAddAll(pSup->pColAgg, pNewAggList);
size_t num = taosArrayGetSize(pSup->pColAgg); size_t num = taosArrayGetSize(pSup->pColAgg);
for(int32_t k = 0; k < num; ++k) { for (int32_t k = 0; k < num; ++k) {
pSup->plist[k] = taosArrayGet(pSup->pColAgg, k); pSup->plist[k] = taosArrayGet(pSup->pColAgg, k);
} }

31
source/dnode/vnode/src/tsdb/tsdbUtil.c
View File

@ -573,7 +573,7 @@ void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *
ASSERT(iCol > 0); ASSERT(iCol > 0);
if (pRow->type == 0) { if (pRow->type == 0) {
tTSRowGetVal(pRow->pTSRow, pTSchema, iCol, pColVal); tRowGet(pRow->pTSRow, pTSchema, iCol, pColVal);
} else if (pRow->type == 1) { } else if (pRow->type == 1) {
SColData *pColData; SColData *pColData;
@ -621,7 +621,7 @@ void tsdbRowIterInit(STSDBRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) {
SColVal *tsdbRowIterNext(STSDBRowIter *pIter) { SColVal *tsdbRowIterNext(STSDBRowIter *pIter) {
if (pIter->pRow->type == 0) { if (pIter->pRow->type == 0) {
if (pIter->i < pIter->pTSchema->numOfCols) { if (pIter->i < pIter->pTSchema->numOfCols) {
tTSRowGetVal(pIter->pRow->pTSRow, pIter->pTSchema, pIter->i, &pIter->colVal); tRowGet(pIter->pRow->pTSRow, pIter->pTSchema, pIter->i, &pIter->colVal);
pIter->i++; pIter->i++;
return &pIter->colVal; return &pIter->colVal;
@ -807,12 +807,8 @@ _exit:
return code; return code;
} }
int32_t tRowMergerGetRow(SRowMerger *pMerger, STSRow **ppRow) { int32_t tRowMergerGetRow(SRowMerger *pMerger, SRow **ppRow) {
int32_t code = 0; return tRowBuild(pMerger->pArray, pMerger->pTSchema, ppRow);
code = tdSTSRowNew(pMerger->pArray, pMerger->pTSchema, ppRow);
return code;
} }
// delete skyline ====================================================== // delete skyline ======================================================
@ -1247,15 +1243,16 @@ int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTS
SColVal cv = {0}; SColVal cv = {0};
if (pRow->type == 0) { if (pRow->type == 0) {
if (TD_IS_TP_ROW(pRow->pTSRow)) { ASSERT(0);
code = tBlockDataAppendTPRow(pBlockData, pRow->pTSRow, pTSchema); // if (TD_IS_TP_ROW(pRow->pTSRow)) {
if (code) goto _err; // code = tBlockDataAppendTPRow(pBlockData, pRow->pTSRow, pTSchema);
} else if (TD_IS_KV_ROW(pRow->pTSRow)) { // if (code) goto _err;
code = tBlockDataAppendKVRow(pBlockData, pRow->pTSRow, pTSchema); // } else if (TD_IS_KV_ROW(pRow->pTSRow)) {
if (code) goto _err; // code = tBlockDataAppendKVRow(pBlockData, pRow->pTSRow, pTSchema);
} else { // if (code) goto _err;
ASSERT(0); // } else {
} // ASSERT(0);
// }
} else { } else {
code = tBlockDataAppendBlockRow(pBlockData, pRow->pBlockData, pRow->iRow); code = tBlockDataAppendBlockRow(pBlockData, pRow->pBlockData, pRow->iRow);
if (code) goto _err; if (code) goto _err;

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

@ -77,52 +77,37 @@ int32_t vnodePreProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) {
tDecoderClear(&dc); tDecoderClear(&dc);
} break; } break;
case TDMT_VND_SUBMIT: { case TDMT_VND_SUBMIT: {
SSubmitMsgIter msgIter = {0}; tDecoderInit(&dc, (uint8_t *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead));
SSubmitReq *pSubmitReq = (SSubmitReq *)pMsg->pCont; tStartDecode(&dc);
SSubmitBlk *pBlock = NULL;
int64_t ctime = taosGetTimestampMs();
tb_uid_t uid;
if (tInitSubmitMsgIter(pSubmitReq, &msgIter) < 0) { int32_t flag;
code = terrno; tDecodeI32v(&dc, &flag);
goto _err;
}
for (;;) { if (flag & SUBMIT_REQ_AUTO_CREATE_TABLE) {
tGetSubmitMsgNext(&msgIter, &pBlock); int64_t ctime = taosGetTimestampMs();
if (pBlock == NULL) break; int64_t nReq;
int64_t uid;
if (msgIter.schemaLen > 0) { tDecodeI64v(&dc, &nReq);
for (int64_t iReq; iReq < nReq; iReq++) {
char *name = NULL; char *name = NULL;
tDecoderInit(&dc, pBlock->data, msgIter.schemaLen); tStartDecode(&dc);
if (tStartDecode(&dc) < 0) { tDecodeI32v(&dc, NULL);
code = TSDB_CODE_INVALID_MSG; tDecodeCStr(&dc, &name);
return code;
}
if (tDecodeI32v(&dc, NULL) < 0) {
code = TSDB_CODE_INVALID_MSG;
return code;
}
if (tDecodeCStr(&dc, &name) < 0) {
code = TSDB_CODE_INVALID_MSG;
return code;
}
uid = metaGetTableEntryUidByName(pVnode->pMeta, name); uid = metaGetTableEntryUidByName(pVnode->pMeta, name);
if (uid == 0) { if (uid == 0) {
uid = tGenIdPI64(); uid = tGenIdPI64();
} }
*(int64_t *)(dc.data + dc.pos) = uid; *(int64_t *)(dc.data + dc.pos) = uid;
*(int64_t *)(dc.data + dc.pos + 8) = ctime; *(int64_t *)(dc.data + dc.pos + 8) = ctime;
pBlock->uid = htobe64(uid);
tEndDecode(&dc); tEndDecode(&dc);
tDecoderClear(&dc);
} }
} }
tEndDecode(&dc);
} break; } break;
case TDMT_VND_DELETE: { case TDMT_VND_DELETE: {
int32_t size; int32_t size;
@ -855,6 +840,28 @@ static int32_t vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const
} }
static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) { static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
#if 1
SDecoder dc = {0};
SSubmitReq2 *pSubmitReq = NULL;
tDecoderInit(&dc, (char *)pReq + sizeof(SMsgHead), len - sizeof(SMsgHead));
tDecodeSSubmitReq2(&dc, &pSubmitReq);
if (pSubmitReq->flag & SUBMIT_REQ_AUTO_CREATE_TABLE) {
// todo
ASSERT(0);
}
for (int32_t iSubmitTbData = 0; iSubmitTbData < taosArrayGetSize(pSubmitReq->aSubmitTbData); iSubmitTbData++) {
SSubmitTbData *pSubmitTbData = taosArrayGet(pSubmitReq->aSubmitTbData, iSubmitTbData);
int32_t code = tsdbInsertTableData(pVnode->pTsdb, version, pSubmitTbData, NULL /*todo*/);
if (code) {
// todo
}
}
#else
SSubmitReq *pSubmitReq = (SSubmitReq *)pReq; SSubmitReq *pSubmitReq = (SSubmitReq *)pReq;
SSubmitRsp submitRsp = {0}; SSubmitRsp submitRsp = {0};
SSubmitMsgIter msgIter = {0}; SSubmitMsgIter msgIter = {0};
@ -1007,6 +1014,8 @@ _exit:
vDebug("vgId:%d, submit success, index:%" PRId64, pVnode->config.vgId, version); vDebug("vgId:%d, submit success, index:%" PRId64, pVnode->config.vgId, version);
return 0; return 0;
#endif
return 0;
} }
static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) { static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {