From 5147dc582a778ff80d756902649d30ef5433b6ec Mon Sep 17 00:00:00 2001
From: Shungang Li <sgli@taosdata.com>
Date: Fri, 30 Aug 2024 17:36:38 +0800
Subject: [PATCH] fix: (last) use insert to update LRU entry

---
 source/dnode/vnode/src/tsdb/tsdbCache.c | 40 +++++++++++--------------
 1 file changed, 18 insertions(+), 22 deletions(-)

diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c
index 586a9d7df9..114a1f7b89 100644
--- a/source/dnode/vnode/src/tsdb/tsdbCache.c
+++ b/source/dnode/vnode/src/tsdb/tsdbCache.c
@@ -757,24 +757,12 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
     rocksdb_free(values_list[0]);
     rocksdb_free(values_list[1]);
 
-    bool       erase = false;
-    LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[0], klen);
-    if (h) {
-      erase = true;
-      (void)taosLRUCacheRelease(pTsdb->lruCache, h, erase);
-    }
-    if (erase) {
-      taosLRUCacheErase(pTsdb->lruCache, keys_list[0], klen);
-    }
-
-    erase = false;
-    h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[1], klen);
-    if (h) {
-      erase = true;
-      (void)taosLRUCacheRelease(pTsdb->lruCache, h, erase);
-    }
-    if (erase) {
-      taosLRUCacheErase(pTsdb->lruCache, keys_list[1], klen);
+    for (int i = 0; i < 2; i++) {
+      LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[i], klen);
+      if (h) {
+        (void)taosLRUCacheRelease(pTsdb->lruCache, h, true);
+        taosLRUCacheErase(pTsdb->lruCache, keys_list[i], klen);
+      }
     }
   }
 
@@ -1027,6 +1015,7 @@ static int32_t tsdbCacheUpdateValue(SValue *pOld, SValue *pNew) {
   TAOS_RETURN(TSDB_CODE_SUCCESS);
 }
 
+#ifdef BUILD_NO_CALL
 static void tsdbCacheUpdateLastCol(SLastCol *pLastCol, SRowKey *pRowKey, SColVal *pColVal) {
   // update rowkey
   pLastCol->rowKey.ts = pRowKey->ts;
@@ -1047,6 +1036,7 @@ static void tsdbCacheUpdateLastCol(SLastCol *pLastCol, SRowKey *pRowKey, SColVal
     pLastCol->dirty = 1;
   }
 }
+#endif
 
 static void tsdbCacheUpdateLastColToNone(SLastCol *pLastCol, ELastCacheStatus cacheStatus) {
   // update rowkey
@@ -1108,11 +1098,12 @@ static int32_t tsdbCachePutToLRU(STsdb *pTsdb, SLastKey *pLastKey, SLastCol *pLa
 
   size_t charge = 0;
   *pLRULastCol = *pLastCol;
+  pLRULastCol->dirty = 1;
   TAOS_CHECK_EXIT(tsdbCacheReallocSLastCol(pLRULastCol, &charge));
 
   LRUStatus status = taosLRUCacheInsert(pTsdb->lruCache, pLastKey, ROCKS_KEY_LEN, pLRULastCol, charge, tsdbCacheDeleter,
                                         NULL, TAOS_LRU_PRIORITY_LOW, &pTsdb->flushState);
-  if (TAOS_LRU_STATUS_OK != status) {
+  if (TAOS_LRU_STATUS_OK != status && TAOS_LRU_STATUS_OK_OVERWRITTEN != status) {
     tsdbError("tsdb/cache/putlru: vgId:%d, failed to insert status %d.", TD_VID(pTsdb->pVnode), status);
     code = TSDB_CODE_INVALID_DATA_FMT;
   }
@@ -1157,12 +1148,13 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
       if (pLastCol->cacheStatus != TSDB_LAST_CACHE_NO_CACHE) {
         int32_t cmp_res = tRowKeyCompare(&pLastCol->rowKey, pRowKey);
         if (cmp_res < 0 || (cmp_res == 0 && !COL_VAL_IS_NONE(pColVal))) {
-          tsdbCacheUpdateLastCol(pLastCol, pRowKey, pColVal);
-          pLastCol->cacheStatus = TSDB_LAST_CACHE_VALID;
+          SLastCol newLastCol = {.rowKey = *pRowKey, .colVal = *pColVal, .cacheStatus = TSDB_LAST_CACHE_VALID};
+          code = tsdbCachePutToLRU(pTsdb, key, &newLastCol);
         }
       }
 
       (void)taosLRUCacheRelease(pCache, h, false);
+      TAOS_CHECK_EXIT(code);
     } else {
       if (!remainCols) {
         remainCols = taosArrayInit(num_keys * 2, sizeof(SIdxKey));
@@ -1880,9 +1872,13 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
       if (h) {
         SLastCol *pLastCol = (SLastCol *)taosLRUCacheValue(pTsdb->lruCache, h);
         if (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey) {
-          tsdbCacheUpdateLastColToNone(pLastCol, TSDB_LAST_CACHE_NO_CACHE);
+          SLastCol noneCol = {.rowKey.ts = TSKEY_MIN,
+                              .colVal = COL_VAL_NONE(cid, pTSchema->columns[i].type),
+                              .cacheStatus = TSDB_LAST_CACHE_NO_CACHE};
+          code = tsdbCachePutToLRU(pTsdb, &lastKey, &noneCol);
         }
         (void)taosLRUCacheRelease(pTsdb->lruCache, h, false);
+        TAOS_CHECK_EXIT(code);
       } else {
         if (!remainCols) {
           remainCols = taosArrayInit(numCols * 2, sizeof(SLastKey));