Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
未验证 提交 66b7d908 编写于 作者: W wade zhang 提交者： GitHub
浏览文件
操作

Merge pull request #21479 from taosdata/fix/TD-24405 

fix(cache): fix deleted data
上级 87957e2b 34b9b322
显示空白变更内容
内联 并排
Showing with 60 addition and 45 deletion
source/dnode/vnode/src/tsdb/tsdbCache.c
浏览文件 @ 66b7d908
......@@ -703,6 +703,7 @@ static int32_t tsdbCacheLoadFromRaw(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArr
*pTmpLastCol = *pLastCol;
pLastCol = pTmpLastCol;
reallocVarData(&pLastCol->colVal);
size_t charge = sizeof(*pLastCol);
if (IS_VAR_DATA_TYPE(pLastCol->colVal.type)) {
charge += pLastCol->colVal.value.nData;
......@@ -789,7 +790,9 @@ static int32_t tsdbCacheLoadFromRocks(STsdb *pTsdb, tb_uid_t uid, SArray *pLastA
code = -1;
}
taosArraySet(pLastArray, idxKey->idx, pLastCol);
SLastCol lastCol = *pLastCol;
reallocVarData(&lastCol.colVal);
taosArraySet(pLastArray, idxKey->idx, &lastCol);
taosArrayRemove(remainCols, j);
taosMemoryFree(values_list[i]);
......@@ -825,7 +828,7 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
SLastCol *pLastCol = (SLastCol *)taosLRUCacheValue(pCache, h);
SLastCol lastCol = *pLastCol;
// reallocVarData(&lastCol.colVal);
reallocVarData(&lastCol.colVal);
taosArrayPush(pLastArray, &lastCol);
if (h) {
......@@ -853,8 +856,8 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
SLastCol lastCol = *pLastCol;
reallocVarData(&lastCol.colVal);
taosArraySet(pLastArray, idxKey->idx, &lastCol);
if (h) {
taosLRUCacheRelease(pCache, h, false);
}
......@@ -940,10 +943,10 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
int num_keys = pTSchema->numOfCols;
char **keys_list = taosMemoryCalloc(num_keys * 2, sizeof(char *));
size_t *keys_list_sizes = taosMemoryCalloc(num_keys * 2, sizeof(size_t));
const size_t klen = ROCKS_KEY_LEN;
for (int i = 0; i < num_keys; ++i) {
int16_t cid = pTSchema->columns[i].colId;
size_t klen = ROCKS_KEY_LEN;
char *keys = taosMemoryCalloc(2, sizeof(SLastKey));
((SLastKey *)keys)[0] = (SLastKey){.ltype = 1, .uid = uid, .cid = cid};
((SLastKey *)keys)[1] = (SLastKey){.ltype = 0, .uid = uid, .cid = cid};
......@@ -960,39 +963,35 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
rocksMayWrite(pTsdb, true, false, false);
rocksdb_multi_get(pTsdb->rCache.db, pTsdb->rCache.readoptions, num_keys * 2, (const char *const *)keys_list,
keys_list_sizes, values_list, values_list_sizes, errs);
for (int i = 0; i < num_keys; ++i) {
taosMemoryFree(keys_list[i]);
}
for (int i = 0; i < num_keys * 2; ++i) {
if (errs[i]) {
rocksdb_free(errs[i]);
}
taosMemoryFree(keys_list);
taosMemoryFree(keys_list_sizes);
}
taosMemoryFree(errs);
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
for (int i = 0; i < num_keys; ++i) {
SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i]);
if (NULL != pLastCol && (pLastCol->ts <= eKey && pLastCol->ts >= sKey)) {
SLastKey *key = &(SLastKey){.ltype = 1, .uid = uid, .cid = pLastCol->colVal.cid};
size_t klen = ROCKS_KEY_LEN;
rocksdb_writebatch_delete(wb, (char *)key, klen);
taosLRUCacheErase(pTsdb->lruCache, key, klen);
rocksdb_writebatch_delete(wb, keys_list[i], klen);
}
taosLRUCacheErase(pTsdb->lruCache, keys_list[i], klen);
pLastCol = tsdbCacheDeserialize(values_list[i + num_keys]);
if (NULL != pLastCol && (pLastCol->ts <= eKey && pLastCol->ts >= sKey)) {
SLastKey *key = &(SLastKey){.ltype = 0, .uid = uid, .cid = pLastCol->colVal.cid};
size_t klen = ROCKS_KEY_LEN;
rocksdb_writebatch_delete(wb, (char *)key, klen);
taosLRUCacheErase(pTsdb->lruCache, key, klen);
rocksdb_writebatch_delete(wb, keys_list[num_keys + i], klen);
}
taosLRUCacheErase(pTsdb->lruCache, keys_list[num_keys + i], klen);
rocksdb_free(values_list[i]);
rocksdb_free(values_list[i + num_keys]);
}
for (int i = 0; i < num_keys; ++i) {
taosMemoryFree(keys_list[i]);
}
taosMemoryFree(keys_list);
taosMemoryFree(keys_list_sizes);
taosMemoryFree(values_list);
taosMemoryFree(values_list_sizes);
......@@ -1871,10 +1870,14 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
if (isLast && (pColData->flag & HAS_VALUE)) {
skipBlock = false;
break;
} else if (pColData->flag & (HAS_VALUE | HAS_NULL)) {
} /*else if (pColData->flag & (HAS_VALUE | HAS_NULL)) {
skipBlock = false;
break;
}*/
}
if (!isLast) {
skipBlock = false;
}
if (skipBlock) {
......@@ -1908,6 +1911,9 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
if (checkRemainingRow) {
bool skipBlock = true;
int inputColIndex = 0;
if (aCols[0] == PRIMARYKEY_TIMESTAMP_COL_ID) {
++inputColIndex;
}
for (int colIndex = 0; colIndex < state->pBlockData->nColData; ++colIndex) {
SColData *pColData = &state->pBlockData->aColData[colIndex];
int16_t cid = pColData->cid;
......@@ -1916,15 +1922,19 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
if (isLast && (pColData->flag & HAS_VALUE)) {
skipBlock = false;
break;
} else if (pColData->flag & (HAS_VALUE | HAS_NULL)) {
} /*else if (pColData->flag & (HAS_VALUE | HAS_NULL)) {
skipBlock = false;
break;
}
}*/
++inputColIndex;
}
}
if (!isLast) {
skipBlock = false;
}
if (skipBlock) {
if (--state->iBlock < 0) {
tsdbDataFReaderClose(state->pDataFReader);
......@@ -2145,10 +2155,15 @@ static bool tsdbKeyDeleted(TSDBKEY *key, SArray *pSkyline, int64_t *iSkyline) {
return false;
} else if (key->ts >= pItemFront->ts && key->ts <= pItemBack->ts) {
if (key->version <= pItemFront->version || (key->ts == pItemBack->ts && key->version <= pItemBack->version)) {
// if (key->version <= pItemFront->version || key->version <= pItemBack->version) {
return true;
} else {
if (*iSkyline > 1) {
--*iSkyline;
} else {
return false;
}
}
} else {
if (*iSkyline > 1) {
--*iSkyline;
......@@ -2959,7 +2974,7 @@ static int32_t mergeLastRowCid(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray,
do {
TSDBROW *pRow = NULL;
nextRowIterGet(&iter, &pRow, &ignoreEarlierTs, true, TARRAY_DATA(aColArray), TARRAY_SIZE(aColArray));
nextRowIterGet(&iter, &pRow, &ignoreEarlierTs, false, TARRAY_DATA(aColArray), TARRAY_SIZE(aColArray));
if (!pRow) {
break;
......
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
浏览文件 @ 66b7d908
......@@ -315,14 +315,14 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
tsdbCacheGetBatch(pr->pTsdb, pKeyInfo->uid, pRow, pr, ltype);
// tsdbCacheGet(pr->pTsdb, pKeyInfo->uid, pRow, pr, ltype);
if (TARRAY_SIZE(pRow) <= 0) {
// taosArrayClearEx(pRow, freeItem);
taosArrayClear(pRow);
taosArrayClearEx(pRow, freeItem);
// taosArrayClear(pRow);
continue;
}
SLastCol* pColVal = taosArrayGet(pRow, 0);
if (COL_VAL_IS_NONE(&pColVal->colVal)) {
// taosArrayClearEx(pRow, freeItem);
taosArrayClear(pRow);
taosArrayClearEx(pRow, freeItem);
// taosArrayClear(pRow);
continue;
}
......@@ -381,8 +381,8 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
}
}
// taosArrayClearEx(pRow, freeItem);
taosArrayClear(pRow);
taosArrayClearEx(pRow, freeItem);
// taosArrayClear(pRow);
}
if (hasRes) {
......@@ -394,20 +394,20 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype);
if (TARRAY_SIZE(pRow) <= 0) {
// taosArrayClearEx(pRow, freeItem);
taosArrayClear(pRow);
taosArrayClearEx(pRow, freeItem);
// taosArrayClear(pRow);
continue;
}
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, 0);
if (COL_VAL_IS_NONE(&pColVal->colVal)) {
// taosArrayClearEx(pRow, freeItem);
taosArrayClear(pRow);
taosArrayClearEx(pRow, freeItem);
// taosArrayClear(pRow);
continue;
}
saveOneRow(pRow, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
// taosArrayClearEx(pRow, freeItem);
taosArrayClear(pRow);
taosArrayClearEx(pRow, freeItem);
// taosArrayClear(pRow);
taosArrayPush(pTableUidList, &uid);
......
source/dnode/vnode/src/tsdb/tsdbMemTable.c
浏览文件 @ 66b7d908
......@@ -190,9 +190,9 @@ int32_t tsdbDeleteTableData(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid
tsdbCacheDeleteLast(pTsdb->lruCache, pTbData->uid, eKey);
}
*/
if (eKey >= pTbData->maxKey && sKey <= pTbData->maxKey) {
// if (eKey >= pTbData->maxKey && sKey <= pTbData->maxKey) {
tsdbCacheDel(pTsdb, suid, uid, sKey, eKey);
}
//}
tsdbTrace("vgId:%d, delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
" at version %" PRId64,
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或