From 7deee1c4c78d8ce9ae51c5475f466a5f6e2c7983 Mon Sep 17 00:00:00 2001 From: Xiaoyu Wang Date: Thu, 2 Mar 2023 18:22:42 +0800 Subject: [PATCH] fix: last cache read problem after schema change --- source/dnode/vnode/src/inc/tsdb.h | 1 + source/dnode/vnode/src/tsdb/tsdbCache.c | 93 ++++++++++++++++----- source/dnode/vnode/src/tsdb/tsdbCacheRead.c | 4 +- 3 files changed, 76 insertions(+), 22 deletions(-) diff --git a/source/dnode/vnode/src/inc/tsdb.h b/source/dnode/vnode/src/inc/tsdb.h index 44b7e95f9e..15087e90f0 100644 --- a/source/dnode/vnode/src/inc/tsdb.h +++ b/source/dnode/vnode/src/inc/tsdb.h @@ -766,6 +766,7 @@ typedef struct SCacheRowsReader { TdThreadMutex readerMutex; SVnode *pVnode; STSchema *pSchema; + STSchema *pCurrSchema; uint64_t uid; uint64_t suid; char **transferBuf; // todo remove it soon diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c index 79e0ccc3e3..f69445fa0c 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCache.c +++ b/source/dnode/vnode/src/tsdb/tsdbCache.c @@ -1390,17 +1390,56 @@ _err: return code; } -static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SCacheRowsReader *pr) { - int32_t code = 0; +static int32_t cloneTSchema(STSchema *pSrc, STSchema **ppDst) { + int32_t len = sizeof(STSchema) + sizeof(STColumn) * pSrc->numOfCols; + *ppDst = taosMemoryMalloc(len); + if (NULL == *ppDst) { + return TSDB_CODE_OUT_OF_MEMORY; + } + memcpy(*ppDst, pSrc, len); + return TSDB_CODE_SUCCESS; +} + +static int32_t updateTSchema(int32_t sversion, SCacheRowsReader *pReader, uint64_t uid) { + if (NULL == pReader->pCurrSchema && sversion == pReader->pSchema->version) { + return cloneTSchema(pReader->pSchema, &pReader->pCurrSchema); + } + + if (NULL != pReader->pCurrSchema && sversion == pReader->pCurrSchema->version) { + return TSDB_CODE_SUCCESS; + } + + taosMemoryFreeClear(pReader->pCurrSchema); + return metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->suid, uid, sversion, &pReader->pCurrSchema); +} + +static int32_t initLastColArray(STSchema *pTSchema, SArray **ppColArray) { + SArray *pColArray = taosArrayInit(pTSchema->numOfCols, sizeof(SLastCol)); + if (NULL == pColArray) { + return TSDB_CODE_OUT_OF_MEMORY; + } + for (int32_t i = 0; i < pTSchema->numOfCols; ++i) { + SLastCol col = {.ts = 0, .colVal = COL_VAL_NULL(pTSchema->columns[i].colId, pTSchema->columns[i].type)}; + taosArrayPush(pColArray, &col); + } + *ppColArray = pColArray; + return TSDB_CODE_SUCCESS; +} + +static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SCacheRowsReader *pr) { STSchema *pTSchema = pr->pSchema; // metaGetTbTSchema(pTsdb->pVnode->pMeta, uid, -1, 1); - int16_t nCol = pTSchema->numOfCols; - int16_t iCol = 0; + int16_t nLastCol = pTSchema->numOfCols; int16_t noneCol = 0; bool setNoneCol = false; - SArray *pColArray = taosArrayInit(nCol, sizeof(SLastCol)); + SArray *pColArray = NULL; SColVal *pColVal = &(SColVal){0}; + int32_t code = initLastColArray(pTSchema, &pColArray); + if (TSDB_CODE_SUCCESS != code) { + return code; + } + TSKEY lastRowTs = TSKEY_MAX; CacheNextRowIter iter = {0}; @@ -1415,6 +1454,13 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SCach break; } + code = updateTSchema(TSDBROW_SVERSION(pRow), pr, uid); + if (TSDB_CODE_SUCCESS != code) { + goto _err; + } + pTSchema = pr->pCurrSchema; + int16_t nCol = pTSchema->numOfCols; + TSKEY rowTs = TSDBROW_TS(pRow); if (lastRowTs == TSKEY_MAX) { @@ -1422,28 +1468,27 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SCach STColumn *pTColumn = &pTSchema->columns[0]; *pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, (SValue){.val = lastRowTs}); - if (taosArrayPush(pColArray, &(SLastCol){.ts = lastRowTs, .colVal = *pColVal}) == NULL) { - code = TSDB_CODE_OUT_OF_MEMORY; - goto _err; - } + taosArraySet(pColArray, 0, &(SLastCol){.ts = lastRowTs, .colVal = *pColVal}); - for (iCol = 1; iCol < nCol; ++iCol) { + for (int16_t iCol = 1; iCol < nCol; ++iCol) { + if (iCol >= nLastCol) { + break; + } + SLastCol *pCol = taosArrayGet(pColArray, iCol); + if (pCol->colVal.cid != pTSchema->columns[iCol].colId) { + continue; + } tsdbRowGetColVal(pRow, pTSchema, iCol, pColVal); - SLastCol lastCol = {.ts = lastRowTs, .colVal = *pColVal}; + *pCol = (SLastCol){.ts = lastRowTs, .colVal = *pColVal}; if (IS_VAR_DATA_TYPE(pColVal->type) && pColVal->value.nData > 0) { - lastCol.colVal.value.pData = taosMemoryMalloc(lastCol.colVal.value.nData); - if (lastCol.colVal.value.pData == NULL) { + pCol->colVal.value.pData = taosMemoryMalloc(pCol->colVal.value.nData); + if (pCol->colVal.value.pData == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY; goto _err; } - memcpy(lastCol.colVal.value.pData, pColVal->value.pData, pColVal->value.nData); - } - - if (taosArrayPush(pColArray, &lastCol) == NULL) { - code = TSDB_CODE_OUT_OF_MEMORY; - goto _err; + memcpy(pCol->colVal.value.pData, pColVal->value.pData, pColVal->value.nData); } if (!COL_VAL_IS_VALUE(pColVal) && !setNoneCol) { @@ -1461,10 +1506,16 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SCach // merge into pColArray setNoneCol = false; - for (iCol = noneCol; iCol < nCol; ++iCol) { + for (int16_t iCol = noneCol; iCol < nCol; ++iCol) { + if (iCol >= nLastCol) { + break; + } // high version's column value SLastCol *lastColVal = (SLastCol *)taosArrayGet(pColArray, iCol); - SColVal *tColVal = &lastColVal->colVal; + if (lastColVal->colVal.cid != pTSchema->columns[iCol].colId) { + continue; + } + SColVal *tColVal = &lastColVal->colVal; tsdbRowGetColVal(pRow, pTSchema, iCol, pColVal); if (!COL_VAL_IS_VALUE(tColVal) && COL_VAL_IS_VALUE(pColVal)) { diff --git a/source/dnode/vnode/src/tsdb/tsdbCacheRead.c b/source/dnode/vnode/src/tsdb/tsdbCacheRead.c index 97b13a3c37..5bb224799a 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCacheRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbCacheRead.c @@ -209,6 +209,8 @@ void* tsdbCacherowsReaderClose(void* pReader) { taosMemoryFree(p->pSchema); } + taosMemoryFree(p->pCurrSchema); + destroyLastBlockLoadInfo(p->pLoadInfo); taosMemoryFree((void*)p->idstr); @@ -303,7 +305,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32 for (int32_t i = 0; i < pr->pSchema->numOfCols; ++i) { struct STColumn* pCol = &pr->pSchema->columns[i]; - SLastCol p = {.ts = INT64_MIN, .colVal.type = pCol->type}; + SLastCol p = {.ts = INT64_MIN, .colVal.type = pCol->type, .colVal.flag = CV_FLAG_NULL}; if (IS_VAR_DATA_TYPE(pCol->type)) { p.colVal.value.pData = taosMemoryCalloc(pCol->bytes, sizeof(char)); -- GitLab