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

cache/read: use new interface with signle query

This commit is contained in:
Minglei Jin 2023-04-21 16:07:22 +08:00
parent 0394c6a50c
commit 01d95b5e95
2 changed files with 35 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -273,6 +273,8 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow
taosMemoryFree(value); taosMemoryFree(value);
} }
} }
rocksdb_free(values_list[i]);
} }
taosMemoryFree(values_list); taosMemoryFree(values_list);
taosMemoryFree(values_list_sizes); taosMemoryFree(values_list_sizes);

102
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
View File

@ -199,7 +199,7 @@ void* tsdbCacherowsReaderClose(void* pReader) {
taosMemoryFree(pReader); taosMemoryFree(pReader);
return NULL; return NULL;
} }
/*
static int32_t doExtractCacheRow(SCacheRowsReader* pr, SLRUCache* lruCache, uint64_t uid, SArray** pRow, static int32_t doExtractCacheRow(SCacheRowsReader* pr, SLRUCache* lruCache, uint64_t uid, SArray** pRow,
LRUHandle** h) { LRUHandle** h) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
@ -222,7 +222,7 @@ static int32_t doExtractCacheRow(SCacheRowsReader* pr, SLRUCache* lruCache, uint
return code; return code;
} }
*/
static void freeItem(void* pItem) { static void freeItem(void* pItem) {
SLastCol* pCol = (SLastCol*)pItem; SLastCol* pCol = (SLastCol*)pItem;
if (IS_VAR_DATA_TYPE(pCol->colVal.type)) { if (IS_VAR_DATA_TYPE(pCol->colVal.type)) {
@ -279,7 +279,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
p->ts = INT64_MIN; p->ts = INT64_MIN;
} }
pLastCols = taosArrayInit(pr->pSchema->numOfCols, sizeof(SLastCol)); pLastCols = taosArrayInit(pr->numOfCols, sizeof(SLastCol));
if (pLastCols == NULL) { if (pLastCols == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _end; goto _end;
@ -303,6 +303,8 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
pr->pDataFReader = NULL; pr->pDataFReader = NULL;
pr->pDataFReaderLast = NULL; pr->pDataFReaderLast = NULL;
char const* lstring = pr->type & CACHESCAN_RETRIEVE_LAST ? "last" : "last_row";
// retrieve the only one last row of all tables in the uid list. // retrieve the only one last row of all tables in the uid list.
if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) { if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) {
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
@ -310,16 +312,9 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
for (int32_t i = 0; i < pr->numOfTables; ++i) { for (int32_t i = 0; i < pr->numOfTables; ++i) {
STableKeyInfo* pKeyInfo = &pr->pTableList[i]; STableKeyInfo* pKeyInfo = &pr->pTableList[i];
code = doExtractCacheRow(pr, lruCache, pKeyInfo->uid, &pRow, &h); tsdbCacheGet(pr->pTsdb, pKeyInfo->uid, &pRow, pr, lstring);
if (code != TSDB_CODE_SUCCESS) { if (TARRAY_SIZE(pRow) <= 0) {
goto _end; taosArrayDestroy(pRow);
}
if (h == NULL) {
continue;
}
if (taosArrayGetSize(pRow) <= 0) {
tsdbCacheRelease(lruCache, h);
continue; continue;
} }
@ -327,47 +322,34 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
bool hasNotNullRow = true; bool hasNotNullRow = true;
int64_t singleTableLastTs = INT64_MAX; int64_t singleTableLastTs = INT64_MAX;
for (int32_t k = 0; k < pr->numOfCols; ++k) { for (int32_t k = 0; k < pr->numOfCols; ++k) {
int32_t slotId = slotIds[k]; SLastCol* p = taosArrayGet(pLastCols, k);
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, k);
if (slotId == -1) { // the primary timestamp if (pColVal->ts > p->ts) {
SLastCol* p = taosArrayGet(pLastCols, 0); if (!COL_VAL_IS_VALUE(&pColVal->colVal) && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
SLastCol* pCol = (SLastCol*)taosArrayGet(pRow, 0); if (!COL_VAL_IS_VALUE(&p->colVal)) {
if (pCol->ts > p->ts) { hasNotNullRow = false;
hasRes = true; }
p->ts = pCol->ts; continue;
p->colVal = pCol->colVal;
singleTableLastTs = pCol->ts;
} }
} else {
SLastCol* p = taosArrayGet(pLastCols, slotId);
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, slotId);
if (pColVal->ts > p->ts) { hasRes = true;
if (!COL_VAL_IS_VALUE(&pColVal->colVal) && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) { p->ts = pColVal->ts;
if (!COL_VAL_IS_VALUE(&p->colVal)) { if (pColVal->ts < singleTableLastTs && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
hasNotNullRow = false; singleTableLastTs = pColVal->ts;
} }
continue;
if (!IS_VAR_DATA_TYPE(pColVal->colVal.type)) {
p->colVal = pColVal->colVal;
} else {
if (COL_VAL_IS_VALUE(&pColVal->colVal)) {
memcpy(p->colVal.value.pData, pColVal->colVal.value.pData, pColVal->colVal.value.nData);
} }
hasRes = true; p->colVal.value.nData = pColVal->colVal.value.nData;
p->ts = pColVal->ts; p->colVal.type = pColVal->colVal.type;
if (pColVal->ts < singleTableLastTs && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) { p->colVal.flag = pColVal->colVal.flag;
singleTableLastTs = pColVal->ts; p->colVal.cid = pColVal->colVal.cid;
}
if (!IS_VAR_DATA_TYPE(pColVal->colVal.type)) {
p->colVal = pColVal->colVal;
} else {
if (COL_VAL_IS_VALUE(&pColVal->colVal)) {
memcpy(p->colVal.value.pData, pColVal->colVal.value.pData, pColVal->colVal.value.nData);
}
p->colVal.value.nData = pColVal->colVal.value.nData;
p->colVal.type = pColVal->colVal.type;
p->colVal.flag = pColVal->colVal.flag;
p->colVal.cid = pColVal->colVal.cid;
}
} }
} }
} }
@ -389,7 +371,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
taosArraySet(pTableUidList, 0, &pKeyInfo->uid); taosArraySet(pTableUidList, 0, &pKeyInfo->uid);
} }
tsdbCacheRelease(lruCache, h); taosArrayDestroy(pRow);
} }
if (hasRes) { if (hasRes) {
@ -398,28 +380,10 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
} else if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_ALL)) { } else if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_ALL)) {
for (int32_t i = pr->tableIndex; i < pr->numOfTables; ++i) { for (int32_t i = pr->tableIndex; i < pr->numOfTables; ++i) {
STableKeyInfo* pKeyInfo = &pr->pTableList[i]; STableKeyInfo* pKeyInfo = &pr->pTableList[i];
/*
code = doExtractCacheRow(pr, lruCache, pKeyInfo->uid, &pRow, &h);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
if (h == NULL) {
continue;
}
if (taosArrayGetSize(pRow) <= 0) {
tsdbCacheRelease(lruCache, h);
continue;
}
saveOneRow(pRow, pResBlock, pr, slotIds, pRes, pr->idstr);
// TODO reset the pRes
tsdbCacheRelease(lruCache, h);
*/
char const* lstring = pr->type & CACHESCAN_RETRIEVE_LAST ? "last" : "last_row";
tsdbCacheGet(pr->pTsdb, pKeyInfo->uid, &pRow, pr, lstring); tsdbCacheGet(pr->pTsdb, pKeyInfo->uid, &pRow, pr, lstring);
if (TARRAY_SIZE(pRow) <= 0) { if (TARRAY_SIZE(pRow) <= 0) {
taosArrayDestroy(pRow);
continue; continue;
} }