/* * Copyright (c) 2019 TAOS Data, Inc. * * This program is free software: you can use, redistribute, and/or modify * it under the terms of the GNU Affero General Public License, version 3 * or later ("AGPL"), as published by the Free Software Foundation. * * This program is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ #include "tsdb.h" // SMapData ======================================================================= void tMapDataReset(SMapData *pMapData) { pMapData->nItem = 0; pMapData->nData = 0; } void tMapDataClear(SMapData *pMapData) { tsdbFree((uint8_t *)pMapData->aOffset); tsdbFree(pMapData->pData); } int32_t tMapDataPutItem(SMapData *pMapData, void *pItem, int32_t (*tPutItemFn)(uint8_t *, void *)) { int32_t code = 0; int32_t offset = pMapData->nData; int32_t nItem = pMapData->nItem; pMapData->nItem++; pMapData->nData += tPutItemFn(NULL, pItem); // alloc code = tsdbRealloc((uint8_t **)&pMapData->aOffset, sizeof(int32_t) * pMapData->nItem); if (code) goto _err; code = tsdbRealloc(&pMapData->pData, pMapData->nData); if (code) goto _err; // put pMapData->aOffset[nItem] = offset; tPutItemFn(pMapData->pData + offset, pItem); _err: return code; } int32_t tMapDataSearch(SMapData *pMapData, void *pSearchItem, int32_t (*tGetItemFn)(uint8_t *, void *), int32_t (*tItemCmprFn)(const void *, const void *), void *pItem) { int32_t code = 0; int32_t lidx = 0; int32_t ridx = pMapData->nItem - 1; int32_t midx; int32_t c; while (lidx <= ridx) { midx = (lidx + ridx) / 2; tMapDataGetItemByIdx(pMapData, midx, pItem, tGetItemFn); c = tItemCmprFn(pSearchItem, pItem); if (c == 0) { goto _exit; } else if (c < 0) { ridx = midx - 1; } else { lidx = midx + 1; } } code = TSDB_CODE_NOT_FOUND; _exit: return code; } void tMapDataGetItemByIdx(SMapData *pMapData, int32_t idx, void *pItem, int32_t (*tGetItemFn)(uint8_t *, void *)) { ASSERT(idx >= 0 && idx < pMapData->nItem); tGetItemFn(pMapData->pData + pMapData->aOffset[idx], pItem); } int32_t tPutMapData(uint8_t *p, SMapData *pMapData) { int32_t n = 0; n += tPutI32v(p ? p + n : p, pMapData->nItem); if (pMapData->nItem) { for (int32_t iItem = 0; iItem < pMapData->nItem; iItem++) { n += tPutI32v(p ? p + n : p, pMapData->aOffset[iItem]); } n += tPutI32v(p ? p + n : p, pMapData->nData); if (p) { memcpy(p + n, pMapData->pData, pMapData->nData); } n += pMapData->nData; } return n; } int32_t tGetMapData(uint8_t *p, SMapData *pMapData) { int32_t n = 0; int32_t offset; tMapDataReset(pMapData); n += tGetI32v(p + n, &pMapData->nItem); if (pMapData->nItem) { if (tsdbRealloc((uint8_t **)&pMapData->aOffset, sizeof(int32_t) * pMapData->nItem)) return -1; for (int32_t iItem = 0; iItem < pMapData->nItem; iItem++) { n += tGetI32v(p + n, &pMapData->aOffset[iItem]); } n += tGetI32v(p + n, &pMapData->nData); if (tsdbRealloc(&pMapData->pData, pMapData->nData)) return -1; memcpy(pMapData->pData, p + n, pMapData->nData); n += pMapData->nData; } return n; } // Memory ======================================================================= int32_t tsdbRealloc(uint8_t **ppBuf, int64_t size) { int32_t code = 0; int64_t bsize = 0; uint8_t *pBuf; if (*ppBuf) { bsize = *(int64_t *)((*ppBuf) - sizeof(int64_t)); } if (bsize >= size) goto _exit; if (bsize == 0) bsize = 64; while (bsize < size) { bsize *= 2; } pBuf = taosMemoryRealloc(*ppBuf ? (*ppBuf) - sizeof(int64_t) : *ppBuf, bsize + sizeof(int64_t)); if (pBuf == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } *(int64_t *)pBuf = bsize; *ppBuf = pBuf + sizeof(int64_t); _exit: return code; } void tsdbFree(uint8_t *pBuf) { if (pBuf) { taosMemoryFree(pBuf - sizeof(int64_t)); } } // TABLEID ======================================================================= int32_t tTABLEIDCmprFn(const void *p1, const void *p2) { TABLEID *pId1 = (TABLEID *)p1; TABLEID *pId2 = (TABLEID *)p2; if (pId1->suid < pId2->suid) { return -1; } else if (pId1->suid > pId2->suid) { return 1; } if (pId1->uid < pId2->uid) { return -1; } else if (pId1->uid > pId2->uid) { return 1; } return 0; } // TSDBKEY ======================================================================= int32_t tsdbKeyCmprFn(const void *p1, const void *p2) { TSDBKEY *pKey1 = (TSDBKEY *)p1; TSDBKEY *pKey2 = (TSDBKEY *)p2; if (pKey1->ts < pKey2->ts) { return -1; } else if (pKey1->ts > pKey2->ts) { return 1; } if (pKey1->version < pKey2->version) { return -1; } else if (pKey1->version > pKey2->version) { return 1; } return 0; } // TSDBKEY ====================================================== static FORCE_INLINE int32_t tPutTSDBKEY(uint8_t *p, TSDBKEY *pKey) { int32_t n = 0; n += tPutI64v(p ? p + n : p, pKey->version); n += tPutI64(p ? p + n : p, pKey->ts); return n; } static FORCE_INLINE int32_t tGetTSDBKEY(uint8_t *p, TSDBKEY *pKey) { int32_t n = 0; n += tGetI64v(p + n, &pKey->version); n += tGetI64(p + n, &pKey->ts); return n; } // SBlockIdx ====================================================== void tBlockIdxReset(SBlockIdx *pBlockIdx) { pBlockIdx->minKey = TSKEY_MAX; pBlockIdx->maxKey = TSKEY_MIN; pBlockIdx->minVersion = VERSION_MAX; pBlockIdx->maxVersion = VERSION_MIN; pBlockIdx->offset = -1; pBlockIdx->size = -1; } int32_t tPutBlockIdx(uint8_t *p, void *ph) { int32_t n = 0; SBlockIdx *pBlockIdx = (SBlockIdx *)ph; n += tPutI64(p ? p + n : p, pBlockIdx->suid); n += tPutI64(p ? p + n : p, pBlockIdx->uid); n += tPutI64(p ? p + n : p, pBlockIdx->minKey); n += tPutI64(p ? p + n : p, pBlockIdx->maxKey); n += tPutI64v(p ? p + n : p, pBlockIdx->minVersion); n += tPutI64v(p ? p + n : p, pBlockIdx->maxVersion); n += tPutI64v(p ? p + n : p, pBlockIdx->offset); n += tPutI64v(p ? p + n : p, pBlockIdx->size); return n; } int32_t tGetBlockIdx(uint8_t *p, void *ph) { int32_t n = 0; SBlockIdx *pBlockIdx = (SBlockIdx *)ph; n += tGetI64(p + n, &pBlockIdx->suid); n += tGetI64(p + n, &pBlockIdx->uid); n += tGetI64(p + n, &pBlockIdx->minKey); n += tGetI64(p + n, &pBlockIdx->maxKey); n += tGetI64v(p + n, &pBlockIdx->minVersion); n += tGetI64v(p + n, &pBlockIdx->maxVersion); n += tGetI64v(p + n, &pBlockIdx->offset); n += tGetI64v(p + n, &pBlockIdx->size); return n; } int32_t tCmprBlockIdx(void const *lhs, void const *rhs) { SBlockIdx *lBlockIdx = *(SBlockIdx **)lhs; SBlockIdx *rBlockIdx = *(SBlockIdx **)rhs; if (lBlockIdx->suid < lBlockIdx->suid) { return -1; } else if (lBlockIdx->suid > lBlockIdx->suid) { return 1; } if (lBlockIdx->uid < lBlockIdx->uid) { return -1; } else if (lBlockIdx->uid > lBlockIdx->uid) { return 1; } return 0; } // SBlock ====================================================== void tBlockReset(SBlock *pBlock) { *pBlock = (SBlock){.minKey = TSDBKEY_MAX, .maxKey = TSDBKEY_MIN, .minVersion = VERSION_MAX, .maxVersion = VERSION_MIN}; } int32_t tPutBlock(uint8_t *p, void *ph) { int32_t n = 0; SBlock *pBlock = (SBlock *)ph; n += tPutTSDBKEY(p ? p + n : p, &pBlock->minKey); n += tPutTSDBKEY(p ? p + n : p, &pBlock->maxKey); n += tPutI64v(p ? p + n : p, pBlock->minVersion); n += tPutI64v(p ? p + n : p, pBlock->maxVersion); n += tPutI32v(p ? p + n : p, pBlock->nRow); n += tPutI8(p ? p + n : p, pBlock->last); n += tPutI8(p ? p + n : p, pBlock->hasDup); n += tPutI8(p ? p + n : p, pBlock->nSubBlock); for (int8_t iSubBlock = 0; iSubBlock < pBlock->nSubBlock; iSubBlock++) { n += tPutI32v(p ? p + n : p, pBlock->aSubBlock[iSubBlock].nRow); n += tPutI8(p ? p + n : p, pBlock->aSubBlock[iSubBlock].cmprAlg); n += tPutI64v(p ? p + n : p, pBlock->aSubBlock[iSubBlock].offset); n += tPutI32v(p ? p + n : p, pBlock->aSubBlock[iSubBlock].szBlockCol); n += tPutI32v(p ? p + n : p, pBlock->aSubBlock[iSubBlock].szVersion); n += tPutI32v(p ? p + n : p, pBlock->aSubBlock[iSubBlock].szTSKEY); n += tPutI32v(p ? p + n : p, pBlock->aSubBlock[iSubBlock].szBlock); n += tPutI64v(p ? p + n : p, pBlock->aSubBlock[iSubBlock].sOffset); n += tPutI32v(p ? p + n : p, pBlock->aSubBlock[iSubBlock].nSma); } return n; } int32_t tGetBlock(uint8_t *p, void *ph) { int32_t n = 0; SBlock *pBlock = (SBlock *)ph; n += tGetTSDBKEY(p + n, &pBlock->minKey); n += tGetTSDBKEY(p + n, &pBlock->maxKey); n += tGetI64v(p + n, &pBlock->minVersion); n += tGetI64v(p + n, &pBlock->maxVersion); n += tGetI32v(p + n, &pBlock->nRow); n += tGetI8(p + n, &pBlock->last); n += tGetI8(p + n, &pBlock->hasDup); n += tGetI8(p + n, &pBlock->nSubBlock); for (int8_t iSubBlock = 0; iSubBlock < pBlock->nSubBlock; iSubBlock++) { n += tGetI32v(p + n, &pBlock->aSubBlock[iSubBlock].nRow); n += tGetI8(p + n, &pBlock->aSubBlock[iSubBlock].cmprAlg); n += tGetI64v(p + n, &pBlock->aSubBlock[iSubBlock].offset); n += tGetI32v(p + n, &pBlock->aSubBlock[iSubBlock].szBlockCol); n += tGetI32v(p + n, &pBlock->aSubBlock[iSubBlock].szVersion); n += tGetI32v(p + n, &pBlock->aSubBlock[iSubBlock].szTSKEY); n += tGetI32v(p + n, &pBlock->aSubBlock[iSubBlock].szBlock); n += tGetI64v(p + n, &pBlock->aSubBlock[iSubBlock].sOffset); n += tGetI32v(p + n, &pBlock->aSubBlock[iSubBlock].nSma); } return n; } int32_t tBlockCmprFn(const void *p1, const void *p2) { SBlock *pBlock1 = (SBlock *)p1; SBlock *pBlock2 = (SBlock *)p2; if (tsdbKeyCmprFn(&pBlock1->maxKey, &pBlock2->minKey) < 0) { return -1; } else if (tsdbKeyCmprFn(&pBlock1->minKey, &pBlock2->maxKey) > 0) { return 1; } return 0; } bool tBlockHasSma(SBlock *pBlock) { if (pBlock->nSubBlock > 1) return false; if (pBlock->last) return false; if (pBlock->hasDup) return false; return pBlock->aSubBlock[0].nSma > 0; } // SBlockCol ====================================================== int32_t tPutBlockCol(uint8_t *p, void *ph) { int32_t n = 0; SBlockCol *pBlockCol = (SBlockCol *)ph; ASSERT(pBlockCol->flag && (pBlockCol->flag != HAS_NONE)); n += tPutI16v(p ? p + n : p, pBlockCol->cid); n += tPutI8(p ? p + n : p, pBlockCol->type); n += tPutI8(p ? p + n : p, pBlockCol->flag); if (pBlockCol->flag != HAS_NULL) { n += tPutI32v(p ? p + n : p, pBlockCol->offset); n += tPutI32v(p ? p + n : p, pBlockCol->szBitmap); n += tPutI32v(p ? p + n : p, pBlockCol->szOffset); n += tPutI32v(p ? p + n : p, pBlockCol->szValue); n += tPutI32v(p ? p + n : p, pBlockCol->szOrigin); } return n; } int32_t tGetBlockCol(uint8_t *p, void *ph) { int32_t n = 0; SBlockCol *pBlockCol = (SBlockCol *)ph; n += tGetI16v(p + n, &pBlockCol->cid); n += tGetI8(p + n, &pBlockCol->type); n += tGetI8(p + n, &pBlockCol->flag); ASSERT(pBlockCol->flag && (pBlockCol->flag != HAS_NONE)); if (pBlockCol->flag != HAS_NULL) { n += tGetI32v(p + n, &pBlockCol->offset); n += tGetI32v(p + n, &pBlockCol->szBitmap); n += tGetI32v(p + n, &pBlockCol->szOffset); n += tGetI32v(p + n, &pBlockCol->szValue); n += tGetI32v(p + n, &pBlockCol->szOrigin); } return n; } int32_t tBlockColCmprFn(const void *p1, const void *p2) { if (((SBlockCol *)p1)->cid < ((SBlockCol *)p2)->cid) { return -1; } else if (((SBlockCol *)p1)->cid > ((SBlockCol *)p2)->cid) { return 1; } return 0; } // SDelIdx ====================================================== int32_t tCmprDelIdx(void const *lhs, void const *rhs) { SDelIdx *lDelIdx = *(SDelIdx **)lhs; SDelIdx *rDelIdx = *(SDelIdx **)rhs; if (lDelIdx->suid < lDelIdx->suid) { return -1; } else if (lDelIdx->suid > lDelIdx->suid) { return 1; } if (lDelIdx->uid < lDelIdx->uid) { return -1; } else if (lDelIdx->uid > lDelIdx->uid) { return 1; } return 0; } int32_t tPutDelIdx(uint8_t *p, void *ph) { SDelIdx *pDelIdx = (SDelIdx *)ph; int32_t n = 0; n += tPutI64(p ? p + n : p, pDelIdx->suid); n += tPutI64(p ? p + n : p, pDelIdx->uid); n += tPutI64v(p ? p + n : p, pDelIdx->offset); n += tPutI64v(p ? p + n : p, pDelIdx->size); return n; } int32_t tGetDelIdx(uint8_t *p, void *ph) { SDelIdx *pDelIdx = (SDelIdx *)ph; int32_t n = 0; n += tGetI64(p + n, &pDelIdx->suid); n += tGetI64(p + n, &pDelIdx->uid); n += tGetI64v(p + n, &pDelIdx->offset); n += tGetI64v(p + n, &pDelIdx->size); return n; } // SDelData ====================================================== int32_t tPutDelData(uint8_t *p, void *ph) { SDelData *pDelData = (SDelData *)ph; int32_t n = 0; n += tPutI64v(p ? p + n : p, pDelData->version); n += tPutI64(p ? p + n : p, pDelData->sKey); n += tPutI64(p ? p + n : p, pDelData->eKey); return n; } int32_t tGetDelData(uint8_t *p, void *ph) { SDelData *pDelData = (SDelData *)ph; int32_t n = 0; n += tGetI64v(p + n, &pDelData->version); n += tGetI64(p + n, &pDelData->sKey); n += tGetI64(p + n, &pDelData->eKey); return n; } int32_t tsdbKeyFid(TSKEY key, int32_t minutes, int8_t precision) { if (key < 0) { return (int)((key + 1) / tsTickPerMin[precision] / minutes - 1); } else { return (int)((key / tsTickPerMin[precision] / minutes)); } } void tsdbFidKeyRange(int32_t fid, int32_t minutes, int8_t precision, TSKEY *minKey, TSKEY *maxKey) { *minKey = fid * minutes * tsTickPerMin[precision]; *maxKey = *minKey + minutes * tsTickPerMin[precision] - 1; } // int tsdFidLevel(int fid, TSKEY now, minute) { // if (fid >= pRtn->maxFid) { // return 0; // } else if (fid >= pRtn->midFid) { // return 1; // } else if (fid >= pRtn->minFid) { // return 2; // } else { // return -1; // } // } // TSDBROW ====================================================== void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) { STColumn *pTColumn = &pTSchema->columns[iCol]; SValue value; ASSERT(iCol > 0); if (pRow->type == 0) { tTSRowGetVal(pRow->pTSRow, pTSchema, iCol, pColVal); } else if (pRow->type == 1) { SColData *pColData; tBlockDataGetColData(pRow->pBlockData, pTColumn->colId, &pColData); if (pColData) { tColDataGetValue(pColData, pRow->iRow, pColVal); } else { *pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type); } } else { ASSERT(0); } } int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow) { int32_t n = 0; n += tPutI64(p, pRow->version); if (p) memcpy(p + n, pRow->pTSRow, pRow->pTSRow->len); n += pRow->pTSRow->len; return n; } int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow) { int32_t n = 0; n += tGetI64(p, &pRow->version); pRow->pTSRow = (STSRow *)(p + n); n += pRow->pTSRow->len; return n; } int32_t tsdbRowCmprFn(const void *p1, const void *p2) { return tsdbKeyCmprFn(&TSDBROW_KEY((TSDBROW *)p1), &TSDBROW_KEY((TSDBROW *)p2)); } // SRowIter ====================================================== void tRowIterInit(SRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) { pIter->pRow = pRow; if (pRow->type == 0) { ASSERT(pTSchema); pIter->pTSchema = pTSchema; pIter->i = 1; } else if (pRow->type == 1) { pIter->pTSchema = NULL; pIter->i = 0; } else { ASSERT(0); } } SColVal *tRowIterNext(SRowIter *pIter) { if (pIter->pRow->type == 0) { if (pIter->i < pIter->pTSchema->numOfCols) { tsdbRowGetColVal(pIter->pRow, pIter->pTSchema, pIter->i, &pIter->colVal); pIter->i++; return &pIter->colVal; } } else { if (pIter->i < taosArrayGetSize(pIter->pRow->pBlockData->aColDataP)) { SColData *pColData = (SColData *)taosArrayGetP(pIter->pRow->pBlockData->aColDataP, pIter->i); tColDataGetValue(pColData, pIter->pRow->iRow, &pIter->colVal); pIter->i++; return &pIter->colVal; } } return NULL; } // SRowMerger ====================================================== int32_t tRowMergerInit(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema) { int32_t code = 0; TSDBKEY key = TSDBROW_KEY(pRow); SColVal *pColVal = &(SColVal){0}; STColumn *pTColumn; pMerger->pTSchema = pTSchema; pMerger->version = key.version; pMerger->pArray = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal)); if (pMerger->pArray == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } // ts pTColumn = &pTSchema->columns[0]; ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP); *pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, (SValue){.ts = key.ts}); if (taosArrayPush(pMerger->pArray, pColVal) == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } // other for (int16_t iCol = 1; iCol < pTSchema->numOfCols; iCol++) { tsdbRowGetColVal(pRow, pTSchema, iCol, pColVal); if (taosArrayPush(pMerger->pArray, pColVal) == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } } _exit: return code; } void tRowMergerClear(SRowMerger *pMerger) { taosArrayDestroy(pMerger->pArray); } int32_t tRowMerge(SRowMerger *pMerger, TSDBROW *pRow) { int32_t code = 0; TSDBKEY key = TSDBROW_KEY(pRow); SColVal *pColVal = &(SColVal){0}; ASSERT(((SColVal *)pMerger->pArray->pData)->value.ts == key.ts); for (int32_t iCol = 1; iCol < pMerger->pTSchema->numOfCols; iCol++) { tsdbRowGetColVal(pRow, pMerger->pTSchema, iCol, pColVal); if (key.version > pMerger->version) { if (!pColVal->isNone) { taosArraySet(pMerger->pArray, iCol, pColVal); } } else if (key.version < pMerger->version) { SColVal *tColVal = (SColVal *)taosArrayGet(pMerger->pArray, iCol); if (tColVal->isNone && !pColVal->isNone) { taosArraySet(pMerger->pArray, iCol, pColVal); } } else { ASSERT(0); } } pMerger->version = key.version; _exit: return code; } int32_t tRowMergerGetRow(SRowMerger *pMerger, STSRow **ppRow) { int32_t code = 0; code = tdSTSRowNew(pMerger->pArray, pMerger->pTSchema, ppRow); return code; } // delete skyline ====================================================== static int32_t tsdbMergeSkyline(SArray *aSkyline1, SArray *aSkyline2, SArray *aSkyline) { int32_t code = 0; int32_t i1 = 0; int32_t n1 = taosArrayGetSize(aSkyline1); int32_t i2 = 0; int32_t n2 = taosArrayGetSize(aSkyline2); TSDBKEY *pSkyline1; TSDBKEY *pSkyline2; TSDBKEY item; int64_t version1 = 0; int64_t version2 = 0; ASSERT(n1 > 0 && n2 > 0); taosArrayClear(aSkyline); while (i1 < n1 && i2 < n2) { pSkyline1 = (TSDBKEY *)taosArrayGet(aSkyline1, i1); pSkyline2 = (TSDBKEY *)taosArrayGet(aSkyline2, i2); if (pSkyline1->ts < pSkyline2->ts) { version1 = pSkyline1->version; i1++; } else if (pSkyline1->ts > pSkyline2->ts) { version2 = pSkyline2->version; i2++; } else { version1 = pSkyline1->version; version2 = pSkyline2->version; i1++; i2++; } item.ts = TMIN(pSkyline1->ts, pSkyline2->ts); item.version = TMAX(version1, version2); if (taosArrayPush(aSkyline, &item) == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } } while (i1 < n1) { pSkyline1 = (TSDBKEY *)taosArrayGet(aSkyline1, i1); item.ts = pSkyline1->ts; item.version = pSkyline1->version; if (taosArrayPush(aSkyline, &item) == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } i1++; } while (i2 < n2) { pSkyline2 = (TSDBKEY *)taosArrayGet(aSkyline2, i2); item.ts = pSkyline2->ts; item.version = pSkyline2->version; if (taosArrayPush(aSkyline, &item) == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } i2++; } _exit: return code; } int32_t tsdbBuildDeleteSkyline(SArray *aDelData, int32_t sidx, int32_t eidx, SArray *aSkyline) { int32_t code = 0; SDelData *pDelData; int32_t midx; taosArrayClear(aSkyline); if (sidx == eidx) { pDelData = (SDelData *)taosArrayGet(aDelData, sidx); taosArrayPush(aSkyline, &(TSDBKEY){.ts = pDelData->sKey, .version = pDelData->version}); taosArrayPush(aSkyline, &(TSDBKEY){.ts = pDelData->eKey, .version = 0}); } else { SArray *aSkyline1 = NULL; SArray *aSkyline2 = NULL; aSkyline1 = taosArrayInit(0, sizeof(TSDBKEY)); aSkyline2 = taosArrayInit(0, sizeof(TSDBKEY)); if (aSkyline1 == NULL || aSkyline2 == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _clear; } midx = (sidx + eidx) / 2; code = tsdbBuildDeleteSkyline(aDelData, sidx, midx, aSkyline1); if (code) goto _clear; code = tsdbBuildDeleteSkyline(aDelData, midx + 1, eidx, aSkyline2); if (code) goto _clear; code = tsdbMergeSkyline(aSkyline1, aSkyline2, aSkyline); _clear: taosArrayDestroy(aSkyline1); taosArrayDestroy(aSkyline2); } return code; } // SColData ======================================== void tColDataReset(SColData *pColData, int16_t cid, int8_t type, int8_t smaOn) { pColData->cid = cid; pColData->type = type; pColData->smaOn = smaOn; pColData->nVal = 0; pColData->flag = 0; pColData->nData = 0; } void tColDataClear(void *ph) { SColData *pColData = (SColData *)ph; tsdbFree(pColData->pBitMap); tsdbFree((uint8_t *)pColData->aOffset); tsdbFree(pColData->pData); } int32_t tColDataAppendValue(SColData *pColData, SColVal *pColVal) { int32_t code = 0; int64_t size; SValue value = {0}; SValue *pValue = &value; ASSERT(pColVal->cid == pColData->cid); ASSERT(pColVal->type == pColData->type); // realloc bitmap size = BIT2_SIZE(pColData->nVal + 1); code = tsdbRealloc(&pColData->pBitMap, size); if (code) goto _exit; // put value if (pColVal->isNone) { pColData->flag |= HAS_NONE; SET_BIT2(pColData->pBitMap, pColData->nVal, 0); } else if (pColVal->isNull) { pColData->flag |= HAS_NULL; SET_BIT2(pColData->pBitMap, pColData->nVal, 1); } else { pColData->flag |= HAS_VALUE; SET_BIT2(pColData->pBitMap, pColData->nVal, 2); pValue = &pColVal->value; } if (IS_VAR_DATA_TYPE(pColData->type)) { // offset code = tsdbRealloc((uint8_t **)&pColData->aOffset, sizeof(int32_t) * (pColData->nVal + 1)); if (code) goto _exit; pColData->aOffset[pColData->nVal] = pColData->nData; // value if ((!pColVal->isNone) && (!pColVal->isNull)) { code = tsdbRealloc(&pColData->pData, pColData->nData + pColVal->value.nData); if (code) goto _exit; memcpy(pColData->pData + pColData->nData, pColVal->value.pData, pColVal->value.nData); pColData->nData += pColVal->value.nData; } } else { code = tsdbRealloc(&pColData->pData, pColData->nData + tPutValue(NULL, pValue, pColVal->type)); if (code) goto _exit; pColData->nData += tPutValue(pColData->pData + pColData->nData, pValue, pColVal->type); } pColData->nVal++; _exit: return code; } int32_t tColDataCopy(SColData *pColDataSrc, SColData *pColDataDest) { int32_t code = 0; int32_t size; pColDataDest->cid = pColDataSrc->cid; pColDataDest->type = pColDataSrc->type; pColDataDest->smaOn = pColDataSrc->smaOn; pColDataDest->nVal = pColDataSrc->nVal; pColDataDest->flag = pColDataSrc->flag; size = BIT2_SIZE(pColDataSrc->nVal); code = tsdbRealloc(&pColDataDest->pBitMap, size); if (code) goto _exit; memcpy(pColDataDest->pBitMap, pColDataSrc->pBitMap, size); if (IS_VAR_DATA_TYPE(pColDataDest->type)) { size = sizeof(int32_t) * pColDataSrc->nVal; code = tsdbRealloc((uint8_t **)&pColDataDest->aOffset, size); if (code) goto _exit; memcpy(pColDataDest->aOffset, pColDataSrc->aOffset, size); } code = tsdbRealloc(&pColDataDest->pData, pColDataSrc->nData); if (code) goto _exit; pColDataDest->nData = pColDataSrc->nData; memcpy(pColDataDest->pData, pColDataSrc->pData, pColDataDest->nData); _exit: return code; } int32_t tColDataGetValue(SColData *pColData, int32_t iVal, SColVal *pColVal) { int32_t code = 0; ASSERT(iVal < pColData->nVal); ASSERT(pColData->flag); if (pColData->flag == HAS_NONE) { *pColVal = COL_VAL_NONE(pColData->cid, pColData->type); goto _exit; } else if (pColData->flag == HAS_NULL) { *pColVal = COL_VAL_NULL(pColData->cid, pColData->type); goto _exit; } else if (pColData->flag != HAS_VALUE) { uint8_t v = GET_BIT2(pColData->pBitMap, iVal); if (v == 0) { *pColVal = COL_VAL_NONE(pColData->cid, pColData->type); goto _exit; } else if (v == 1) { *pColVal = COL_VAL_NULL(pColData->cid, pColData->type); goto _exit; } } // get value SValue value; if (IS_VAR_DATA_TYPE(pColData->type)) { if (iVal + 1 < pColData->nVal) { value.nData = pColData->aOffset[iVal + 1] - pColData->aOffset[iVal]; } else { value.nData = pColData->nData - pColData->aOffset[iVal]; } value.pData = pColData->pData + pColData->aOffset[iVal]; } else { tGetValue(pColData->pData + tDataTypes[pColData->type].bytes * iVal, &value, pColData->type); } *pColVal = COL_VAL_VALUE(pColData->cid, pColData->type, value); _exit: return code; } static FORCE_INLINE int32_t tColDataPCmprFn(const void *p1, const void *p2) { SColData *pColData1 = *(SColData **)p1; SColData *pColData2 = *(SColData **)p2; if (pColData1->cid < pColData2->cid) { return -1; } else if (pColData1->cid > pColData2->cid) { return 1; } return 0; } // SBlockData ====================================================== int32_t tBlockDataInit(SBlockData *pBlockData) { int32_t code = 0; pBlockData->nRow = 0; pBlockData->aVersion = NULL; pBlockData->aTSKEY = NULL; pBlockData->aColDataP = taosArrayInit(0, sizeof(SColData *)); if (pBlockData->aColDataP == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } pBlockData->aColData = taosArrayInit(0, sizeof(SColData)); if (pBlockData->aColData == NULL) { taosArrayDestroy(pBlockData->aColDataP); code = TSDB_CODE_OUT_OF_MEMORY; goto _exit; } _exit: return code; } void tBlockDataReset(SBlockData *pBlockData) { pBlockData->nRow = 0; taosArrayClear(pBlockData->aColDataP); } void tBlockDataClear(SBlockData *pBlockData) { tsdbFree((uint8_t *)pBlockData->aVersion); tsdbFree((uint8_t *)pBlockData->aTSKEY); taosArrayDestroy(pBlockData->aColDataP); taosArrayDestroyEx(pBlockData->aColData, tColDataClear); } int32_t tBlockDataAddColData(SBlockData *pBlockData, int32_t iColData, SColData **ppColData) { int32_t code = 0; SColData *pColData = NULL; int32_t idx = taosArrayGetSize(pBlockData->aColDataP); if (idx >= taosArrayGetSize(pBlockData->aColData)) { if (taosArrayPush(pBlockData->aColData, &((SColData){0})) == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _err; } } pColData = (SColData *)taosArrayGet(pBlockData->aColData, idx); if (taosArrayInsert(pBlockData->aColDataP, iColData, &pColData) == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _err; } *ppColData = pColData; return code; _err: *ppColData = NULL; return code; } int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema) { int32_t code = 0; // TSDBKEY code = tsdbRealloc((uint8_t **)&pBlockData->aVersion, sizeof(int64_t) * (pBlockData->nRow + 1)); if (code) goto _err; code = tsdbRealloc((uint8_t **)&pBlockData->aTSKEY, sizeof(TSKEY) * (pBlockData->nRow + 1)); if (code) goto _err; pBlockData->aVersion[pBlockData->nRow] = TSDBROW_VERSION(pRow); pBlockData->aTSKEY[pBlockData->nRow] = TSDBROW_TS(pRow); // OTHER int32_t iColData = 0; SRowIter *pIter = &((SRowIter){0}); SColData *pColData; SColVal *pColVal; tRowIterInit(pIter, pRow, pTSchema); pColVal = tRowIterNext(pIter); pColData = (iColData < taosArrayGetSize(pBlockData->aColDataP)) ? (SColData *)taosArrayGetP(pBlockData->aColDataP, iColData) : NULL; while (pColVal && pColData) { if (pColVal->cid == pColData->cid) { code = tColDataAppendValue(pColData, pColVal); if (code) goto _err; pColVal = tRowIterNext(pIter); } else if (pColVal->cid > pColData->cid) { code = tColDataAppendValue(pColData, &(COL_VAL_NONE(pColData->cid, pColData->type))); if (code) goto _err; } else { code = tBlockDataAddColData(pBlockData, iColData, &pColData); if (code) goto _err; // append a NONE tColDataReset(pColData, pColVal->cid, pColVal->type, 0); for (int32_t iRow = 0; iRow < pBlockData->nRow; iRow++) { code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColVal->cid, pColVal->type)); if (code) goto _err; } code = tColDataAppendValue(pColData, pColVal); if (code) goto _err; pColVal = tRowIterNext(pIter); } pColData = ((++iColData) < taosArrayGetSize(pBlockData->aColDataP)) ? (SColData *)taosArrayGetP(pBlockData->aColDataP, iColData) : NULL; } while (pColData) { code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColData->cid, pColData->type)); if (code) goto _err; pColData = ((++iColData) < taosArrayGetSize(pBlockData->aColDataP)) ? (SColData *)taosArrayGetP(pBlockData->aColDataP, iColData) : NULL; } while (pColVal) { code = tBlockDataAddColData(pBlockData, iColData, &pColData); if (code) goto _err; tColDataReset(pColData, pColVal->cid, pColVal->type, 0); for (int32_t iRow = 0; iRow < pBlockData->nRow; iRow++) { code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColVal->cid, pColVal->type)); if (code) goto _err; } code = tColDataAppendValue(pColData, pColVal); if (code) goto _err; iColData++; pColVal = tRowIterNext(pIter); } pBlockData->nRow++; return code; _err: return code; } int32_t tBlockDataMerge(SBlockData *pBlockData1, SBlockData *pBlockData2, SBlockData *pBlockData) { int32_t code = 0; tBlockDataReset(pBlockData); // loop to merge int32_t iRow1 = 0; int32_t nRow1 = pBlockData1->nRow; int32_t iRow2 = 0; int32_t nRow2 = pBlockData2->nRow; TSDBROW row1; TSDBROW row2; int32_t c; while (iRow1 < nRow1 && iRow2 < nRow2) { row1 = tsdbRowFromBlockData(pBlockData1, iRow1); row2 = tsdbRowFromBlockData(pBlockData2, iRow2); c = tsdbKeyCmprFn(&TSDBROW_KEY(&row1), &TSDBROW_KEY(&row2)); if (c < 0) { code = tBlockDataAppendRow(pBlockData, &row1, NULL); if (code) goto _exit; iRow1++; } else if (c > 0) { code = tBlockDataAppendRow(pBlockData, &row2, NULL); if (code) goto _exit; iRow2++; } else { ASSERT(0); } } while (iRow1 < nRow1) { row1 = tsdbRowFromBlockData(pBlockData1, iRow1); code = tBlockDataAppendRow(pBlockData, &row1, NULL); if (code) goto _exit; iRow1++; } while (iRow2 < nRow2) { row2 = tsdbRowFromBlockData(pBlockData2, iRow2); code = tBlockDataAppendRow(pBlockData, &row2, NULL); if (code) goto _exit; iRow2++; } _exit: return code; } int32_t tBlockDataCopy(SBlockData *pBlockDataSrc, SBlockData *pBlockDataDest) { int32_t code = 0; SColData *pColDataSrc; SColData *pColDataDest; ASSERT(pBlockDataSrc->nRow > 0); tBlockDataReset(pBlockDataDest); pBlockDataDest->nRow = pBlockDataSrc->nRow; // TSDBKEY code = tsdbRealloc((uint8_t **)&pBlockDataDest->aVersion, sizeof(int64_t) * pBlockDataSrc->nRow); if (code) goto _exit; code = tsdbRealloc((uint8_t **)&pBlockDataDest->aTSKEY, sizeof(TSKEY) * pBlockDataSrc->nRow); if (code) goto _exit; memcpy(pBlockDataDest->aVersion, pBlockDataSrc->aVersion, sizeof(int64_t) * pBlockDataSrc->nRow); memcpy(pBlockDataDest->aTSKEY, pBlockDataSrc->aTSKEY, sizeof(TSKEY) * pBlockDataSrc->nRow); // other for (size_t iColData = 0; iColData < taosArrayGetSize(pBlockDataSrc->aColDataP); iColData++) { pColDataSrc = (SColData *)taosArrayGetP(pBlockDataSrc->aColDataP, iColData); code = tBlockDataAddColData(pBlockDataDest, iColData, &pColDataDest); if (code) goto _exit; code = tColDataCopy(pColDataSrc, pColDataDest); if (code) goto _exit; } _exit: return code; } void tBlockDataGetColData(SBlockData *pBlockData, int16_t cid, SColData **ppColData) { ASSERT(cid != PRIMARYKEY_TIMESTAMP_COL_ID); SColData *pColData = &(SColData){.cid = cid}; void *p = taosArraySearch(pBlockData->aColDataP, &pColData, tColDataPCmprFn, TD_EQ); if (p == NULL) { *ppColData = NULL; } else { *ppColData = *(SColData **)p; } } // ALGORITHM ============================== void tsdbCalcColDataSMA(SColData *pColData, SColumnDataAgg *pColAgg) { SColVal colVal; SColVal *pColVal = &colVal; *pColAgg = (SColumnDataAgg){.colId = pColData->cid}; for (int32_t iVal = 0; iVal < pColData->nVal; iVal++) { tColDataGetValue(pColData, iVal, pColVal); if (pColVal->isNone || pColVal->isNull) { pColAgg->numOfNull++; } else { switch (pColData->type) { case TSDB_DATA_TYPE_NULL: break; case TSDB_DATA_TYPE_BOOL: break; case TSDB_DATA_TYPE_TINYINT: break; case TSDB_DATA_TYPE_SMALLINT: break; case TSDB_DATA_TYPE_INT: break; case TSDB_DATA_TYPE_BIGINT: break; case TSDB_DATA_TYPE_FLOAT: break; case TSDB_DATA_TYPE_DOUBLE: break; case TSDB_DATA_TYPE_VARCHAR: break; case TSDB_DATA_TYPE_TIMESTAMP: break; case TSDB_DATA_TYPE_NCHAR: break; case TSDB_DATA_TYPE_UTINYINT: break; case TSDB_DATA_TYPE_USMALLINT: break; case TSDB_DATA_TYPE_UINT: break; case TSDB_DATA_TYPE_UBIGINT: break; case TSDB_DATA_TYPE_JSON: break; case TSDB_DATA_TYPE_VARBINARY: break; case TSDB_DATA_TYPE_DECIMAL: break; case TSDB_DATA_TYPE_BLOB: break; case TSDB_DATA_TYPE_MEDIUMBLOB: break; default: ASSERT(0); } } } }