diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 3889784f2f..763ff61e80 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -672,7 +672,6 @@ typedef struct { SArray* pArray; // Array of SUseDbRsp } SUseDbBatchRsp; - int32_t tSerializeSUseDbBatchRsp(void* buf, int32_t bufLen, SUseDbBatchRsp* pRsp); int32_t tDeserializeSUseDbBatchRsp(void* buf, int32_t bufLen, SUseDbBatchRsp* pRsp); void tFreeSUseDbBatchRsp(SUseDbBatchRsp* pRsp); @@ -685,7 +684,6 @@ int32_t tSerializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp int32_t tDeserializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp* pRsp); void tFreeSUserAuthBatchRsp(SUserAuthBatchRsp* pRsp); - typedef struct { char db[TSDB_DB_FNAME_LEN]; } SCompactDbReq; @@ -1554,7 +1552,9 @@ typedef struct SVDropStbReq { int32_t tEncodeSVDropStbReq(SCoder* pCoder, const SVDropStbReq* pReq); int32_t tDecodeSVDropStbReq(SCoder* pCoder, SVDropStbReq* pReq); +#define TD_CREATE_IF_NOT_EXISTS 0x1 typedef struct SVCreateTbReq { + int32_t flags; tb_uid_t uid; int64_t ctime; const char* name; diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 6278b52a04..85845dd5c4 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -1255,7 +1255,6 @@ int32_t tDeserializeSGetUserAuthRspImpl(SCoder *pDecoder, SGetUserAuthRsp *pRsp) return 0; } - int32_t tDeserializeSGetUserAuthRsp(void *buf, int32_t bufLen, SGetUserAuthRsp *pRsp) { SCoder decoder = {0}; tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER); @@ -2091,7 +2090,7 @@ void tFreeSUseDbBatchRsp(SUseDbBatchRsp *pRsp) { taosArrayDestroy(pRsp->pArray); } -int32_t tSerializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp* pRsp){ +int32_t tSerializeSUserAuthBatchRsp(void *buf, int32_t bufLen, SUserAuthBatchRsp *pRsp) { SCoder encoder = {0}; tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER); @@ -2110,7 +2109,7 @@ int32_t tSerializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp return tlen; } -int32_t tDeserializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp* pRsp){ +int32_t tDeserializeSUserAuthBatchRsp(void *buf, int32_t bufLen, SUserAuthBatchRsp *pRsp) { SCoder decoder = {0}; tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER); @@ -2136,7 +2135,7 @@ int32_t tDeserializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchR return 0; } -void tFreeSUserAuthBatchRsp(SUserAuthBatchRsp* pRsp){ +void tFreeSUserAuthBatchRsp(SUserAuthBatchRsp *pRsp) { int32_t numOfBatch = taosArrayGetSize(pRsp->pArray); for (int32_t i = 0; i < numOfBatch; ++i) { SGetUserAuthRsp *pUserAuthRsp = taosArrayGet(pRsp->pArray, i); @@ -2146,7 +2145,6 @@ void tFreeSUserAuthBatchRsp(SUserAuthBatchRsp* pRsp){ taosArrayDestroy(pRsp->pArray); } - int32_t tSerializeSDbCfgReq(void *buf, int32_t bufLen, SDbCfgReq *pReq) { SCoder encoder = {0}; tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER); @@ -3746,6 +3744,7 @@ STSchema *tdGetSTSChemaFromSSChema(SSchema **pSchema, int32_t nCols) { int tEncodeSVCreateTbReq(SCoder *pCoder, const SVCreateTbReq *pReq) { if (tStartEncode(pCoder) < 0) return -1; + if (tEncodeI32v(pCoder, pReq->flags) < 0) return -1; if (tEncodeI64(pCoder, pReq->uid) < 0) return -1; if (tEncodeI64(pCoder, pReq->ctime) < 0) return -1; @@ -3771,6 +3770,7 @@ int tDecodeSVCreateTbReq(SCoder *pCoder, SVCreateTbReq *pReq) { if (tStartDecode(pCoder) < 0) return -1; + if (tDecodeI32v(pCoder, &pReq->flags) < 0) return -1; if (tDecodeI64(pCoder, &pReq->uid) < 0) return -1; if (tDecodeI64(pCoder, &pReq->ctime) < 0) return -1; diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index 397e074061..ffe176f85e 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -196,7 +196,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) { int8_t type; int64_t ctime; tb_uid_t suid; - int c, ret; + int c = 0, ret; // search & delete the name idx tdbDbcOpen(pMeta->pNameIdx, &pNameIdxc, &pMeta->txn); diff --git a/source/dnode/vnode/src/tsdb/tsdbMemTable2.c b/source/dnode/vnode/src/tsdb/tsdbMemTable2.c index ae0d4630d1..e7b733369a 100644 --- a/source/dnode/vnode/src/tsdb/tsdbMemTable2.c +++ b/source/dnode/vnode/src/tsdb/tsdbMemTable2.c @@ -15,10 +15,11 @@ #include "tsdb.h" -typedef struct SMemTable SMemTable; -typedef struct SMemData SMemData; -typedef struct SMemSkipList SMemSkipList; -typedef struct SMemSkipListCfg SMemSkipListCfg; +typedef struct SMemTable SMemTable; +typedef struct SMemData SMemData; +typedef struct SMemSkipList SMemSkipList; +typedef struct SMemSkipListNode SMemSkipListNode; +typedef struct SMemSkipListCurosr SMemSkipListCurosr; struct SMemTable { STsdb *pTsdb; @@ -32,15 +33,17 @@ struct SMemTable { SMemData **pBuckets; }; -struct SMemSkipListCfg { - int8_t maxLevel; - int32_t nKey; - int32_t nData; +struct SMemSkipListNode { + int8_t level; + SMemSkipListNode *forwards[]; }; struct SMemSkipList { - int8_t level; - uint32_t seed; + uint32_t seed; + int8_t maxLevel; + int8_t level; + int32_t size; + SMemSkipListNode pHead[]; }; struct SMemData { @@ -55,6 +58,20 @@ struct SMemData { SMemSkipList sl; }; +struct SMemSkipListCurosr { + SMemSkipList *pSl; + SMemSkipListNode *pNodeC; +}; + +#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2) +#define SL_NODE_HALF_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)) +#define SL_NODE_FORWARD(n, l) ((n)->forwards[l]) +#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)]) +#define SL_NODE_DATA(n) (&SL_NODE_BACKWARD(n, (n)->level)) + +#define SL_HEAD_NODE(sl) ((sl)->pHead) +#define SL_TAIL_NODE(sl) ((SMemSkipListNode *)&SL_NODE_FORWARD(SL_HEAD_NODE(sl), (sl)->maxLevel)) + // SMemTable int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) { SMemTable *pMemTb = NULL; @@ -76,6 +93,7 @@ int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) { pMemTb->pBuckets = taosMemoryCalloc(pMemTb->nBucket, sizeof(*pMemTb->pBuckets)); if (pMemTb->pBuckets == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; + taosMemoryFree(pMemTb); return -1; } @@ -83,8 +101,121 @@ int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) { return 0; } -int32_t tsdbMemTableDestroy2(STsdb *pTsdb, SMemTable *pMT) { - // TODO +int32_t tsdbMemTableDestroy2(STsdb *pTsdb, SMemTable *pMemTb) { + if (pMemTb) { + // loop to destroy the contents (todo) + taosMemoryFree(pMemTb->pBuckets); + taosMemoryFree(pMemTb); + } + return 0; +} + +int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *pSubmitBlk) { + SMemData *pMemData; + STsdb *pTsdb = pMemTb->pTsdb; + SVnode *pVnode = pTsdb->pVnode; + SVBufPool *pPool = pVnode->inUse; + int32_t hash; + int32_t tlen; + uint8_t buf[16]; + int32_t rlen; + const uint8_t *p; + SMemSkipListNode *pSlNode; + const STSRow *pTSRow; + SMemSkipListCurosr slc = {0}; + + // search hash + hash = (pSubmitBlk->suid + pSubmitBlk->uid) % pMemTb->nBucket; + for (pMemData = pMemTb->pBuckets[hash]; pMemData; pMemData = pMemData->pHashNext) { + if (pMemData->suid == pSubmitBlk->suid && pMemData->uid == pSubmitBlk->uid) break; + } + + // create pMemData if need + if (pMemData == NULL) { + int8_t maxLevel = pVnode->config.tsdbCfg.slLevel; + int32_t tsize = sizeof(*pMemData) + SL_NODE_HALF_SIZE(maxLevel) * 2; + SMemSkipListNode *pHead, *pTail; + + pMemData = vnodeBufPoolMalloc(pPool, tsize); + if (pMemData == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + pMemData->pHashNext = NULL; + pMemData->suid = pSubmitBlk->suid; + pMemData->uid = pSubmitBlk->uid; + pMemData->minKey = TSKEY_MAX; + pMemData->maxKey = TSKEY_MIN; + pMemData->minVer = -1; + pMemData->maxVer = -1; + pMemData->nRows = 0; + pMemData->sl.seed = taosRand(); + pMemData->sl.maxLevel = maxLevel; + pMemData->sl.level = 0; + pMemData->sl.size = 0; + pHead = SL_HEAD_NODE(&pMemData->sl); + pTail = SL_TAIL_NODE(&pMemData->sl); + pHead->level = maxLevel; + pTail->level = maxLevel; + for (int iLevel = 0; iLevel < maxLevel; iLevel++) { + SL_NODE_FORWARD(pHead, iLevel) = pTail; + SL_NODE_FORWARD(pTail, iLevel) = pHead; + } + + // add to MemTable + hash = (pMemData->suid + pMemData->uid) % pMemTb->nBucket; + pMemData->pHashNext = pMemTb->pBuckets[hash]; + pMemTb->pBuckets[hash] = pMemData; + pMemTb->nHash++; + } + +// loop to insert data to skiplist +#if 0 + tsdbMemSkipListCursorOpen(&slc, &pMemData->sl); + p = pSubmitBlk->pData; + for (;;) { + if (p - (uint8_t *)pSubmitBlk->pData >= pSubmitBlk->nData) break; + + const uint8_t *pt = p; + p = tGetBinary(p, &pTSRow, &rlen); + + // check the row (todo) + + // move the cursor to position to write (todo) + int32_t c; + tsdbMemSkipListCursorMoveTo(&slc, pTSRow, version, &c); + ASSERT(c); + + // encode row + int8_t level = tsdbMemSkipListRandLevel(&pMemData->sl); + int32_t tsize = SL_NODE_SIZE(level) + sizeof(version) + (p - pt); + pSlNode = vnodeBufPoolMalloc(pPool, tsize); + pSlNode->level = level; + + uint8_t *pData = SL_NODE_DATA(pSlNode); + *(int64_t *)pData = version; + pData += sizeof(version); + memcpy(pData, pt, p - pt); + + // insert row + tsdbMemSkipListCursorPut(&slc, pSlNode); + + // update status + if (pTSRow->ts < pMemData->minKey) pMemData->minKey = pTSRow->ts; + if (pTSRow->ts > pMemData->maxKey) pMemData->maxKey = pTSRow->ts; + } + tsdbMemSkipListCursorClose(&slc); +#endif + + if (pMemData->minVer == -1) pMemData->minVer = version; + if (pMemData->maxVer == -1 || pMemData->maxVer < version) pMemData->maxVer = version; + + if (pMemTb->minKey < pMemData->minKey) pMemTb->minKey = pMemData->minKey; + if (pMemTb->maxKey < pMemData->maxKey) pMemTb->maxKey = pMemData->maxKey; + if (pMemTb->minVer == -1) pMemTb->minVer = version; + if (pMemTb->maxVer == -1 || pMemTb->maxVer < version) pMemTb->maxVer = version; + return 0; } diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 8460400b59..403c02b440 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -360,7 +360,11 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, // do create table if (metaCreateTable(pVnode->pMeta, version, pCreateReq) < 0) { - cRsp.code = terrno; + if (pCreateReq->flags & TD_CREATE_IF_NOT_EXISTS && terrno == TSDB_CODE_TDB_TABLE_ALREADY_EXIST) { + cRsp.code = TSDB_CODE_SUCCESS; + } else { + cRsp.code = terrno; + } } else { cRsp.code = TSDB_CODE_SUCCESS; tsdbFetchTbUidList(pVnode->pTsdb, &pStore, pCreateReq->ctb.suid, pCreateReq->uid); @@ -529,6 +533,13 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in } } + msgIter.uid = createTbReq.uid; + if (createTbReq.type == TSDB_CHILD_TABLE) { + msgIter.suid = createTbReq.ctb.suid; + } else { + msgIter.suid = 0; + } + tCoderClear(&coder); }