Merge remote-tracking branch 'origin/hotfix/TS-544' into fix/TD-5903

cmake/install.inc

 IF (TD_LINUX)
   SET(TD_MAKE_INSTALL_SH "${TD_COMMUNITY_DIR}/packaging/tools/make_install.sh")
   INSTALL(CODE "MESSAGE(\"make install script: ${TD_MAKE_INSTALL_SH}\")")
-  INSTALL(CODE "execute_process(COMMAND chmod 777 ${TD_MAKE_INSTALL_SH})")
-  INSTALL(CODE "execute_process(COMMAND ${TD_MAKE_INSTALL_SH} ${TD_COMMUNITY_DIR} ${PROJECT_BINARY_DIR} Linux ${TD_VER_NUMBER})")
+  INSTALL(CODE "execute_process(COMMAND bash ${TD_MAKE_INSTALL_SH} ${TD_COMMUNITY_DIR} ${PROJECT_BINARY_DIR} Linux ${TD_VER_NUMBER})")
 ELSEIF (TD_WINDOWS)
   IF (TD_POWER)
     SET(CMAKE_INSTALL_PREFIX C:/PowerDB)
@@ -41,6 +40,5 @@ ELSEIF (TD_WINDOWS)
 ELSEIF (TD_DARWIN)
   SET(TD_MAKE_INSTALL_SH "${TD_COMMUNITY_DIR}/packaging/tools/make_install.sh")
   INSTALL(CODE "MESSAGE(\"make install script: ${TD_MAKE_INSTALL_SH}\")")
-  INSTALL(CODE "execute_process(COMMAND chmod 777 ${TD_MAKE_INSTALL_SH})")
-  INSTALL(CODE "execute_process(COMMAND ${TD_MAKE_INSTALL_SH} ${TD_COMMUNITY_DIR} ${PROJECT_BINARY_DIR} Darwin ${TD_VER_NUMBER})")
+  INSTALL(CODE "execute_process(COMMAND bash ${TD_MAKE_INSTALL_SH} ${TD_COMMUNITY_DIR} ${PROJECT_BINARY_DIR} Darwin ${TD_VER_NUMBER})")
 ENDIF ()

src/client/src/tscSQLParser.c

@@ -5610,7 +5610,7 @@ int32_t validateOrderbyNode(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SSqlNode* pSq
   bool udf = false;
 
   if (pQueryInfo->pUdfInfo && taosArrayGetSize(pQueryInfo->pUdfInfo) > 0) {
-    int32_t usize = taosArrayGetSize(pQueryInfo->pUdfInfo);
+    int32_t usize = (int32_t)taosArrayGetSize(pQueryInfo->pUdfInfo);
     
     for (int32_t i = 0; i < usize; ++i) {
       SUdfInfo* pUdfInfo = taosArrayGet(pQueryInfo->pUdfInfo, i);
@@ -8406,7 +8406,7 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
           
           for (int32_t j = 0; j < usize; ++j) {
             SUdfInfo* pUdfInfo = taosArrayGet(pQueryInfo->pUdfInfo, j);
-            int32_t len = strlen(pUdfInfo->name);
+            int32_t len = (int32_t)strlen(pUdfInfo->name);
             if (len == t->n && strncasecmp(info.name, pUdfInfo->name, t->n) == 0) {
               exist = 1;
               break;

src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/Utils.java

@@ -121,7 +121,7 @@ public class Utils {
     }
 
     private static void findValuesClauseRangeSet(String preparedSql, RangeSet<Integer> clauseRangeSet) {
-        Matcher matcher = Pattern.compile("(values|,)\\s*(\\([^)]*\\))").matcher(preparedSql);
+        Matcher matcher = Pattern.compile("(values||,)\\s*(\\([^)]*\\))").matcher(preparedSql);
         while (matcher.find()) {
             int start = matcher.start(2);
             int end = matcher.end(2);

src/connector/jdbc/src/test/java/com/taosdata/jdbc/utils/UtilsTest.java

@@ -73,6 +73,48 @@ public class UtilsTest {
         Assert.assertEquals(expected, actual);
     }
 
+    @Test
+    public void multiValuesAndWhitespace() {
+        // given
+        String nativeSql = "INSERT INTO ? (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (?)  VALUES (?,?,?,?)  (?,?,?,?) (?,?,?,?)";
+        Object[] parameters = Stream.of("d1", 1, 100, 3.14, "abc", 4, 200, 3.1415, "xyz", 5, 300, 3.141592, "uvw", 6).toArray();
+
+        // when
+        String actual = Utils.getNativeSql(nativeSql, parameters);
+
+        // then
+        String expected = "INSERT INTO d1 (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (1)  VALUES (100,3.14,'abc',4)  (200,3.1415,'xyz',5) (300,3.141592,'uvw',6)";
+        Assert.assertEquals(expected, actual);
+    }
+
+    @Test
+    public void multiValuesNoSeparator() {
+        // given
+        String nativeSql = "INSERT INTO ? (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (?)  VALUES (?,?,?,?)(?,?,?,?)(?,?,?,?)";
+        Object[] parameters = Stream.of("d1", 1, 100, 3.14, "abc", 4, 200, 3.1415, "xyz", 5, 300, 3.141592, "uvw", 6).toArray();
+
+        // when
+        String actual = Utils.getNativeSql(nativeSql, parameters);
+
+        // then
+        String expected = "INSERT INTO d1 (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (1)  VALUES (100,3.14,'abc',4)(200,3.1415,'xyz',5)(300,3.141592,'uvw',6)";
+        Assert.assertEquals(expected, actual);
+    }
+
+    @Test
+    public void multiValuesMultiSeparator() {
+        // given
+        String nativeSql = "INSERT INTO ? (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (?)  VALUES (?,?,?,?) (?,?,?,?), (?,?,?,?)";
+        Object[] parameters = Stream.of("d1", 1, 100, 3.14, "abc", 4, 200, 3.1415, "xyz", 5, 300, 3.141592, "uvw", 6).toArray();
+
+        // when
+        String actual = Utils.getNativeSql(nativeSql, parameters);
+
+        // then
+        String expected = "INSERT INTO d1 (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (1)  VALUES (100,3.14,'abc',4) (200,3.1415,'xyz',5), (300,3.141592,'uvw',6)";
+        Assert.assertEquals(expected, actual);
+    }
+
     @Test
     public void lineTerminator() {
         // given
@@ -100,6 +142,32 @@ public class UtilsTest {
         Assert.assertEquals(expected, actual);
     }
 
+    @Test
+    public void lineTerminatorAndMultiValuesAndNoneOrMoreWhitespace() {
+        String nativeSql = "INSERT Into ? TAGS(?) VALUES(?,?,\r\n?,?),(?,? ,\r\n?,?) t? tags (?) Values (?,?,?\r\n,?) (?,?,?,?) t? Tags(?) values  (?,?,?,?) , (?,?,?,?)";
+        Object[] parameters = Stream.of("t1", "abc", 100, 1.1, "xxx", "xxx", 200, 2.2, "xxx", "xxx", 2, "bcd", 300, 3.3, "xxx", "xxx", 400, 4.4, "xxx", "xxx", 3, "cde", 500, 5.5, "xxx", "xxx", 600, 6.6, "xxx", "xxx").toArray();
+
+        // when
+        String actual = Utils.getNativeSql(nativeSql, parameters);
+
+        // then
+        String expected = "INSERT Into t1 TAGS('abc') VALUES(100,1.1,\r\n'xxx','xxx'),(200,2.2 ,\r\n'xxx','xxx') t2 tags ('bcd') Values (300,3.3,'xxx'\r\n,'xxx') (400,4.4,'xxx','xxx') t3 Tags('cde') values  (500,5.5,'xxx','xxx') , (600,6.6,'xxx','xxx')";
+        Assert.assertEquals(expected, actual);
+    }
+
+    @Test
+    public void multiValuesAndNoneOrMoreWhitespace() {
+        String nativeSql = "INSERT INTO ? USING traces TAGS (?, ?) VALUES (?, ?, ?, ?, ?, ?, ?)  (?, ?, ?, ?, ?, ?, ?)";
+        Object[] parameters = Stream.of("t1", "t1", "t2", 1632968284000L, 111.111, 119.001, 0.4, 90, 99.1, "WGS84", 1632968285000L, 111.21109999999999, 120.001, 0.5, 91, 99.19999999999999, "WGS84").toArray();
+
+        // when
+        String actual = Utils.getNativeSql(nativeSql, parameters);
+
+        // then
+        String expected = "INSERT INTO t1 USING traces TAGS ('t1', 't2') VALUES (1632968284000, 111.111, 119.001, 0.4, 90, 99.1, 'WGS84')  (1632968285000, 111.21109999999999, 120.001, 0.5, 91, 99.19999999999999, 'WGS84')";
+        Assert.assertEquals(expected, actual);
+    }
+
     @Test
     public void replaceNothing() {
         // given

src/kit/shell/CMakeLists.txt

@@ -26,6 +26,8 @@ ENDIF ()
 
   SET_TARGET_PROPERTIES(shell PROPERTIES OUTPUT_NAME taos)
 ELSEIF (TD_WINDOWS)
+  ADD_DEFINITIONS(-DUNICODE)
+  ADD_DEFINITIONS(-D_UNICODE)
   LIST(APPEND SRC ./src/shellEngine.c)
   LIST(APPEND SRC ./src/shellMain.c)
   LIST(APPEND SRC ./src/shellWindows.c)

src/kit/shell/src/shellMain.c

@@ -95,6 +95,9 @@ SShellArguments args = {
  */
 int main(int argc, char* argv[]) {
   /*setlocale(LC_ALL, "en_US.UTF-8"); */
+#ifdef WINDOWS
+  SetConsoleOutputCP(CP_UTF8);
+#endif
 
   if (!checkVersion()) {
     exit(EXIT_FAILURE);

src/kit/shell/src/shellWindows.c

@@ -256,13 +256,16 @@ int32_t shellReadCommand(TAOS *con, char command[]) {
   cmd.command = (char *)calloc(1, MAX_COMMAND_SIZE);
 
   // Read input.
-  char c;
+  void *console = GetStdHandle(STD_INPUT_HANDLE);
+  unsigned long read;
+  wchar_t c;
+  char mbStr[16];
   while (1) {
-    c = getchar();
-
+    int ret = ReadConsole(console, &c, 1, &read, NULL);
+    int size = WideCharToMultiByte(CP_UTF8, 0, &c, read, mbStr, sizeof(mbStr), NULL, NULL);
+    mbStr[size] = 0;
     switch (c) {
       case '\n':
-      case '\r':
         if (isReadyGo(&cmd)) {
           sprintf(command, "%s%s", cmd.buffer, cmd.command);
           free(cmd.buffer);
@@ -275,8 +278,12 @@ int32_t shellReadCommand(TAOS *con, char command[]) {
           updateBuffer(&cmd);
         }
         break;
+      case '\r':
+        break;
       default:
-        insertChar(&cmd, c);
+        for (int i = 0; i < size; ++i) {
+          insertChar(&cmd, mbStr[i]);
+        }
     }
   }
 

src/tsdb/inc/tsdbMeta.h

@@ -41,6 +41,7 @@ typedef struct STable {
   int16_t        restoreColumnNum;
   bool           hasRestoreLastColumn;
   int            lastColSVersion;
+  int16_t        cacheLastConfigVersion;
   T_REF_DECLARE()
 } STable;
 

src/tsdb/inc/tsdbint.h

@@ -79,8 +79,8 @@ struct STsdbRepo {
   STsdbCfg        save_config;    // save apply config
   bool            config_changed; // config changed flag
   pthread_mutex_t save_mutex;     // protect save config
-  
-  uint8_t         hasCachedLastColumn;
+
+  int16_t         cacheLastConfigVersion;
 
   STsdbAppH       appH;
   STsdbStat       stat;
@@ -110,7 +110,8 @@ int        tsdbUnlockRepo(STsdbRepo* pRepo);
 STsdbMeta* tsdbGetMeta(STsdbRepo* pRepo);
 int        tsdbCheckCommit(STsdbRepo* pRepo);
 int        tsdbRestoreInfo(STsdbRepo* pRepo);
-int        tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg);
+UNUSED_FUNC int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg);
+int32_t    tsdbLoadLastCache(STsdbRepo *pRepo, STable* pTable);
 void       tsdbGetRootDir(int repoid, char dirName[]);
 void       tsdbGetDataDir(int repoid, char dirName[]);
 

src/tsdb/src/tsdbCommitQueue.c

@@ -146,7 +146,9 @@ static void tsdbApplyRepoConfig(STsdbRepo *pRepo) {
 
   if (oldCfg.cacheLastRow != pRepo->config.cacheLastRow) {
     if (tsdbLockRepo(pRepo) < 0) return;
-    tsdbCacheLastData(pRepo, &oldCfg);
+    // tsdbCacheLastData(pRepo, &oldCfg);
+    // lazy load last cache when query or update
+    ++pRepo->cacheLastConfigVersion;
     tsdbUnlockRepo(pRepo);
   }
 

src/tsdb/src/tsdbCompact.c

@@ -64,7 +64,9 @@ int tsdbCompact(STsdbRepo *pRepo) { return tsdbAsyncCompact(pRepo); }
 void *tsdbCompactImpl(STsdbRepo *pRepo) {
   // Check if there are files in TSDB FS to compact
   if (REPO_FS(pRepo)->cstatus->pmf == NULL) {
-    tsdbInfo("vgId:%d no file to compact in FS", REPO_ID(pRepo));
+    pRepo->compactState = TSDB_NO_COMPACT;
+    tsem_post(&(pRepo->readyToCommit));
+    tsdbInfo("vgId:%d compact over, no file to compact in FS", REPO_ID(pRepo));
     return NULL;
   }
 

src/tsdb/src/tsdbMain.c

@@ -562,7 +562,7 @@ static STsdbRepo *tsdbNewRepo(STsdbCfg *pCfg, STsdbAppH *pAppH) {
     return NULL;
   }
   pRepo->config_changed = false;
-  atomic_store_8(&pRepo->hasCachedLastColumn, 0);
+  pRepo->cacheLastConfigVersion = 0;
 
   code = tsem_init(&(pRepo->readyToCommit), 0, 1);
   if (code != 0) {
@@ -711,7 +711,7 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
       tsdbGetBlockStatis(pReadh, pBlockStatis, (int)numColumns);
       loadStatisData = true;
     }
-
+    TSDB_WLOCK_TABLE(pTable);  // lock when update pTable->lastCols[]
     for (int16_t i = 0; i < numColumns && numColumns > pTable->restoreColumnNum; ++i) {
       STColumn *pCol = schemaColAt(pSchema, i);
       // ignore loaded columns
@@ -760,6 +760,7 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
         break;
       }
     }
+    TSDB_WUNLOCK_TABLE(pTable);
   }
 
 out:
@@ -774,7 +775,6 @@ out:
 }
 
 static int tsdbRestoreLastRow(STsdbRepo *pRepo, STable *pTable, SReadH* pReadh, SBlockIdx *pIdx) {
-  ASSERT(pTable->lastRow == NULL);
   if (tsdbLoadBlockInfo(pReadh, NULL) < 0) {
     return -1;
   }
@@ -788,21 +788,32 @@ static int tsdbRestoreLastRow(STsdbRepo *pRepo, STable *pTable, SReadH* pReadh,
   // Get the data in row
   
   STSchema *pSchema = tsdbGetTableSchema(pTable);
-  pTable->lastRow = taosTMalloc(memRowMaxBytesFromSchema(pSchema));
-  if (pTable->lastRow == NULL) {
+  SMemRow   lastRow = taosTMalloc(memRowMaxBytesFromSchema(pSchema));
+  if (lastRow == NULL) {
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     return -1;
   }
-  memRowSetType(pTable->lastRow, SMEM_ROW_DATA);
-  tdInitDataRow(memRowDataBody(pTable->lastRow), pSchema);
+  memRowSetType(lastRow, SMEM_ROW_DATA);
+  tdInitDataRow(memRowDataBody(lastRow), pSchema);
   for (int icol = 0; icol < schemaNCols(pSchema); icol++) {
     STColumn *pCol = schemaColAt(pSchema, icol);
     SDataCol *pDataCol = pReadh->pDCols[0]->cols + icol;
-    tdAppendColVal(memRowDataBody(pTable->lastRow), tdGetColDataOfRow(pDataCol, pBlock->numOfRows - 1), pCol->type,
+    tdAppendColVal(memRowDataBody(lastRow), tdGetColDataOfRow(pDataCol, pBlock->numOfRows - 1), pCol->type,
                    pCol->offset);
   }
 
-  pTable->lastKey = memRowKey(pTable->lastRow);
+  TSKEY lastKey = memRowKey(lastRow);
+  
+  // during the load data in file, new data would be inserted and last row has been updated
+  TSDB_WLOCK_TABLE(pTable);
+  if (pTable->lastRow == NULL) {
+    pTable->lastKey = lastKey;
+    pTable->lastRow = lastRow;
+    TSDB_WUNLOCK_TABLE(pTable);
+  } else {
+    TSDB_WUNLOCK_TABLE(pTable);
+    taosTZfree(lastRow);
+  }
 
   return 0;
 }
@@ -874,14 +885,105 @@ int tsdbRestoreInfo(STsdbRepo *pRepo) {
 
   tsdbDestroyReadH(&readh);
 
-  if (CACHE_LAST_NULL_COLUMN(pCfg)) {
-    atomic_store_8(&pRepo->hasCachedLastColumn, 1);
+  // if (CACHE_LAST_NULL_COLUMN(pCfg)) {
+  //   atomic_store_8(&pRepo->hasCachedLastColumn, 1);
+  // }
+
+  return 0;
+}
+
+int32_t tsdbLoadLastCache(STsdbRepo *pRepo, STable *pTable) {
+  SFSIter    fsiter;
+  SReadH     readh;
+  SDFileSet *pSet;
+  int cacheLastRowTableNum = 0;
+  int cacheLastColTableNum = 0;
+
+  bool cacheLastRow = CACHE_LAST_ROW(&(pRepo->config));
+  bool cacheLastCol = CACHE_LAST_NULL_COLUMN(&(pRepo->config));
+
+  tsdbDebug("tsdbLoadLastCache for %s, cacheLastRow:%d, cacheLastCol:%d", pTable->name->data, cacheLastRow, cacheLastCol);
+
+  pTable->cacheLastConfigVersion = pRepo->cacheLastConfigVersion;
+
+  if (!cacheLastRow && pTable->lastRow != NULL) {
+    taosTZfree(pTable->lastRow);
+    pTable->lastRow = NULL;
   }
+  if (!cacheLastCol && pTable->lastCols != NULL) {
+    tsdbFreeLastColumns(pTable);
+  }
+
+  if (!cacheLastRow && !cacheLastCol) {
+    return 0;
+  }
+
+  cacheLastRowTableNum = (cacheLastRow && pTable->lastRow  == NULL) ? 1 : 0;
+  cacheLastColTableNum = (cacheLastCol && pTable->lastCols == NULL) ? 1 : 0;
+
+  if (cacheLastRowTableNum == 0 && cacheLastColTableNum == 0) {
+    return 0;
+  }
+
+  if (tsdbInitReadH(&readh, pRepo) < 0) {
+    return -1;
+  }
+
+  tsdbRLockFS(REPO_FS(pRepo));
+  tsdbFSIterInit(&fsiter, REPO_FS(pRepo), TSDB_FS_ITER_BACKWARD);
+
+  while ((cacheLastRowTableNum > 0 || cacheLastColTableNum > 0) && (pSet = tsdbFSIterNext(&fsiter)) != NULL) {
+    if (tsdbSetAndOpenReadFSet(&readh, pSet) < 0) {
+      tsdbUnLockFS(REPO_FS(pRepo));
+      tsdbDestroyReadH(&readh);
+      return -1;
+    }
+
+    if (tsdbLoadBlockIdx(&readh) < 0) {
+      tsdbUnLockFS(REPO_FS(pRepo));
+      tsdbDestroyReadH(&readh);
+      return -1;
+    }
+
+    // tsdbDebug("tsdbRestoreInfo restore vgId:%d,table:%s", REPO_ID(pRepo), pTable->name->data);
+
+    if (tsdbSetReadTable(&readh, pTable) < 0) {
+      tsdbUnLockFS(REPO_FS(pRepo));
+      tsdbDestroyReadH(&readh);
+      return -1;
+    }
+
+    SBlockIdx *pIdx = readh.pBlkIdx;
+
+    if (pIdx && (cacheLastRowTableNum > 0) && (pTable->lastRow == NULL)) {
+      if (tsdbRestoreLastRow(pRepo, pTable, &readh, pIdx) != 0) {
+        tsdbUnLockFS(REPO_FS(pRepo));
+        tsdbDestroyReadH(&readh);
+        return -1;
+      }
+      cacheLastRowTableNum -= 1;
+    }
+
+    // restore NULL columns
+    if (pIdx && (cacheLastColTableNum > 0) && !pTable->hasRestoreLastColumn) {
+      if (tsdbRestoreLastColumns(pRepo, pTable, &readh) != 0) {
+        tsdbUnLockFS(REPO_FS(pRepo));
+        tsdbDestroyReadH(&readh);
+        return -1;
+      }
+      if (pTable->hasRestoreLastColumn) {
+        cacheLastColTableNum -= 1;
+      }
+    }
+  }
+
+  tsdbUnLockFS(REPO_FS(pRepo));
+  tsdbDestroyReadH(&readh);
 
   return 0;
 }
 
-int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
+UNUSED_FUNC int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
   bool cacheLastRow = false, cacheLastCol = false;
   SFSIter    fsiter;
   SReadH     readh;
@@ -915,9 +1017,9 @@ int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
 
   // if close last option,need to free data
   if (need_free_last_row || need_free_last_col) {
-    if (need_free_last_col) {
-      atomic_store_8(&pRepo->hasCachedLastColumn, 0);
-    }
+    // if (need_free_last_col) {
+    //   atomic_store_8(&pRepo->hasCachedLastColumn, 0);
+    // }
     tsdbInfo("free cache last data since cacheLast option changed");    
     for (int i = 1; i <= maxTableIdx; i++) {
       STable *pTable = pMeta->tables[i];
@@ -995,9 +1097,9 @@ int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
 
   tsdbDestroyReadH(&readh);
 
-  if (cacheLastCol) {
-    atomic_store_8(&pRepo->hasCachedLastColumn, 1);
-  }
+  // if (cacheLastCol) {
+  //   atomic_store_8(&pRepo->hasCachedLastColumn, 1);
+  // }
   
   return 0;
 }

src/tsdb/src/tsdbMemTable.c

@@ -996,7 +996,8 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SMemRow ro
     if ((value == NULL) || isNull(value, pTCol->type)) {
       continue;
     }
-
+    // lock
+    TSDB_WLOCK_TABLE(pTable); 
     SDataCol *pDataCol = &(pLatestCols[idx]);
     if (pDataCol->pData == NULL) {
       pDataCol->pData = malloc(pTCol->bytes);
@@ -1012,6 +1013,8 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SMemRow ro
     memcpy(pDataCol->pData, value, bytes);
     //tsdbInfo("updateTableLatestColumn vgId:%d cache column %d for %d,%s", REPO_ID(pRepo), j, pDataCol->bytes, (char*)pDataCol->pData);
     pDataCol->ts = memRowKey(row);
+    // unlock
+    TSDB_WUNLOCK_TABLE(pTable); 
   }
 }
 
@@ -1058,5 +1061,8 @@ static int tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SMemRow r
       updateTableLatestColumn(pRepo, pTable, row);
     }
   }
+
+  pTable->cacheLastConfigVersion = pRepo->cacheLastConfigVersion;
+
   return 0;
 }

src/tsdb/src/tsdbMeta.c

@@ -346,7 +346,7 @@ int tsdbUpdateTableTagValue(STsdbRepo *repo, SUpdateTableTagValMsg *pMsg) {
     tsdbError(
         "vgId:%d failed to update tag value of table %s since version out of date, client tag version %d server tag "
         "version %d",
-        REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), pMsg->tversion, schemaVersion(pTable->tagSchema));
+        REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), pMsg->tversion, schemaVersion(pTable->pSuper->tagSchema));
     terrno = TSDB_CODE_TDB_TAG_VER_OUT_OF_DATE;
     return -1;
   }
@@ -627,27 +627,30 @@ int16_t tsdbGetLastColumnsIndexByColId(STable* pTable, int16_t colId) {
 }
 
 int tsdbInitColIdCacheWithSchema(STable* pTable, STSchema* pSchema) {
-  ASSERT(pTable->lastCols == NULL);
+  TSDB_WLOCK_TABLE(pTable);
+  if (pTable->lastCols == NULL) {
+    int16_t numOfColumn = pSchema->numOfCols;
 
-  int16_t numOfColumn = pSchema->numOfCols;
+    pTable->lastCols = (SDataCol *)malloc(numOfColumn * sizeof(SDataCol));
+    if (pTable->lastCols == NULL) {
+      TSDB_WUNLOCK_TABLE(pTable);
+      return -1;
+    }
 
-  pTable->lastCols = (SDataCol*)malloc(numOfColumn * sizeof(SDataCol));
-  if (pTable->lastCols == NULL) {
-    return -1;
-  }
+    for (int16_t i = 0; i < numOfColumn; ++i) {
+      STColumn *pCol = schemaColAt(pSchema, i);
+      SDataCol *pDataCol = &(pTable->lastCols[i]);
+      pDataCol->bytes = 0;
+      pDataCol->pData = NULL;
+      pDataCol->colId = pCol->colId;
+    }
 
-  for (int16_t i = 0; i < numOfColumn; ++i) {
-    STColumn *pCol = schemaColAt(pSchema, i);
-    SDataCol* pDataCol = &(pTable->lastCols[i]);
-    pDataCol->bytes = 0;
-    pDataCol->pData = NULL;
-    pDataCol->colId = pCol->colId;
+    pTable->lastColSVersion = schemaVersion(pSchema);
+    pTable->maxColNum = numOfColumn;
+    pTable->restoreColumnNum = 0;
+    pTable->hasRestoreLastColumn = false;
   }
-
-  pTable->lastColSVersion = schemaVersion(pSchema);
-  pTable->maxColNum = numOfColumn;
-  pTable->restoreColumnNum = 0;
-  pTable->hasRestoreLastColumn = false;
+  TSDB_WUNLOCK_TABLE(pTable);
   return 0;
 }
 
@@ -797,6 +800,7 @@ static STable *tsdbNewTable() {
 
   pTable->lastCols = NULL;
   pTable->restoreColumnNum = 0;
+  pTable->cacheLastConfigVersion = 0;
   pTable->maxColNum = 0;
   pTable->hasRestoreLastColumn = false;
   pTable->lastColSVersion = -1;

src/tsdb/src/tsdbRead.c

@@ -156,6 +156,7 @@ typedef struct STableGroupSupporter {
 static STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList);
 static int32_t checkForCachedLastRow(STsdbQueryHandle* pQueryHandle, STableGroupInfo *groupList);
 static int32_t checkForCachedLast(STsdbQueryHandle* pQueryHandle);
+static int32_t lazyLoadCacheLast(STsdbQueryHandle* pQueryHandle);
 static int32_t tsdbGetCachedLastRow(STable* pTable, SMemRow* pRes, TSKEY* lastKey);
 
 static void    changeQueryHandleForInterpQuery(TsdbQueryHandleT pHandle);
@@ -589,6 +590,28 @@ void tsdbResetQueryHandleForNewTable(TsdbQueryHandleT queryHandle, STsdbQueryCon
   pQueryHandle->next = doFreeColumnInfoData(pQueryHandle->next);
 }
 
+static int32_t lazyLoadCacheLast(STsdbQueryHandle* pQueryHandle) {
+  STsdbRepo* pRepo = pQueryHandle->pTsdb;
+
+  size_t  numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
+  int32_t code = 0;
+  for (size_t i = 0; i < numOfTables; ++i) {
+    STableCheckInfo* pCheckInfo = taosArrayGet(pQueryHandle->pTableCheckInfo, i);
+    STable*          pTable = pCheckInfo->pTableObj;
+    if (pTable->cacheLastConfigVersion == pRepo->cacheLastConfigVersion) {
+      continue;
+    }
+    code = tsdbLoadLastCache(pRepo, pTable);
+    if (code != 0) {
+      tsdbError("%p uid:%" PRId64 ", tid:%d, failed to load last cache since %s", pQueryHandle, pTable->tableId.uid,
+                pTable->tableId.tid, tstrerror(terrno));
+      break;
+    }
+  }
+
+  return code;
+}
+
 TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, SMemRef* pMemRef) {
   pCond->twindow = updateLastrowForEachGroup(groupList);
 
@@ -602,6 +625,8 @@ TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STable
     return NULL;
   }
 
+  lazyLoadCacheLast(pQueryHandle);
+
   int32_t code = checkForCachedLastRow(pQueryHandle, groupList);
   if (code != TSDB_CODE_SUCCESS) { // set the numOfTables to be 0
     terrno = code;
@@ -616,13 +641,14 @@ TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STable
   return pQueryHandle;
 }
 
-
 TsdbQueryHandleT tsdbQueryCacheLast(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, SMemRef* pMemRef) {
   STsdbQueryHandle *pQueryHandle = (STsdbQueryHandle*) tsdbQueryTables(tsdb, pCond, groupList, qId, pMemRef);
   if (pQueryHandle == NULL) {
     return NULL;
   }
 
+  lazyLoadCacheLast(pQueryHandle);
+
   int32_t code = checkForCachedLast(pQueryHandle);
   if (code != TSDB_CODE_SUCCESS) { // set the numOfTables to be 0
     terrno = code;
@@ -2781,6 +2807,9 @@ static bool loadCachedLast(STsdbQueryHandle* pQueryHandle) {
     }
     
     int32_t i = 0, j = 0;
+
+    // lock pTable->lastCols[i] as it would be released when schema update(tsdbUpdateLastColSchema)
+    TSDB_RLOCK_TABLE(pTable);
     while(i < tgNumOfCols && j < numOfCols) {
       pColInfo = taosArrayGet(pQueryHandle->pColumns, i);
       if (pTable->lastCols[j].colId < pColInfo->info.colId) {
@@ -2867,6 +2896,7 @@ static bool loadCachedLast(STsdbQueryHandle* pQueryHandle) {
       i++;
       j++;
     }
+    TSDB_RUNLOCK_TABLE(pTable);
 
     // leave the real ts column as the last row, because last function only (not stable) use the last row as res
     if (priKey != TSKEY_INITIAL_VAL) {
@@ -3198,7 +3228,9 @@ int32_t checkForCachedLast(STsdbQueryHandle* pQueryHandle) {
 
   int32_t code = 0;
   
-  if (pQueryHandle->pTsdb && atomic_load_8(&pQueryHandle->pTsdb->hasCachedLastColumn)){
+  STsdbRepo* pRepo = pQueryHandle->pTsdb;
+
+  if (pRepo && CACHE_LAST_NULL_COLUMN(&(pRepo->config))) {
     pQueryHandle->cachelastrow = TSDB_CACHED_TYPE_LAST;
   }
 

src/util/src/tcompression.c

@@ -346,7 +346,7 @@ int tsCompressBoolImp(const char *const input, const int nelements, char *const
       /* t = (~((( uint8_t)1) << (7-i%BITS_PER_BYTE))); */
       output[pos] |= t;
     } else {
-      uError("Invalid compress bool value:%d", output[pos]);
+      uError("Invalid compress bool value:%d", input[i]);
       return -1;
     }
   }

tests/pytest/compress/compressChangeVersion.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import os
+import sys
+sys.path.insert(0, os.getcwd())
+from util.log import *
+from util.sql import *
+from util.dnodes import *
+import taos
+import threading
+import subprocess
+from random import choice
+
+ 
+class TwoClients:
+    def initConnection(self):
+        self.host = "chenhaoran01"
+        self.user = "root"
+        self.password = "taosdata"
+        self.config = "/home/chr/cfg/single/"     
+        self.port =6030 
+        self.rowNum = 10
+        self.ts = 1537146000000  
+
+    def getBuildPath(self):
+        selfPath = os.path.dirname(os.path.realpath(__file__))
+
+        if ("community" in selfPath):
+            projPath = selfPath[:selfPath.find("community")]
+        else:
+            projPath = selfPath[:selfPath.find("tests")]
+
+        for root, dirs, files in os.walk(projPath):
+            if ("taosd" in files):
+                rootRealPath = os.path.dirname(os.path.realpath(root))
+                if ("packaging" not in rootRealPath):
+                    buildPath = root[:len(root)-len("/build/bin")]
+                    break
+        return buildPath
+
+    def run(self):
+        buildPath = self.getBuildPath()
+        if (buildPath == ""):
+            tdLog.exit("taosd not found!")
+        else:
+            tdLog.info("taosd found in %s" % buildPath)
+        binPath = buildPath+ "/build/bin/"
+        walFilePath = "/var/lib/taos/mnode_bak/wal/"
+        
+        # new taos client
+        conn1 = taos.connect(host=self.host, user=self.user, password=self.password, config=self.config )
+        print(conn1)
+        cur1 = conn1.cursor()
+        tdSql.init(cur1, True)
+
+        # create backgroud db and tb
+        tdSql.execute("drop database if exists db1")
+        os.system("%staosdemo -f compress/insertDataDb1.json -y " % binPath)
+        # create foreground db and tb 
+        tdSql.execute("drop database if exists foredb")
+        tdSql.execute("create database foredb")
+        tdSql.execute("use foredb")
+        print("123test")
+        tdSql.execute("create stable if not exists stb (ts timestamp, dataInt int, dataDouble  double,dataStr nchar(200)) tags(loc nchar(50),t1 int)")
+        tdSql.execute("create table tb1 using stb tags('beijing1', 10)")
+        tdSql.execute("insert into tb1 values(1614218412000,8635,98.861,'qazwsxedcrfvtgbyhnujmikolp1')(1614218422000,8636,98.862,'qazwsxedcrfvtgbyhnujmikolp2')")
+        tdSql.execute("create table tb2 using stb tags('beijing2', 11)")
+        tdSql.execute("insert into tb2 values(1614218432000,8647,98.863,'qazwsxedcrfvtgbyhnujmikolp3')")
+        tdSql.execute("insert into tb2 values(1614218442000,8648,98.864,'qazwsxedcrfvtgbyhnujmikolp4')")
+
+
+        # check data correct 
+        tdSql.execute("use db1")
+        tdSql.query("select count(tbname) from stb0")
+        tdSql.checkData(0, 0, 50000)
+        tdSql.query("select count(*) from stb0")
+        tdSql.checkData(0, 0, 5000000)        
+        tdSql.execute("use foredb")
+        tdSql.query("select count (tbname) from stb")
+        tdSql.checkData(0, 0, 2)
+        tdSql.query("select count (*) from stb")
+        tdSql.checkData(0, 0, 4)
+        tdSql.query("select * from tb1 order by ts")
+        tdSql.checkData(0, 3, "qazwsxedcrfvtgbyhnujmikolp1")
+        tdSql.query("select * from tb2 order by ts")
+        tdSql.checkData(1, 3, "qazwsxedcrfvtgbyhnujmikolp4")
+
+
+
+        # delete useless file       
+        testcaseFilename = os.path.split(__file__)[-1]
+        os.system("rm -rf ./insert_res.txt")
+        # os.system("rm -rf compress/%s.sql" % testcaseFilename )       
+
+clients = TwoClients()
+clients.initConnection()
+# clients.getBuildPath()
+clients.run()
\ No newline at end of file

tests/pytest/compress/insertDataDb1.json

+{
+    "filetype": "insert",
+    "cfgdir": "/etc/taos",
+    "host": "127.0.0.1",
+    "port": 6030,
+    "user": "root",
+    "password": "taosdata",
+    "thread_count": 4,
+    "thread_count_create_tbl": 4,
+    "result_file": "./insert_res.txt",
+    "confirm_parameter_prompt": "no",
+    "insert_interval": 0,
+    "interlace_rows": 10,
+    "num_of_records_per_req": 1000,
+    "max_sql_len": 1024000,
+    "databases": [{
+        "dbinfo": {
+            "name": "db1",
+            "drop": "yes",
+            "replica": 1,
+            "days": 10,
+            "cache": 50,
+            "blocks": 8,
+            "precision": "ms",
+            "keep": 3650,
+            "minRows": 100,
+            "maxRows": 4096,
+            "comp":2,
+            "walLevel":1,
+            "cachelast":0,
+            "quorum":1,
+            "fsync":3000,
+            "update": 0
+        },
+        "super_tables": [{
+            "name": "stb0",
+            "child_table_exists":"no",
+            "childtable_count": 50000,
+            "childtable_prefix": "stb00_",
+            "auto_create_table": "no",
+            "batch_create_tbl_num": 100,
+            "data_source": "rand",
+            "insert_mode": "taosc",
+            "insert_rows": 100,
+            "childtable_limit": 0,
+            "childtable_offset":0,
+            "interlace_rows": 0,
+            "insert_interval":0,
+            "max_sql_len": 1024000,
+            "disorder_ratio": 0,
+            "disorder_range": 1000,
+            "timestamp_step": 1,
+            "start_timestamp": "2020-10-01 00:00:00.000",
+            "sample_format": "csv",
+            "sample_file": "./sample.csv",
+            "tags_file": "",
+            "columns": [{"type": "INT"},{"type": "TINYINT"},{"type": "smallint"},{"type": "bool"},{"type": "bigint"},{"type": "float"},{"type": "double"}, {"type": "BINARY","len": 32}, {"type": "nchar","len": 32}],
+            "tags": [{"type": "TINYINT", "count":2}, {"type": "BINARY", "len": 16, "count":1}]
+        }]
+    }]
+}
+

tests/pytest/tools/taosdemoTestTblAlt.py

@@ -54,27 +54,36 @@ class TDTestCase:
         binPath = buildPath + "/build/bin/"
 
         if(threadID == 0):
-            os.system("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT -m t" %
+            print("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT" %
+                      (binPath, self.numberOfTables, self.numberOfRecords))
+            os.system("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT" %
                       (binPath, self.numberOfTables, self.numberOfRecords))
         if(threadID == 1):
             time.sleep(2)
             print("use test")
-            while True:
+            max_try = 100
+            count = 0
+            while (count < max_try):
                 try:
                     tdSql.execute("use test")
                     break
                 except Exception as e:
                     tdLog.info("use database test failed")
-                    time.sleep(1)
+                    time.sleep(2)
+                    count += 1
+                    print("try %d times" % count)
                     continue
 
             # check if all the tables have heen created
-            while True:
+            count = 0
+            while (count < max_try):
                 try:
                     tdSql.query("show tables")
                 except Exception as e:
                     tdLog.info("show tables test failed")
-                    time.sleep(1)
+                    time.sleep(2)
+                    count += 1
+                    print("try %d times" % count)
                     continue
 
                 rows = tdSql.queryRows
@@ -83,13 +92,17 @@ class TDTestCase:
                     break
                 time.sleep(1)
             # check if there are any records in the last created table
-            while True:
+            count = 0
+            while (count < max_try):
                 print("query started")
+                print("try %d times" % count)
                 try:
-                    tdSql.query("select * from test.t7")
+                    tdSql.query("select * from test.d7")
                 except Exception as e:
                     tdLog.info("select * test failed")
                     time.sleep(2)
+                    count += 1
+                    print("try %d times" % count)
                     continue
 
                 rows = tdSql.queryRows
@@ -100,8 +113,8 @@ class TDTestCase:
 
             print("alter table test.meters add column c10 int")
             tdSql.execute("alter table test.meters add column c10 int")
-            print("insert into test.t7 values (now, 1, 2, 3, 4, 0)")
-            tdSql.execute("insert into test.t7 values (now, 1, 2, 3, 4, 0)")
+            print("insert into test.d7 values (now, 1, 2, 3, 4, 0)")
+            tdSql.execute("insert into test.d7 values (now, 1, 2, 3, 4, 0)")
 
     def run(self):
         tdSql.prepare()