Merge branch 'main' into test/TS-3241

cmake/taostools_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 0681d8b
+        GIT_TAG ffc2e6f
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

include/common/tcommon.h

@@ -341,6 +341,8 @@ typedef struct {
     float       f;
   };
   size_t length;
+  bool keyEscaped;
+  bool valueEscaped;
 } SSmlKv;
 
 #define QUERY_ASC_FORWARD_STEP  1

include/common/ttokendef.h

@@ -346,6 +346,7 @@
 #define TK_VIEW                           328
 #define TK_WAL                            329
 
+
 #define TK_NK_SPACE   600
 #define TK_NK_COMMENT 601
 #define TK_NK_ILLEGAL 602

include/libs/stream/tstream.h

@@ -273,6 +273,7 @@ typedef struct SStreamId {
 typedef struct SCheckpointInfo {
   int64_t id;
   int64_t version;   // offset in WAL
+  int64_t currentVer;// current offset in WAL, not serialize it
 } SCheckpointInfo;
 
 typedef struct SStreamStatus {
@@ -537,6 +538,7 @@ void    streamTaskInputFail(SStreamTask* pTask);
 int32_t streamTryExec(SStreamTask* pTask);
 int32_t streamSchedExec(SStreamTask* pTask);
 int32_t streamTaskOutput(SStreamTask* pTask, SStreamDataBlock* pBlock);
+bool    streamTaskShouldStop(const SStreamStatus* pStatus);
 
 int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz);
 

include/util/tdef.h

@@ -368,11 +368,11 @@ typedef enum ELogicConditionType {
 #define TSDB_MIN_STT_TRIGGER            1
 #define TSDB_MAX_STT_TRIGGER            16
 #define TSDB_DEFAULT_SST_TRIGGER        1
-#define TSDB_MIN_HASH_PREFIX            0
-#define TSDB_MAX_HASH_PREFIX            128
+#define TSDB_MIN_HASH_PREFIX            (2 - TSDB_TABLE_NAME_LEN)
+#define TSDB_MAX_HASH_PREFIX            (TSDB_TABLE_NAME_LEN - 2)
 #define TSDB_DEFAULT_HASH_PREFIX        0
-#define TSDB_MIN_HASH_SUFFIX            0
-#define TSDB_MAX_HASH_SUFFIX            128
+#define TSDB_MIN_HASH_SUFFIX            (2 - TSDB_TABLE_NAME_LEN)
+#define TSDB_MAX_HASH_SUFFIX            (TSDB_TABLE_NAME_LEN - 2)
 #define TSDB_DEFAULT_HASH_SUFFIX        0
 
 #define TSDB_DB_MIN_WAL_RETENTION_PERIOD -1

include/util/tutil.h

@@ -81,14 +81,22 @@ static FORCE_INLINE void taosEncryptPass_c(uint8_t *inBuf, size_t len, char *tar
 
 static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen, int32_t method, int32_t prefix,
                                              int32_t suffix) {
-  if (prefix == 0 && suffix == 0) {
+  if ((prefix == 0 && suffix == 0) || (tblen <= (prefix + suffix)) || (tblen <= -1 * (prefix + suffix)) || prefix * suffix < 0) {
     return MurmurHash3_32(tbname, tblen);
-  } else {
-    if (tblen <= (prefix + suffix)) {
-      return MurmurHash3_32(tbname, tblen);
-    } else {
+  } else if (prefix > 0 || suffix > 0) {
     return MurmurHash3_32(tbname + prefix, tblen - prefix - suffix);
+  } else {
+    char tbName[TSDB_TABLE_FNAME_LEN];
+    int32_t offset = 0;
+    if (prefix < 0) {
+      offset = -1 * prefix;
+      strncpy(tbName, tbname, offset);
+    }
+    if (suffix < 0) {
+      strncpy(tbName + offset, tbname + tblen + suffix, -1 * suffix);
+      offset += -1 *suffix;
     }
+    return MurmurHash3_32(tbName, offset);
   }
 }
 

packaging/tools/makeclient.sh

@@ -197,7 +197,8 @@ if [[ $productName == "TDengine" ]]; then
       mkdir -p ${install_dir}/connector
       if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
           if [ "$osType" != "Darwin" ]; then
-              [ -f ${build_dir}/lib/*.jar ] && cp ${build_dir}/lib/*.jar ${install_dir}/connector || :
+              jars=$(ls ${build_dir}/lib/*.jar 2>/dev/null|wc -l)
+              [ "${jars}" != "0" ] && cp ${build_dir}/lib/*.jar ${install_dir}/connector || :
           fi
           git clone --depth 1 https://github.com/taosdata/driver-go ${install_dir}/connector/go
           rm -rf ${install_dir}/connector/go/.git ||:

packaging/tools/makepkg.sh

@@ -338,7 +338,20 @@ if [ "$verMode" == "cluster" ] || [ "$verMode" == "cloud" ]; then
     connector_dir="${code_dir}/connector"
     mkdir -p ${install_dir}/connector
     if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
-        [ -f ${build_dir}/lib/*.jar ] && cp ${build_dir}/lib/*.jar ${install_dir}/connector || :
+        tmp_pwd=`pwd`
+    	  cd ${install_dir}/connector
+    	  if [ ! -d taos-connector-jdbc ];then
+          	git clone -b main --depth=1 https://github.com/taosdata/taos-connector-jdbc.git ||:
+    	  fi
+    	  cd taos-connector-jdbc
+    	  mvn clean package -Dmaven.test.skip=true
+    	  echo  ${build_dir}/lib/
+    	  cp target/*.jar  ${build_dir}/lib/
+    	  cd ${install_dir}/connector
+    	  rm -rf taos-connector-jdbc
+    	  cd ${tmp_pwd}
+   	    jars=$(ls ${build_dir}/lib/*.jar 2>/dev/null|wc -l)
+        [ "${jars}" != "0" ] && cp ${build_dir}/lib/*.jar ${install_dir}/connector || :
         git clone --depth 1 https://github.com/taosdata/driver-go ${install_dir}/connector/go
         rm -rf ${install_dir}/connector/go/.git ||:
 

source/client/inc/clientSml.h

@@ -70,7 +70,7 @@ extern "C" {
 #define VALUE_LEN 6
 
 #define OTD_JSON_FIELDS_NUM     4
-#define MAX_RETRY_TIMES 100
+#define MAX_RETRY_TIMES 10
 typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;
 
 typedef enum {
@@ -107,6 +107,7 @@ typedef struct {
   int32_t colsLen;
   int32_t timestampLen;
 
+  bool    measureEscaped;
   SArray *colArray;
 } SSmlLineInfo;
 
@@ -206,6 +207,19 @@ typedef struct {
 
 #define IS_SAME_KEY (maxKV->keyLen == kv.keyLen && memcmp(maxKV->key, kv.key, kv.keyLen) == 0)
 
+#define IS_SLASH_LETTER_IN_MEASUREMENT(sql)                                                           \
+  (*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE))
+
+#define MOVE_FORWARD_ONE(sql, len) (memmove((void *)((sql)-1), (sql), len))
+
+#define PROCESS_SLASH_IN_MEASUREMENT(key, keyLen)           \
+  for (int i = 1; i < keyLen; ++i) {         \
+    if (IS_SLASH_LETTER_IN_MEASUREMENT(key + i)) {          \
+      MOVE_FORWARD_ONE(key + i, keyLen - i); \
+      keyLen--;                              \
+    }                                        \
+  }
+
 extern int64_t smlFactorNS[3];
 extern int64_t smlFactorS[3];
 
@@ -237,6 +251,7 @@ uint8_t           smlGetTimestampLen(int64_t num);
 void              clearColValArray(SArray* pCols);
 void              smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag);
 
+void freeSSmlKv(void* data);
 int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
 int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
 int32_t smlParseJSON(SSmlHandle *info, char *payload);

source/client/src/clientSml.c

@@ -535,7 +535,7 @@ static int32_t smlGenerateSchemaAction(SSchema *colField, SHashObj *colHash, SSm
   if (index) {
     if (colField[*index].type != kv->type) {
       uError("SML:0x%" PRIx64 " point type and db type mismatch. point type: %d, db type: %d, key: %s", info->id, colField[*index].type, kv->type, kv->key);
-      return TSDB_CODE_TSC_INVALID_VALUE;
+      return TSDB_CODE_SML_INVALID_DATA;
     }
 
     if ((colField[*index].type == TSDB_DATA_TYPE_VARCHAR &&
@@ -765,8 +765,12 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) {
 
     size_t superTableLen = 0;
     void  *superTable = taosHashGetKey(tmp, &superTableLen);
+    char* measure = taosMemoryMalloc(superTableLen);
+    memcpy(measure, superTable, superTableLen);
+    PROCESS_SLASH_IN_MEASUREMENT(measure, superTableLen);
     memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
-    memcpy(pName.tname, superTable, superTableLen);
+    memcpy(pName.tname, measure, superTableLen);
+    taosMemoryFree(measure);
 
     code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
 
@@ -1049,7 +1053,7 @@ void smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag) {
   //  }
   //  taosMemoryFree(tag->key);
   taosArrayDestroy(tag->cols);
-  taosArrayDestroy(tag->tags);
+  taosArrayDestroyEx(tag->tags, freeSSmlKv);
   taosMemoryFree(tag);
 }
 
@@ -1063,6 +1067,12 @@ void clearColValArray(SArray *pCols) {
   }
 }
 
+void freeSSmlKv(void* data){
+  SSmlKv *kv = (SSmlKv*)data;
+  if(kv->keyEscaped) taosMemoryFree((void*)(kv->key));
+  if(kv->valueEscaped) taosMemoryFree((void*)(kv->value));
+}
+
 void smlDestroyInfo(SSmlHandle *info) {
   if (!info) return;
   qDestroyQuery(info->pQuery);
@@ -1098,11 +1108,11 @@ void smlDestroyInfo(SSmlHandle *info) {
   }
   taosArrayDestroy(info->valueJsonArray);
 
-  taosArrayDestroy(info->preLineTagKV);
+  taosArrayDestroyEx(info->preLineTagKV, freeSSmlKv);
 
   if (!info->dataFormat) {
     for (int i = 0; i < info->lineNum; i++) {
-      taosArrayDestroy(info->lines[i].colArray);
+      taosArrayDestroyEx(info->lines[i].colArray, freeSSmlKv);
       if (info->parseJsonByLib) {
         taosMemoryFree(info->lines[i].tags);
       }
@@ -1165,8 +1175,10 @@ static int32_t smlPushCols(SArray *colsArray, SArray *cols) {
   }
   for (size_t i = 0; i < taosArrayGetSize(cols); i++) {
     SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
+    terrno = 0;
     taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
     if (terrno == TSDB_CODE_DUP_KEY) {
+      taosHashCleanup(kvHash);
       return terrno;
     }
   }
@@ -1240,12 +1252,13 @@ static int32_t smlParseLineBottom(SSmlHandle *info) {
       uDebug("SML:0x%" PRIx64 " smlParseLineBottom add meta, format:%d, linenum:%d", info->id, info->dataFormat,
              info->lineNum);
       SSmlSTableMeta *meta = smlBuildSTableMeta(info->dataFormat);
+      taosHashPut(info->superTables, elements->measure, elements->measureLen, &meta, POINTER_BYTES);
+      terrno = 0;
       smlInsertMeta(meta->tagHash, meta->tags, tinfo->tags);
       if (terrno == TSDB_CODE_DUP_KEY) {
         return terrno;
       }
       smlInsertMeta(meta->colHash, meta->cols, elements->colArray);
-      taosHashPut(info->superTables, elements->measure, elements->measureLen, &meta, POINTER_BYTES);
     }
   }
   uDebug("SML:0x%" PRIx64 " smlParseLineBottom end, format:%d, linenum:%d", info->id, info->dataFormat, info->lineNum);
@@ -1304,9 +1317,15 @@ static int32_t smlInsertData(SSmlHandle *info) {
     uDebug("SML:0x%" PRIx64 " smlInsertData table:%s, uid:%" PRIu64 ", format:%d", info->id, pName.tname,
            tableData->uid, info->dataFormat);
 
+    int   measureLen = tableData->sTableNameLen;
+    char* measure = (char*)taosMemoryMalloc(tableData->sTableNameLen);
+    memcpy(measure, tableData->sTableName, tableData->sTableNameLen);
+    PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen);
+
     code = smlBindData(info->pQuery, info->dataFormat, tableData->tags, (*pMeta)->cols, tableData->cols,
-                       (*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen,
+                       (*pMeta)->tableMeta, tableData->childTableName, measure, measureLen,
                        info->ttl, info->msgBuf.buf, info->msgBuf.len);
+    taosMemoryFree(measure);
     if (code != TSDB_CODE_SUCCESS) {
       uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
       return code;
@@ -1420,14 +1439,14 @@ static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char
 
     char cTmp = 0;  // for print tmp if is raw
     if (info->isRawLine) {
-      cTmp = tmp[len - 1];
-      tmp[len - 1] = '\0';
+      cTmp = tmp[len];
+      tmp[len] = '\0';
     }
 
     uDebug("SML:0x%" PRIx64 " smlParseLine israw:%d, numLines:%d, protocol:%d, len:%d, sql:%s", info->id,
            info->isRawLine, numLines, info->protocol, len, tmp);
     if (info->isRawLine) {
-      tmp[len - 1] = cTmp;
+      tmp[len] = cTmp;
     }
 
     if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
@@ -1449,6 +1468,7 @@ static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char
       code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
     }
     if (code != TSDB_CODE_SUCCESS) {
+      tmp[len] = '\0';
       uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, tmp);
       return code;
     }
@@ -1494,8 +1514,8 @@ static int smlProcess(SSmlHandle *info, char *lines[], char *rawLine, char *rawL
 
   do {
     code = smlModifyDBSchemas(info);
-    if (code == 0) break;
-    taosMsleep(500);
+    if (code == 0 || code == TSDB_CODE_SML_INVALID_DATA) break;
+    taosMsleep(100);
     uInfo("SML:0x%" PRIx64 " smlModifyDBSchemas retry code:%s, times:%d", info->id, tstrerror(code), retryNum);
   } while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES);
 

source/client/src/clientSmlLine.c

@@ -21,32 +21,33 @@
 #include "clientSml.h"
 
 // comma ,
-// #define IS_SLASH_COMMA(sql) (*(sql) == COMMA && *((sql)-1) == SLASH)
 #define IS_COMMA(sql) (*(sql) == COMMA && *((sql)-1) != SLASH)
 // space
-// #define IS_SLASH_SPACE(sql) (*(sql) == SPACE && *((sql)-1) == SLASH)
 #define IS_SPACE(sql) (*(sql) == SPACE && *((sql)-1) != SLASH)
 // equal =
-// #define IS_SLASH_EQUAL(sql) (*(sql) == EQUAL && *((sql)-1) == SLASH)
 #define IS_EQUAL(sql) (*(sql) == EQUAL && *((sql)-1) != SLASH)
 // quote "
-// #define IS_SLASH_QUOTE(sql) (*(sql) == QUOTE && *((sql)-1) == SLASH)
-#define IS_QUOTE(sql) (*(sql) == QUOTE && *((sql)-1) != SLASH)
+//#define IS_QUOTE(sql) (*(sql) == QUOTE && *((sql)-1) != SLASH)
 // SLASH
-// #define IS_SLASH_SLASH(sql) (*(sql) == SLASH && *((sql)-1) == SLASH)
 
-#define IS_SLASH_LETTER(sql)                                                                           \
-  (*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == EQUAL || *(sql) == QUOTE || \
-                           *(sql) == SLASH))  //  (IS_SLASH_COMMA(sql) || IS_SLASH_SPACE(sql) || IS_SLASH_EQUAL(sql) ||
-                                              //  IS_SLASH_QUOTE(sql) || IS_SLASH_SLASH(sql))
+#define IS_SLASH_LETTER_IN_FIELD_VALUE(sql)                                                           \
+  (*((sql)-1) == SLASH && (*(sql) == QUOTE || *(sql) == SLASH))
 
-#define MOVE_FORWARD_ONE(sql, len) (memmove((void *)((sql)-1), (sql), len))
+#define IS_SLASH_LETTER_IN_TAG_FIELD_KEY(sql)                                                         \
+  (*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == EQUAL))
 
-#define PROCESS_SLASH(key, keyLen)           \
+#define PROCESS_SLASH_IN_FIELD_VALUE(key, keyLen)           \
   for (int i = 1; i < keyLen; ++i) {                        \
-    if (IS_SLASH_LETTER(key + i)) {          \
+    if (IS_SLASH_LETTER_IN_FIELD_VALUE(key + i)) {          \
+      MOVE_FORWARD_ONE(key + i, keyLen - i); \
+      keyLen--;                              \
+    }                                        \
+  }
+
+#define PROCESS_SLASH_IN_TAG_FIELD_KEY(key, keyLen)           \
+  for (int i = 1; i < keyLen; ++i) {         \
+    if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(key + i)) {          \
       MOVE_FORWARD_ONE(key + i, keyLen - i); \
-      i--;                                   \
       keyLen--;                              \
     }                                        \
   }
@@ -151,7 +152,17 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
 
       SSmlSTableMeta *sMeta = NULL;
       if (unlikely(tmp == NULL)) {
-        STableMeta *pTableMeta = smlGetMeta(info, currElement->measure, currElement->measureLen);
+        char* measure = currElement->measure;
+        int   measureLen = currElement->measureLen;
+        if(currElement->measureEscaped){
+          measure = (char*)taosMemoryMalloc(currElement->measureLen);
+          memcpy(measure, currElement->measure, currElement->measureLen);
+          PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen);
+        }
+        STableMeta *pTableMeta = smlGetMeta(info, measure, measureLen);
+        if(currElement->measureEscaped){
+          taosMemoryFree(measure);
+        }
         if (pTableMeta == NULL) {
           info->dataFormat = false;
           info->reRun = true;
@@ -171,17 +182,18 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
       info->maxTagKVs = (*tmp)->tags;
     }
   }
-  taosArrayClear(preLineKV);
+  taosArrayClearEx(preLineKV, freeSSmlKv);
 
   while (*sql < sqlEnd) {
     if (unlikely(IS_SPACE(*sql))) {
       break;
     }
 
-    bool hasSlash = false;
     // parse key
     const char *key = *sql;
     size_t      keyLen = 0;
+    bool        keyEscaped = false;
+    size_t      keyLenEscaped = 0;
     while (*sql < sqlEnd) {
       if (unlikely(IS_COMMA(*sql))) {
         smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
@@ -192,16 +204,14 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
         (*sql)++;
         break;
       }
-      if (!hasSlash) {
-        hasSlash = (*(*sql) == SLASH);
+      if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
+        keyLenEscaped++;
+        keyEscaped = true;
       }
       (*sql)++;
     }
-    if (unlikely(hasSlash)) {
-      PROCESS_SLASH(key, keyLen)
-    }
 
-    if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
+    if (unlikely(IS_INVALID_COL_LEN(keyLen - keyLenEscaped))) {
       smlBuildInvalidDataMsg(&info->msgBuf, "invalid key or key is too long than 64", key);
       return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
     }
@@ -209,7 +219,8 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
     // parse value
     const char *value = *sql;
     size_t      valueLen = 0;
-    hasSlash = false;
+    bool        valueEscaped = false;
+    size_t      valueLenEscaped = 0;
     while (*sql < sqlEnd) {
       // parse value
       if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
@@ -219,8 +230,9 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
         return TSDB_CODE_SML_INVALID_DATA;
       }
 
-      if (!hasSlash) {
-        hasSlash = (*(*sql) == SLASH);
+      if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
+        valueLenEscaped++;
+        valueEscaped = true;
       }
 
       (*sql)++;
@@ -232,15 +244,24 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
       return TSDB_CODE_SML_INVALID_DATA;
     }
 
-    if (unlikely(hasSlash)) {
-      PROCESS_SLASH(value, valueLen)
-    }
-
-    if (unlikely(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
+    if (unlikely(valueLen - valueLenEscaped > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
       return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
     }
 
-    SSmlKv kv = {.key = key, .keyLen = keyLen, .type = TSDB_DATA_TYPE_NCHAR, .value = value, .length = valueLen};
+    if (keyEscaped){
+      char *tmp = (char*)taosMemoryMalloc(keyLen);
+      memcpy(tmp, key, keyLen);
+      PROCESS_SLASH_IN_TAG_FIELD_KEY(tmp, keyLen);
+      key = tmp;
+    }
+    if (valueEscaped){
+      char *tmp = (char*)taosMemoryMalloc(valueLen);
+      memcpy(tmp, value, valueLen);
+      PROCESS_SLASH_IN_TAG_FIELD_KEY(tmp, valueLen);
+      value = tmp;
+    }
+    SSmlKv kv = {.key = key, .keyLen = keyLen, .type = TSDB_DATA_TYPE_NCHAR, .value = value, .length = valueLen, .keyEscaped = keyEscaped, .valueEscaped = valueEscaped};
+    taosArrayPush(preLineKV, &kv);
     if (info->dataFormat) {
       if (unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)) {
         info->dataFormat = false;
@@ -266,7 +287,6 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
         info->needModifySchema = true;
       }
     }
-    taosArrayPush(preLineKV, &kv);
 
     cnt++;
     if (IS_SPACE(*sql)) {
@@ -285,6 +305,11 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
     return TSDB_CODE_OUT_OF_MEMORY;
   }
   tinfo->tags = taosArrayDup(preLineKV, NULL);
+  for(size_t i = 0; i < taosArrayGetSize(preLineKV); i++){
+    SSmlKv *kv = (SSmlKv *)taosArrayGet(preLineKV, i);
+    if(kv->keyEscaped)kv->key = NULL;
+    if(kv->valueEscaped)kv->value = NULL;
+  }
 
   smlSetCTableName(tinfo);
   tinfo->uid = info->uid++;
@@ -321,7 +346,17 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
       SSmlSTableMeta **tmp =
           (SSmlSTableMeta **)taosHashGet(info->superTables, currElement->measure, currElement->measureLen);
       if (unlikely(tmp == NULL)) {
-        STableMeta *pTableMeta = smlGetMeta(info, currElement->measure, currElement->measureLen);
+        char* measure = currElement->measure;
+        int   measureLen = currElement->measureLen;
+        if(currElement->measureEscaped){
+          measure = (char*)taosMemoryMalloc(currElement->measureLen);
+          memcpy(measure, currElement->measure, currElement->measureLen);
+          PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen);
+        }
+        STableMeta *pTableMeta = smlGetMeta(info, measure, measureLen);
+        if(currElement->measureEscaped){
+          taosMemoryFree(measure);
+        }
         if (pTableMeta == NULL) {
           info->dataFormat = false;
           info->reRun = true;
@@ -352,10 +387,11 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
       break;
     }
 
-    bool hasSlash = false;
     // parse key
     const char *key = *sql;
     size_t      keyLen = 0;
+    bool        keyEscaped = false;
+    size_t      keyLenEscaped = 0;
     while (*sql < sqlEnd) {
       if (unlikely(IS_COMMA(*sql))) {
         smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
@@ -366,16 +402,14 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
         (*sql)++;
         break;
       }
-      if (!hasSlash) {
-        hasSlash = (*(*sql) == SLASH);
+      if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
+        keyLenEscaped++;
+        keyEscaped = true;
       }
       (*sql)++;
     }
-    if (unlikely(hasSlash)) {
-      PROCESS_SLASH(key, keyLen)
-    }
 
-    if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
+    if (unlikely(IS_INVALID_COL_LEN(keyLen - keyLenEscaped))) {
       smlBuildInvalidDataMsg(&info->msgBuf, "invalid key or key is too long than 64", key);
       return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
     }
@@ -383,11 +417,13 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
     // parse value
     const char *value = *sql;
     size_t      valueLen = 0;
-    hasSlash = false;
+    bool        valueEscaped = false;
+    size_t      valueLenEscaped = 0;
     bool        isInQuote = false;
+    const char *escapeChar = NULL;
     while (*sql < sqlEnd) {
       // parse value
-      if (unlikely(IS_QUOTE(*sql))) {
+      if (unlikely(*(*sql) == QUOTE && (*(*sql - 1) != SLASH || (*sql - 1) == escapeChar))) {
         isInQuote = !isInQuote;
         (*sql)++;
         continue;
@@ -395,13 +431,12 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
       if (!isInQuote) {
         if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
           break;
-        } else if (unlikely(IS_EQUAL(*sql))) {
-          smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
-          return TSDB_CODE_SML_INVALID_DATA;
         }
       }
-      if (!hasSlash) {
-        hasSlash = (*(*sql) == SLASH);
+      if (IS_SLASH_LETTER_IN_FIELD_VALUE(*sql) && (*sql - 1) != escapeChar) {
+        escapeChar = *sql;
+        valueEscaped = true;
+        valueLenEscaped++;
       }
 
       (*sql)++;
@@ -416,9 +451,6 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
       smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", value);
       return TSDB_CODE_SML_INVALID_DATA;
     }
-    if (unlikely(hasSlash)) {
-      PROCESS_SLASH(value, valueLen)
-    }
 
     SSmlKv  kv = {.key = key, .keyLen = keyLen, .value = value, .length = valueLen};
     int32_t ret = smlParseValue(&kv, &info->msgBuf);
@@ -427,11 +459,28 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
       return ret;
     }
 
+    if (keyEscaped){
+      char *tmp = (char*)taosMemoryMalloc(kv.keyLen);
+      memcpy(tmp, key, kv.keyLen);
+      PROCESS_SLASH_IN_TAG_FIELD_KEY(tmp, kv.keyLen);
+      kv.key = tmp;
+      kv.keyEscaped = keyEscaped;
+    }
+
+    if (valueEscaped){
+      char *tmp = (char*)taosMemoryMalloc(kv.length);
+      memcpy(tmp, kv.value, kv.length);
+      PROCESS_SLASH_IN_FIELD_VALUE(tmp, kv.length);
+      kv.value = tmp;
+      kv.valueEscaped = valueEscaped;
+    }
+
     if (info->dataFormat) {
       // cnt begin 0, add ts so + 2
       if (unlikely(cnt + 2 > info->currSTableMeta->tableInfo.numOfColumns)) {
         info->dataFormat = false;
         info->reRun = true;
+        freeSSmlKv(&kv);
         return TSDB_CODE_SUCCESS;
       }
       // bind data
@@ -440,22 +489,26 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
         uDebug("smlBuildCol error, retry");
         info->dataFormat = false;
         info->reRun = true;
+        freeSSmlKv(&kv);
         return TSDB_CODE_SUCCESS;
       }
       if (cnt >= taosArrayGetSize(info->masColKVs)) {
         info->dataFormat = false;
         info->reRun = true;
+        freeSSmlKv(&kv);
         return TSDB_CODE_SUCCESS;
       }
       SSmlKv *maxKV = (SSmlKv *)taosArrayGet(info->masColKVs, cnt);
       if (kv.type != maxKV->type) {
         info->dataFormat = false;
         info->reRun = true;
+        freeSSmlKv(&kv);
         return TSDB_CODE_SUCCESS;
       }
       if (unlikely(!IS_SAME_KEY)) {
         info->dataFormat = false;
         info->reRun = true;
+        freeSSmlKv(&kv);
         return TSDB_CODE_SUCCESS;
       }
 
@@ -463,6 +516,7 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
         maxKV->length = kv.length;
         info->needModifySchema = true;
       }
+      freeSSmlKv(&kv);
     } else {
       if (currElement->colArray == NULL) {
         currElement->colArray = taosArrayInit_s(sizeof(SSmlKv), 1);
@@ -487,10 +541,12 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
   elements->measure = sql;
 
   // parse measure
+  size_t measureLenEscaped = 0;
   while (sql < sqlEnd) {
-    if (unlikely((sql != elements->measure) && IS_SLASH_LETTER(sql))) {
-      MOVE_FORWARD_ONE(sql, sqlEnd - sql);
-      sqlEnd--;
+    if (unlikely((sql != elements->measure) && IS_SLASH_LETTER_IN_MEASUREMENT(sql))) {
+      elements->measureEscaped = true;
+      measureLenEscaped++;
+      sql++;
       continue;
     }
     if (unlikely(IS_COMMA(sql))) {
@@ -503,7 +559,7 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
     sql++;
   }
   elements->measureLen = sql - elements->measure;
-  if (unlikely(IS_INVALID_TABLE_LEN(elements->measureLen))) {
+  if (unlikely(IS_INVALID_TABLE_LEN(elements->measureLen - measureLenEscaped))) {
     smlBuildInvalidDataMsg(&info->msgBuf, "measure is empty or too large than 192", NULL);
     return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
   }
@@ -581,7 +637,9 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
                .keyLen = TS_LEN,
                .type = TSDB_DATA_TYPE_TIMESTAMP,
                .i = ts,
-               .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
+               .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes,
+               .keyEscaped = false,
+               .valueEscaped = false};
   if (info->dataFormat) {
     uDebug("SML:0x%" PRIx64 " smlParseInfluxString format true, ts:%" PRId64, info->id, ts);
     ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 0);

source/client/src/clientTmq.c

@@ -373,9 +373,6 @@ int32_t tmq_list_append(tmq_list_t* list, const char* src) {
   SArray* container = &list->container;
   if (src == NULL || src[0] == 0) return -1;
   char* topic = taosStrdup(src);
-  if (topic[0] != '`') {
-    strtolower(topic, src);
-  }
   if (taosArrayPush(container, &topic) == NULL) return -1;
   return 0;
 }
@@ -1243,9 +1240,6 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
   taosMemoryFree(pParam);
 
   if (code != 0) {
-    tscWarn("consumer:0x%" PRIx64 " msg from vgId:%d discarded, epoch %d, since %s, reqId:0x%" PRIx64, tmq->consumerId,
-            vgId, epoch, tstrerror(code), requestId);
-
     if (pMsg->pData) taosMemoryFree(pMsg->pData);
     if (pMsg->pEpSet) taosMemoryFree(pMsg->pEpSet);
 
@@ -1267,6 +1261,9 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
       taosWriteQitem(tmq->mqueue, pRspWrapper);
     } else if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {  // poll data while insert
       taosMsleep(500);
+    } else{
+      tscError("consumer:0x%" PRIx64 " msg from vgId:%d discarded, epoch %d, since %s, reqId:0x%" PRIx64, tmq->consumerId,
+               vgId, epoch, tstrerror(code), requestId);
     }
 
     goto CREATE_MSG_FAIL;

source/client/test/smlTest.cpp

@@ -50,8 +50,9 @@ TEST(testCase, smlParseInfluxString_Test) {
   int ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(elements.measure, sql);
-  ASSERT_EQ(elements.measureLen, strlen(",st"));
-  ASSERT_EQ(elements.measureTagsLen, strlen(",st,t1=3,t2=4,t3=t3"));
+  ASSERT_EQ(elements.measureLen, strlen("\\,st"));
+  ASSERT_EQ(elements.measureEscaped, true);
+  ASSERT_EQ(elements.measureTagsLen, strlen("\\,st,t1=3,t2=4,t3=t3"));
 
   ASSERT_EQ(elements.tags, sql + elements.measureLen + 1);
   ASSERT_EQ(elements.tagsLen, strlen("t1=3,t2=4,t3=t3"));
@@ -204,7 +205,26 @@ TEST(testCase, smlParseCols_Error_Test) {
                         "st,t=1 c=-3.402823466e+39u64 1626006833639000000",
                         "st,t=1 c=-339u64 1626006833639000000",
                         "st,t=1 c=18446744073709551616u64 1626006833639000000",
-                        "st,t=1 c=1=2 1626006833639000000"};
+                        "st,t=1 c=1=2 1626006833639000000,",
+                        // escape error test
+                        // measure comma,space
+                        "s,t,t=1 c=1 1626006833639000000,",
+                        "s t,t=1 c=1 1626006833639000000,",
+                        //tag key comma,equal,space
+                        "st,t,t=1 c=2 1626006833639000000,",
+                        "st,t=t=1 c=2 1626006833639000000,",
+                        "st,t t=1 c=2 1626006833639000000,",
+                        //tag value comma,equal,space
+                        "st,tt=a,a c=2 1626006833639000000,",
+                        "st,t=t=a a c=2 1626006833639000000,",
+                        "st,t t=a=a c=2 1626006833639000000,",
+                        //field key comma,equal,space
+                        "st,tt=aa c,1=2 1626006833639000000,",
+                        "st,tt=aa c=1=2 1626006833639000000,",
+                        "st,tt=aa c 1=2 1626006833639000000,",
+                        //field value    double quote,slash
+                        "st,tt=aa c=\"a\"a\" 1626006833639000000,",
+  };
 
   SSmlHandle *info = smlBuildSmlInfo(NULL);
   info->protocol = TSDB_SML_LINE_PROTOCOL;
@@ -256,16 +276,18 @@ TEST(testCase, smlParseCols_Test) {
   ASSERT_EQ(strncasecmp(kv->key, "cb=in", 5), 0);
   ASSERT_EQ(kv->keyLen, 5);
   ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BINARY);
-  ASSERT_EQ(kv->length, 17);
-  ASSERT_EQ(strncasecmp(kv->value, "pass,it ", 8), 0);
+  ASSERT_EQ(kv->length, 18);
+  ASSERT_EQ(kv->keyEscaped, true);
+  ASSERT_EQ(kv->valueEscaped, false);
+  ASSERT_EQ(strncasecmp(kv->value, "pass\\,it ", 9), 0);
 
   // nchar
   kv = (SSmlKv *)taosArrayGet(elements.colArray, 2);
   ASSERT_EQ(strncasecmp(kv->key, "cnch", 4), 0);
   ASSERT_EQ(kv->keyLen, 4);
   ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
-  ASSERT_EQ(kv->length, 8);
-  ASSERT_EQ(strncasecmp(kv->value, "ii=sd", 5), 0);
+  ASSERT_EQ(kv->length, 9);
+  ASSERT_EQ(strncasecmp(kv->value, "ii\\=sd", 5), 0);
 
   // bool
   kv = (SSmlKv *)taosArrayGet(elements.colArray, 3);

source/common/src/tglobal.c

@@ -198,7 +198,7 @@ int32_t tsTransPullupInterval = 2;
 int32_t tsMqRebalanceInterval = 2;
 int32_t tsStreamCheckpointTickInterval = 1;
 int32_t tsTtlUnit = 86400;
-int32_t tsTtlPushInterval = 86400;
+int32_t tsTtlPushInterval = 3600;
 int32_t tsGrantHBInterval = 60;
 int32_t tsUptimeInterval = 300;    // seconds
 char    tsUdfdResFuncs[512] = "";  // udfd resident funcs that teardown when udfd exits

source/dnode/mgmt/mgmt_vnode/src/vmInt.c

@@ -119,6 +119,7 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal)
         pVnode->pFetchQ->threadId);
   while (!taosQueueEmpty(pVnode->pFetchQ)) taosMsleep(10);
 
+  tqNotifyClose(pVnode->pImpl->pTq);
   dInfo("vgId:%d, wait for vnode stream queue:%p is empty", pVnode->vgId, pVnode->pStreamQ);
   while (!taosQueueEmpty(pVnode->pStreamQ)) taosMsleep(10);
 
@@ -141,7 +142,6 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal)
   dInfo("vgId:%d, vnode is closed", pVnode->vgId);
 
   if (commitAndRemoveWal) {
-    char path[TSDB_FILENAME_LEN] = {0};
     snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d%swal", TD_DIRSEP, pVnode->vgId, TD_DIRSEP);
     dInfo("vgId:%d, remove all wals, path:%s", pVnode->vgId, path);
     tfsRmdir(pMgmt->pTfs, path);

source/dnode/mnode/impl/src/mndDb.c

@@ -373,6 +373,8 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   if (pCfg->sstTrigger < TSDB_MIN_STT_TRIGGER || pCfg->sstTrigger > TSDB_MAX_STT_TRIGGER) return -1;
   if (pCfg->hashPrefix < TSDB_MIN_HASH_PREFIX || pCfg->hashPrefix > TSDB_MAX_HASH_PREFIX) return -1;
   if (pCfg->hashSuffix < TSDB_MIN_HASH_SUFFIX || pCfg->hashSuffix > TSDB_MAX_HASH_SUFFIX) return -1;
+  if ((pCfg->hashSuffix * pCfg->hashPrefix) < 0) return -1;
+  if ((pCfg->hashPrefix + pCfg->hashSuffix) >= (TSDB_TABLE_NAME_LEN - 1)) return -1;
   if (pCfg->tsdbPageSize < TSDB_MIN_TSDB_PAGESIZE || pCfg->tsdbPageSize > TSDB_MAX_TSDB_PAGESIZE) return -1;
   if (taosArrayGetSize(pCfg->pRetensions) != pCfg->numOfRetensions) return -1;
 
@@ -409,8 +411,6 @@ static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->walRollPeriod < 0) pCfg->walRollPeriod = TSDB_REPS_DEF_DB_WAL_ROLL_PERIOD;
   if (pCfg->walSegmentSize < 0) pCfg->walSegmentSize = TSDB_DEFAULT_DB_WAL_SEGMENT_SIZE;
   if (pCfg->sstTrigger <= 0) pCfg->sstTrigger = TSDB_DEFAULT_SST_TRIGGER;
-  if (pCfg->hashPrefix < 0) pCfg->hashPrefix = TSDB_DEFAULT_HASH_PREFIX;
-  if (pCfg->hashSuffix < 0) pCfg->hashSuffix = TSDB_DEFAULT_HASH_SUFFIX;
   if (pCfg->tsdbPageSize <= 0) pCfg->tsdbPageSize = TSDB_DEFAULT_TSDB_PAGESIZE;
 }
 
@@ -553,6 +553,10 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
     int32_t dbLen = strlen(dbObj.name) + 1;
     mInfo("db:%s, hashPrefix adjust from %d to %d", dbObj.name, dbObj.cfg.hashPrefix, dbObj.cfg.hashPrefix + dbLen);
     dbObj.cfg.hashPrefix += dbLen;
+  } else if (dbObj.cfg.hashPrefix < 0) {
+    int32_t dbLen = strlen(dbObj.name) + 1;
+    mInfo("db:%s, hashPrefix adjust from %d to %d", dbObj.name, dbObj.cfg.hashPrefix, dbObj.cfg.hashPrefix - dbLen);
+    dbObj.cfg.hashPrefix -= dbLen;
   }
 
   SVgObj *pVgroups = NULL;
@@ -1788,6 +1792,8 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
     int16_t hashPrefix = pDb->cfg.hashPrefix;
     if (hashPrefix > 0) {
       hashPrefix = pDb->cfg.hashPrefix - strlen(pDb->name) - 1;
+    } else if (hashPrefix < 0) {
+      hashPrefix = pDb->cfg.hashPrefix + strlen(pDb->name) + 1;
     }
     colDataSetVal(pColInfo, rows, (const char *)&hashPrefix, false);
 

source/dnode/vnode/src/inc/tq.h

@@ -180,16 +180,9 @@ int32_t tqStreamTasksScanWal(STQ* pTq);
 
 // tq util
 char*   createStreamTaskIdStr(int64_t streamId, int32_t taskId);
-void    createStreamTaskOffsetKey(char* dst, uint64_t streamId, uint32_t taskId);
 int32_t tqAddInputBlockNLaunchTask(SStreamTask* pTask, SStreamQueueItem* pQueueItem, int64_t ver);
-int32_t launchTaskForWalBlock(SStreamTask* pTask, SFetchRet* pRet, STqOffset* pOffset);
 int32_t tqExtractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg);
 
-void    doSaveTaskOffset(STqOffsetStore* pOffsetStore, const char* pKey, int64_t ver);
-void    saveOffsetForAllTasks(STQ* pTq, int64_t ver);
-void    initOffsetForAllRestoreTasks(STQ* pTq);
-int32_t transferToWalReadTask(SStreamMeta* pStreamMeta, SArray* pTaskList);
-
 #ifdef __cplusplus
 }
 #endif

source/dnode/vnode/src/inc/vnodeInt.h

@@ -190,6 +190,7 @@ int32_t tsdbSetKeepCfg(STsdb* pTsdb, STsdbCfg* pCfg);
 int  tqInit();
 void tqCleanUp();
 STQ* tqOpen(const char* path, SVnode* pVnode);
+void tqNotifyClose(STQ*);
 void tqClose(STQ*);
 int  tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver);
 int  tqRegisterPushHandle(STQ* pTq, void* pHandle, const SMqPollReq* pRequest, SRpcMsg* pRpcMsg, SMqDataRsp* pDataRsp,

source/dnode/vnode/src/tq/tq.c

@@ -154,6 +154,31 @@ void tqClose(STQ* pTq) {
   taosMemoryFree(pTq);
 }
 
+void tqNotifyClose(STQ* pTq) {
+  if (pTq != NULL) {
+    taosWLockLatch(&pTq->pStreamMeta->lock);
+
+    void* pIter = NULL;
+    while (1) {
+      pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
+      if (pIter == NULL) {
+        break;
+      }
+
+      SStreamTask* pTask = *(SStreamTask**)pIter;
+      tqDebug("vgId:%d s-task:%s set dropping flag", pTq->pStreamMeta->vgId, pTask->id.idStr);
+      pTask->status.taskStatus = TASK_STATUS__STOP;
+
+      int64_t st = taosGetTimestampMs();
+      qKillTask(pTask->exec.pExecutor, TSDB_CODE_SUCCESS);
+      int64_t el = taosGetTimestampMs() - st;
+      tqDebug("vgId:%d s-task:%s is closed in %" PRId64 "ms", pTq->pStreamMeta->vgId, pTask->id.idStr, el);
+    }
+
+    taosWUnLockLatch(&pTq->pStreamMeta->lock);
+  }
+}
+
 static int32_t doSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqDataRsp* pRsp, int32_t epoch,
                              int64_t consumerId, int32_t type) {
   int32_t len = 0;
@@ -573,6 +598,7 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver) {
   pTask->pMsgCb = &pTq->pVnode->msgCb;
   pTask->pMeta = pTq->pStreamMeta;
   pTask->chkInfo.version = ver;
+  pTask->chkInfo.currentVer = ver;
 
   // expand executor
   if (pTask->fillHistory) {
@@ -966,14 +992,21 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
   int32_t* pRef = taosMemoryMalloc(sizeof(int32_t));
   *pRef = 1;
 
+  taosWLockLatch(&pTq->pStreamMeta->lock);
+
   void* pIter = NULL;
   while (1) {
     pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
-    if (pIter == NULL) break;
+    if (pIter == NULL) {
+      break;
+    }
+
     SStreamTask* pTask = *(SStreamTask**)pIter;
-    if (pTask->taskLevel != TASK_LEVEL__SOURCE) continue;
+    if (pTask->taskLevel != TASK_LEVEL__SOURCE) {
+      continue;
+    }
 
-    qDebug("delete req enqueue stream task: %d, ver: %" PRId64, pTask->id.taskId, ver);
+    qDebug("s-task:%s delete req enqueue, ver: %" PRId64, pTask->id.idStr, ver);
 
     if (!failed) {
       SStreamRefDataBlock* pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0);
@@ -983,15 +1016,13 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
       atomic_add_fetch_32(pRefBlock->dataRef, 1);
 
       if (tAppendDataToInputQueue(pTask, (SStreamQueueItem*)pRefBlock) < 0) {
-        qError("stream task input del failed, task id %d", pTask->id.taskId);
-
         atomic_sub_fetch_32(pRef, 1);
         taosFreeQitem(pRefBlock);
         continue;
       }
 
       if (streamSchedExec(pTask) < 0) {
-        qError("stream task launch failed, task id %d", pTask->id.taskId);
+        qError("s-task:%s stream task launch failed", pTask->id.idStr);
         continue;
       }
 
@@ -1000,8 +1031,9 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
     }
   }
 
+  taosWUnLockLatch(&pTq->pStreamMeta->lock);
+
   int32_t ref = atomic_sub_fetch_32(pRef, 1);
-  /*A(ref >= 0);*/
   if (ref == 0) {
     blockDataDestroy(pDelBlock);
     taosMemoryFree(pRef);
@@ -1032,23 +1064,9 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
   }
   blockDataDestroy(pDelBlock);
 #endif
-
   return 0;
 }
 
-static int32_t addSubmitBlockNLaunchTask(STqOffsetStore* pOffsetStore, SStreamTask* pTask, SStreamDataSubmit2* pSubmit,
-                                         const char* key, int64_t ver) {
-  doSaveTaskOffset(pOffsetStore, key, ver);
-  int32_t code = tqAddInputBlockNLaunchTask(pTask, (SStreamQueueItem*)pSubmit, ver);
-
-  // remove the offset, if all functions are completed successfully.
-  if (code == TSDB_CODE_SUCCESS) {
-    tqOffsetDelete(pOffsetStore, key);
-  }
-
-  return code;
-}
-
 int32_t tqProcessSubmitReq(STQ* pTq, SPackedData submit) {
 #if 0
   void* pIter = NULL;
@@ -1310,8 +1328,6 @@ int32_t tqStartStreamTasks(STQ* pTq) {
   }
 
   tqDebug("vgId:%d start wal scan stream tasks, tasks:%d", vgId, numOfTasks);
-  initOffsetForAllRestoreTasks(pTq);
-
   pRunReq->head.vgId = vgId;
   pRunReq->streamId = 0;
   pRunReq->taskId = WAL_READ_TASKS_ID;

source/dnode/vnode/src/tq/tqRead.c

@@ -1023,6 +1023,7 @@ int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
   }
 
   // update the table list handle for each stream scanner/wal reader
+  taosWLockLatch(&pTq->pStreamMeta->lock);
   while (1) {
     pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
     if (pIter == NULL) {
@@ -1039,5 +1040,7 @@ int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
     }
   }
 
+  taosWUnLockLatch(&pTq->pStreamMeta->lock);
+
   return 0;
 }

source/dnode/vnode/src/tq/tqRestore.c

@@ -15,12 +15,12 @@
 
 #include "tq.h"
 
-static int32_t doCreateReqsByScanWal(SStreamMeta* pStreamMeta, STqOffsetStore* pOffsetStore, bool* pScanIdle);
+static int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle);
 
 // this function should be executed by stream threads.
 // there is a case that the WAL increases more fast than the restore procedure, and this restore procedure
 // will not stop eventually.
-int tqStreamTasksScanWal(STQ* pTq) {
+int32_t tqStreamTasksScanWal(STQ* pTq) {
   int32_t vgId = TD_VID(pTq->pVnode);
   SStreamMeta* pMeta = pTq->pStreamMeta;
   int64_t st = taosGetTimestampMs();
@@ -31,7 +31,7 @@ int tqStreamTasksScanWal(STQ* pTq) {
 
     // check all restore tasks
     bool shouldIdle = true;
-    doCreateReqsByScanWal(pTq->pStreamMeta, pTq->pOffsetStore, &shouldIdle);
+    createStreamRunReq(pTq->pStreamMeta, &shouldIdle);
 
     int32_t times = 0;
 
@@ -76,7 +76,7 @@ static SArray* extractTaskIdList(SStreamMeta* pStreamMeta, int32_t numOfTasks) {
   return pTaskIdList;
 }
 
-int32_t doCreateReqsByScanWal(SStreamMeta* pStreamMeta, STqOffsetStore* pOffsetStore, bool* pScanIdle) {
+int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
   *pScanIdle = true;
   bool    noNewDataInWal = true;
   int32_t vgId = pStreamMeta->vgId;
@@ -89,6 +89,8 @@ int32_t doCreateReqsByScanWal(SStreamMeta* pStreamMeta, STqOffsetStore* pOffsetS
   tqDebug("vgId:%d start to check wal to extract new submit block for %d tasks", vgId, numOfTasks);
   SArray* pTaskIdList = extractTaskIdList(pStreamMeta, numOfTasks);
 
+  // update the new task number
+  numOfTasks = taosArrayGetSize(pTaskIdList);
   for (int32_t i = 0; i < numOfTasks; ++i) {
     int32_t*     pTaskId = taosArrayGet(pTaskIdList, i);
     SStreamTask* pTask = streamMetaAcquireTask(pStreamMeta, *pTaskId);
@@ -96,23 +98,20 @@ int32_t doCreateReqsByScanWal(SStreamMeta* pStreamMeta, STqOffsetStore* pOffsetS
       continue;
     }
 
+    int32_t status = pTask->status.taskStatus;
     if (pTask->taskLevel != TASK_LEVEL__SOURCE) {
+      tqDebug("s-task:%s not source task, no need to start", pTask->id.idStr);
       streamMetaReleaseTask(pStreamMeta, pTask);
       continue;
     }
 
-    if (pTask->status.taskStatus == TASK_STATUS__RECOVER_PREPARE ||
-        pTask->status.taskStatus == TASK_STATUS__WAIT_DOWNSTREAM) {
-      tqDebug("s-task:%s skip push data, not ready for processing, status %d", pTask->id.idStr,
-              pTask->status.taskStatus);
+    if (streamTaskShouldStop(&pTask->status) || status == TASK_STATUS__RECOVER_PREPARE ||
+        status == TASK_STATUS__WAIT_DOWNSTREAM) {
+      tqDebug("s-task:%s skip push data, not ready for processing, status %d", pTask->id.idStr, status);
       streamMetaReleaseTask(pStreamMeta, pTask);
       continue;
     }
 
-    // check if offset value exists
-    char key[128] = {0};
-    createStreamTaskOffsetKey(key, pTask->id.streamId, pTask->id.taskId);
-
     if (tInputQueueIsFull(pTask)) {
       tqDebug("vgId:%d s-task:%s input queue is full, do nothing", vgId, pTask->id.idStr);
       streamMetaReleaseTask(pStreamMeta, pTask);
@@ -121,19 +120,15 @@ int32_t doCreateReqsByScanWal(SStreamMeta* pStreamMeta, STqOffsetStore* pOffsetS
 
     *pScanIdle = false;
 
-    // check if offset value exists
-    STqOffset* pOffset = tqOffsetRead(pOffsetStore, key);
-    ASSERT(pOffset != NULL);
-
     // seek the stored version and extract data from WAL
-    int32_t code = walReadSeekVer(pTask->exec.pWalReader, pOffset->val.version);
+    int32_t code = walReadSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer);
     if (code != TSDB_CODE_SUCCESS) {  // no data in wal, quit
       streamMetaReleaseTask(pStreamMeta, pTask);
       continue;
     }
 
     // append the data for the stream
-    tqDebug("vgId:%d wal reader seek to ver:%" PRId64 " %s", vgId, pOffset->val.version, pTask->id.idStr);
+    tqDebug("vgId:%d s-task:%s wal reader seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer);
 
     SPackedData packData = {0};
     code = extractSubmitMsgFromWal(pTask->exec.pWalReader, &packData);
@@ -152,14 +147,13 @@ int32_t doCreateReqsByScanWal(SStreamMeta* pStreamMeta, STqOffsetStore* pOffsetS
 
     noNewDataInWal = false;
 
-    tqDebug("s-task:%s submit data extracted from WAL", pTask->id.idStr);
     code = tqAddInputBlockNLaunchTask(pTask, (SStreamQueueItem*)p, packData.ver);
     if (code == TSDB_CODE_SUCCESS) {
-      pOffset->val.version = walReaderGetCurrentVer(pTask->exec.pWalReader);
+      pTask->chkInfo.currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);
       tqDebug("s-task:%s set the ver:%" PRId64 " from WALReader after extract block from WAL", pTask->id.idStr,
-              pOffset->val.version);
+              pTask->chkInfo.currentVer);
     } else {
-      // do nothing
+      tqError("s-task:%s append input queue failed, ver:%"PRId64, pTask->id.idStr, pTask->chkInfo.currentVer);
     }
 
     streamDataSubmitDestroy(p);

source/dnode/vnode/src/tq/tqUtil.c

@@ -25,21 +25,6 @@ char* createStreamTaskIdStr(int64_t streamId, int32_t taskId) {
   return taosStrdup(buf);
 }
 
-// stream_task:stream_id:task_id
-void createStreamTaskOffsetKey(char* dst, uint64_t streamId, uint32_t taskId) {
-  int32_t n = 12;
-  char* p = dst;
-
-  memcpy(p, "stream_task:", n);
-  p += n;
-
-  int32_t inc = tintToHex(streamId, p);
-  p += inc;
-
-  *(p++) = ':';
-  tintToHex(taskId, p);
-}
-
 int32_t tqAddInputBlockNLaunchTask(SStreamTask* pTask, SStreamQueueItem* pQueueItem, int64_t ver) {
   int32_t code = tAppendDataToInputQueue(pTask, pQueueItem);
   if (code < 0) {
@@ -55,75 +40,6 @@ int32_t tqAddInputBlockNLaunchTask(SStreamTask* pTask, SStreamQueueItem* pQueueI
   return TSDB_CODE_SUCCESS;
 }
 
-void initOffsetForAllRestoreTasks(STQ* pTq) {
-  void* pIter = NULL;
-
-  while(1) {
-    pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
-    if (pIter == NULL) {
-      break;
-    }
-
-    SStreamTask* pTask = *(SStreamTask**)pIter;
-    if (pTask->taskLevel != TASK_LEVEL__SOURCE) {
-      continue;
-    }
-
-    if (pTask->status.taskStatus == TASK_STATUS__RECOVER_PREPARE || pTask->status.taskStatus == TASK_STATUS__WAIT_DOWNSTREAM) {
-      tqDebug("s-task:%s skip push data, since not ready, status %d", pTask->id.idStr, pTask->status.taskStatus);
-      continue;
-    }
-
-    char key[128] = {0};
-    createStreamTaskOffsetKey(key, pTask->id.streamId, pTask->id.taskId);
-
-    STqOffset* pOffset = tqOffsetRead(pTq->pOffsetStore, key);
-    if (pOffset == NULL) {
-      doSaveTaskOffset(pTq->pOffsetStore, key, pTask->chkInfo.version);
-    }
-  }
-}
-
-void saveOffsetForAllTasks(STQ* pTq, int64_t ver) {
-  void* pIter = NULL;
-
-  while(1) {
-    pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
-    if (pIter == NULL) {
-      break;
-    }
-
-    SStreamTask* pTask = *(SStreamTask**)pIter;
-    if (pTask->taskLevel != TASK_LEVEL__SOURCE) {
-      continue;
-    }
-
-    if (pTask->status.taskStatus == TASK_STATUS__RECOVER_PREPARE || pTask->status.taskStatus == TASK_STATUS__WAIT_DOWNSTREAM) {
-      tqDebug("s-task:%s skip push data, not ready for processing, status %d", pTask->id.idStr,
-              pTask->status.taskStatus);
-      continue;
-    }
-
-    char key[128] = {0};
-    createStreamTaskOffsetKey(key, pTask->id.streamId, pTask->id.taskId);
-
-    STqOffset* pOffset = tqOffsetRead(pTq->pOffsetStore, key);
-    if (pOffset == NULL) {
-      doSaveTaskOffset(pTq->pOffsetStore, key, ver);
-    }
-  }
-}
-
-void doSaveTaskOffset(STqOffsetStore* pOffsetStore, const char* pKey, int64_t ver) {
-  STqOffset offset = {0};
-  tqOffsetResetToLog(&offset.val, ver);
-
-  tstrncpy(offset.subKey, pKey, tListLen(offset.subKey));
-
-  // keep the offset info in the offset store
-  tqOffsetWrite(pOffsetStore, &offset);
-}
-
 static int32_t tqInitDataRsp(SMqDataRsp* pRsp, const SMqPollReq* pReq, int8_t subType) {
   pRsp->reqOffset = pReq->reqOffset;
 

source/libs/command/src/command.c

@@ -278,7 +278,12 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char* dbName, ch
 
   char* retentions = buildRetension(pCfg->pRetensions);
   int32_t dbFNameLen = strlen(dbFName);
-  int32_t hashPrefix = (pCfg->hashPrefix > (dbFNameLen + 1)) ? (pCfg->hashPrefix - dbFNameLen - 1) : 0;
+  int32_t hashPrefix = 0;
+  if (pCfg->hashPrefix > 0) {
+    hashPrefix = pCfg->hashPrefix - dbFNameLen - 1;
+  } else if (pCfg->hashPrefix < 0) {
+    hashPrefix = pCfg->hashPrefix + dbFNameLen + 1;
+  }
 
   len += sprintf(
       buf2 + VARSTR_HEADER_SIZE,

source/libs/executor/src/executor.c

@@ -1108,6 +1108,11 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
       pScanBaseInfo->dataReader = NULL;
 
       // let's seek to the next version in wal file
+      int64_t firstVer = walGetFirstVer(pInfo->tqReader->pWalReader->pWal);
+      if (pOffset->version + 1 < firstVer){
+        pOffset->version = firstVer - 1;
+      }
+
       if (tqSeekVer(pInfo->tqReader, pOffset->version + 1, id) < 0) {
         qError("tqSeekVer failed ver:%" PRId64 ", %s", pOffset->version + 1, id);
         return -1;

source/libs/executor/src/sysscanoperator.c

@@ -1482,11 +1482,7 @@ static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) {
           pInfo->pIdx->init = 1;
           SSDataBlock* blk = sysTableBuildUserTablesByUids(pOperator);
           return blk;
-        } else if (flt == -2) {
-          qDebug("%s failed to get sys table info by idx, empty result", GET_TASKID(pTaskInfo));
-          return NULL;
-        } else if (flt == -1) {
-          // not idx
+        } else if ((flt == -1) || (flt == -2)) {
           qDebug("%s failed to get sys table info by idx, scan sys table one by one", GET_TASKID(pTaskInfo));
         }
       } else if (pCondition != NULL && (pInfo->pIdx != NULL && pInfo->pIdx->init == 1)) {

source/libs/executor/src/timewindowoperator.c

@@ -2541,6 +2541,20 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
   }
 
   while (1) {
+    if (isTaskKilled(pTaskInfo)) {
+
+      if (pInfo->pUpdated != NULL) {
+        pInfo->pUpdated = taosArrayDestroy(pInfo->pUpdated);
+      }
+
+      if (pInfo->pUpdatedMap != NULL) {
+        tSimpleHashCleanup(pInfo->pUpdatedMap);
+        pInfo->pUpdatedMap = NULL;
+      }
+
+      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
+    }
+
     SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
     if (pBlock == NULL) {
       pOperator->status = OP_RES_TO_RETURN;
@@ -2635,6 +2649,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
   while ((pIte = tSimpleHashIterate(pInfo->pUpdatedMap, pIte, &iter)) != NULL) {
     taosArrayPush(pInfo->pUpdated, pIte);
   }
+
   tSimpleHashCleanup(pInfo->pUpdatedMap);
   pInfo->pUpdatedMap = NULL;
   taosArraySort(pInfo->pUpdated, winKeyCmprImpl);

source/libs/function/CMakeLists.txt

@@ -79,6 +79,26 @@ ENDIF ()
 target_link_libraries(
     udf1 PUBLIC os ${LINK_JEMALLOC})
 
+
+add_library(udf1_dup STATIC MODULE test/udf1_dup.c)
+target_include_directories(
+        udf1_dup
+        PUBLIC
+            "${TD_SOURCE_DIR}/include/libs/function"
+            "${TD_SOURCE_DIR}/include/util"
+            "${TD_SOURCE_DIR}/include/common"
+            "${TD_SOURCE_DIR}/include/client"
+            "${TD_SOURCE_DIR}/include/os"
+        PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
+
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    ADD_DEPENDENCIES(udf1_dup jemalloc)
+ENDIF ()
+
+target_link_libraries(
+    udf1_dup PUBLIC os ${LINK_JEMALLOC})
+
 add_library(udf2 STATIC MODULE test/udf2.c)
 target_include_directories(
         udf2
@@ -99,6 +119,26 @@ target_link_libraries(
     udf2 PUBLIC os ${LINK_JEMALLOC}
 )
 
+add_library(udf2_dup STATIC MODULE test/udf2_dup.c)
+target_include_directories(
+        udf2_dup
+        PUBLIC
+            "${TD_SOURCE_DIR}/include/libs/function"
+            "${TD_SOURCE_DIR}/include/util"
+            "${TD_SOURCE_DIR}/include/common"
+            "${TD_SOURCE_DIR}/include/client"
+            "${TD_SOURCE_DIR}/include/os"
+        PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
+
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    ADD_DEPENDENCIES(udf2_dup jemalloc)
+ENDIF ()
+
+target_link_libraries(
+    udf2_dup PUBLIC os ${LINK_JEMALLOC}
+)
+
 #SET(EXECUTABLE_OUTPUT_PATH ${CMAKE_BINARY_DIR}/build/bin)
 add_executable(udfd src/udfd.c)
 target_include_directories(

source/libs/function/test/udf1_dup.c

+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#ifdef LINUX
+#include <unistd.h>
+#endif
+#ifdef WINDOWS
+#include <windows.h>
+#endif
+#include "taosudf.h"
+
+
+DLL_EXPORT int32_t udf1_dup_init() { return 0; }
+
+DLL_EXPORT int32_t udf1_dup_destroy() { return 0; }
+
+DLL_EXPORT int32_t udf1_dup(SUdfDataBlock *block, SUdfColumn *resultCol) {
+  SUdfColumnData *resultData = &resultCol->colData;
+  for (int32_t i = 0; i < block->numOfRows; ++i) {
+    int j = 0;
+    for (; j < block->numOfCols; ++j) {
+      if (udfColDataIsNull(block->udfCols[j], i)) {
+        udfColDataSetNull(resultCol, i);
+        break;
+      }
+    }
+    if (j == block->numOfCols) {
+      int32_t luckyNum = 2;
+      udfColDataSet(resultCol, i, (char *)&luckyNum, false);
+    }
+  }
+  // to simulate actual processing delay by udf
+#ifdef LINUX
+  usleep(1 * 1000);  // usleep takes sleep time in us (1 millionth of a second)
+#endif
+#ifdef WINDOWS
+  Sleep(1);
+#endif
+  resultData->numOfRows = block->numOfRows;
+  return 0;
+}
+

source/libs/function/test/udf2_dup.c

+#include <math.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "taosudf.h"
+
+DLL_EXPORT int32_t udf2_dup_init() { return 0; }
+
+DLL_EXPORT int32_t udf2_dup_destroy() { return 0; }
+
+DLL_EXPORT int32_t udf2_dup_start(SUdfInterBuf* buf) {
+  *(int64_t*)(buf->buf) = 0;
+  buf->bufLen = sizeof(double);
+  buf->numOfResult = 1;
+  return 0;
+}
+
+DLL_EXPORT int32_t udf2_dup(SUdfDataBlock* block, SUdfInterBuf* interBuf, SUdfInterBuf* newInterBuf) {
+  double sumSquares = 0;
+  if (interBuf->numOfResult == 1) {
+    sumSquares = *(double*)interBuf->buf;
+  }
+  int8_t numNotNull = 0;
+  for (int32_t i = 0; i < block->numOfCols; ++i) {
+    SUdfColumn* col = block->udfCols[i];
+    if (!(col->colMeta.type == TSDB_DATA_TYPE_INT || col->colMeta.type == TSDB_DATA_TYPE_DOUBLE)) {
+      return TSDB_CODE_UDF_INVALID_INPUT;
+    }
+  }
+  for (int32_t i = 0; i < block->numOfCols; ++i) {
+    for (int32_t j = 0; j < block->numOfRows; ++j) {
+      SUdfColumn* col = block->udfCols[i];
+      if (udfColDataIsNull(col, j)) {
+        continue;
+      }
+      switch (col->colMeta.type) {
+        case TSDB_DATA_TYPE_INT: {
+          char*   cell = udfColDataGetData(col, j);
+          int32_t num = *(int32_t*)cell;
+          sumSquares += (double)num * num;
+          break;
+        }
+        case TSDB_DATA_TYPE_DOUBLE: {
+          char*  cell = udfColDataGetData(col, j);
+          double num = *(double*)cell;
+          sumSquares += num * num;
+          break;
+        }
+        default:
+          break;
+      }
+      ++numNotNull;
+    }
+  }
+
+  *(double*)(newInterBuf->buf) = sumSquares;
+  newInterBuf->bufLen = sizeof(double);
+
+  if (interBuf->numOfResult == 0 && numNotNull == 0) {
+    newInterBuf->numOfResult = 0;
+  } else {
+    newInterBuf->numOfResult = 1;
+  }
+  return 0;
+}
+
+DLL_EXPORT int32_t udf2_dup_finish(SUdfInterBuf* buf, SUdfInterBuf* resultData) {
+  if (buf->numOfResult == 0) {
+    resultData->numOfResult = 0;
+    return 0;
+  }
+  double sumSquares = *(double*)(buf->buf);
+  *(double*)(resultData->buf) = sqrt(sumSquares) + 100;
+  resultData->bufLen = sizeof(double);
+  resultData->numOfResult = 1;
+  return 0;
+}

source/libs/parser/inc/sql.y

@@ -221,8 +221,8 @@ db_options(A) ::= db_options(B) WAL_RETENTION_SIZE NK_MINUS(D) NK_INTEGER(C).
 db_options(A) ::= db_options(B) WAL_ROLL_PERIOD NK_INTEGER(C).                    { A = setDatabaseOption(pCxt, B, DB_OPTION_WAL_ROLL_PERIOD, &C); }
 db_options(A) ::= db_options(B) WAL_SEGMENT_SIZE NK_INTEGER(C).                   { A = setDatabaseOption(pCxt, B, DB_OPTION_WAL_SEGMENT_SIZE, &C); }
 db_options(A) ::= db_options(B) STT_TRIGGER NK_INTEGER(C).                        { A = setDatabaseOption(pCxt, B, DB_OPTION_STT_TRIGGER, &C); }
-db_options(A) ::= db_options(B) TABLE_PREFIX NK_INTEGER(C).                       { A = setDatabaseOption(pCxt, B, DB_OPTION_TABLE_PREFIX, &C); }
-db_options(A) ::= db_options(B) TABLE_SUFFIX NK_INTEGER(C).                       { A = setDatabaseOption(pCxt, B, DB_OPTION_TABLE_SUFFIX, &C); }
+db_options(A) ::= db_options(B) TABLE_PREFIX signed(C).                           { A = setDatabaseOption(pCxt, B, DB_OPTION_TABLE_PREFIX, C); }
+db_options(A) ::= db_options(B) TABLE_SUFFIX signed(C).                           { A = setDatabaseOption(pCxt, B, DB_OPTION_TABLE_SUFFIX, C); }
 
 alter_db_options(A) ::= alter_db_option(B).                                       { A = createAlterDatabaseOptions(pCxt); A = setAlterDatabaseOption(pCxt, A, &B); }
 alter_db_options(A) ::= alter_db_options(B) alter_db_option(C).                   { A = setAlterDatabaseOption(pCxt, B, &C); }

source/libs/parser/src/parAstCreater.c

@@ -1024,12 +1024,28 @@ static SNode* setDatabaseOptionImpl(SAstCreateContext* pCxt, SNode* pOptions, ED
     case DB_OPTION_STT_TRIGGER:
       pDbOptions->sstTrigger = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
-    case DB_OPTION_TABLE_PREFIX:
-      pDbOptions->tablePrefix = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+    case DB_OPTION_TABLE_PREFIX: {
+      SValueNode *pNode = (SValueNode *)pVal;
+      if (TSDB_DATA_TYPE_BIGINT == pNode->node.resType.type || TSDB_DATA_TYPE_UBIGINT == pNode->node.resType.type) {
+        pDbOptions->tablePrefix = taosStr2Int32(pNode->literal, NULL, 10);
+      } else {
+        snprintf(pCxt->pQueryCxt->pMsg, pCxt->pQueryCxt->msgLen, "invalid table_prefix data type");
+        pCxt->errCode = TSDB_CODE_PAR_SYNTAX_ERROR;        
+      }
+      nodesDestroyNode((SNode*)pNode);
       break;
-    case DB_OPTION_TABLE_SUFFIX:
-      pDbOptions->tableSuffix = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+    }
+    case DB_OPTION_TABLE_SUFFIX:{
+      SValueNode *pNode = (SValueNode *)pVal;
+      if (TSDB_DATA_TYPE_BIGINT == pNode->node.resType.type || TSDB_DATA_TYPE_UBIGINT == pNode->node.resType.type) {
+        pDbOptions->tableSuffix = taosStr2Int32(pNode->literal, NULL, 10);
+      } else {
+        snprintf(pCxt->pQueryCxt->pMsg, pCxt->pQueryCxt->msgLen, "invalid table_suffix data type");
+        pCxt->errCode = TSDB_CODE_PAR_SYNTAX_ERROR;        
+      }
+      nodesDestroyNode((SNode*)pNode);
       break;
+    }
     default:
       break;
   }

source/libs/parser/src/parTranslater.c

@@ -4177,6 +4177,34 @@ static int32_t checkDbRetentionsOption(STranslateContext* pCxt, SNodeList* pRete
   return TSDB_CODE_SUCCESS;
 }
 
+
+static int32_t checkDbTbPrefixSuffixOptions(STranslateContext* pCxt, int32_t tbPrefix, int32_t tbSuffix) {
+  if (tbPrefix < TSDB_MIN_HASH_PREFIX || tbPrefix > TSDB_MAX_HASH_PREFIX) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
+                                   "Invalid option table_prefix: %d valid range: [%d, %d]", tbPrefix,
+                                   TSDB_MIN_HASH_PREFIX, TSDB_MAX_HASH_PREFIX);
+  }
+
+  if (tbSuffix < TSDB_MIN_HASH_SUFFIX || tbSuffix > TSDB_MAX_HASH_SUFFIX) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
+                                   "Invalid option table_suffix: %d valid range: [%d, %d]", tbSuffix,
+                                   TSDB_MIN_HASH_SUFFIX, TSDB_MAX_HASH_SUFFIX);
+  }
+
+  if ((tbPrefix * tbSuffix) < 0) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
+                                   "Invalid option table_prefix & table_suffix: mixed usage not allowed");
+  }
+
+  if ((tbPrefix + tbSuffix) >= (TSDB_TABLE_NAME_LEN - 1)) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
+                                   "Invalid option table_prefix & table_suffix: exceed max table name length");
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 static int32_t checkOptionsDependency(STranslateContext* pCxt, const char* pDbName, SDatabaseOptions* pOptions) {
   int32_t daysPerFile = pOptions->daysPerFile;
   int64_t daysToKeep0 = pOptions->keep[0];
@@ -4284,10 +4312,7 @@ static int32_t checkDatabaseOptions(STranslateContext* pCxt, const char* pDbName
     code = checkDbRangeOption(pCxt, "sstTrigger", pOptions->sstTrigger, TSDB_MIN_STT_TRIGGER, TSDB_MAX_STT_TRIGGER);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = checkDbRangeOption(pCxt, "tablePrefix", pOptions->tablePrefix, TSDB_MIN_HASH_PREFIX, TSDB_MAX_HASH_PREFIX);
-  }
-  if (TSDB_CODE_SUCCESS == code) {
-    code = checkDbRangeOption(pCxt, "tableSuffix", pOptions->tableSuffix, TSDB_MIN_HASH_SUFFIX, TSDB_MAX_HASH_SUFFIX);
+    code = checkDbTbPrefixSuffixOptions(pCxt, pOptions->tablePrefix, pOptions->tableSuffix);
   }
   if (TSDB_CODE_SUCCESS == code) {
     code = checkOptionsDependency(pCxt, pDbName, pOptions);

source/libs/planner/src/planOptimizer.c

@@ -2526,7 +2526,7 @@ static bool tbCntScanOptIsEligibleAggFuncs(SNodeList* pAggFuncs) {
       return false;
     }
   }
-  return true;
+  return LIST_LENGTH(pAggFuncs) > 0;
 }
 
 static bool tbCntScanOptIsEligibleAgg(SAggLogicNode* pAgg) {

source/libs/stream/src/stream.c

@@ -16,7 +16,7 @@
 #include "streamInc.h"
 #include "ttimer.h"
 
-#define STREAM_TASK_INPUT_QUEUEU_CAPACITY 100000
+#define STREAM_TASK_INPUT_QUEUEU_CAPACITY 3000
 
 int32_t streamInit() {
   int8_t old;

source/libs/stream/src/streamExec.c

@@ -17,6 +17,11 @@
 
 #define STREAM_EXEC_MAX_BATCH_NUM 100
 
+bool streamTaskShouldStop(const SStreamStatus* pStatus) {
+  int32_t status = atomic_load_8((int8_t*) &pStatus->taskStatus);
+  return (status == TASK_STATUS__STOP) || (status == TASK_STATUS__DROPPING);
+}
+
 static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray* pRes) {
   int32_t code = TSDB_CODE_SUCCESS;
   void*   pExecutor = pTask->exec.pExecutor;
@@ -66,7 +71,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray*
 
   // pExecutor
   while (1) {
-    if (pTask->status.taskStatus == TASK_STATUS__DROPPING) {
+    if (streamTaskShouldStop(&pTask->status)) {
       return 0;
     }
 
@@ -106,7 +111,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray*
       continue;
     }
 
-    qDebug("task %d(child %d) executed and get block", pTask->id.taskId, pTask->selfChildId);
+    qDebug("s-task:%s (child %d) executed and get block", pTask->id.idStr, pTask->selfChildId);
 
     SSDataBlock block = {0};
     assignOneDataBlock(&block, output);
@@ -134,7 +139,7 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
 
     int32_t batchCnt = 0;
     while (1) {
-      if (atomic_load_8(&pTask->status.taskStatus) == TASK_STATUS__DROPPING) {
+      if (streamTaskShouldStop(&pTask->status)) {
         taosArrayDestroy(pRes);
         return 0;
       }
@@ -270,7 +275,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
       }
     }
 
-    if (pTask->status.taskStatus == TASK_STATUS__DROPPING) {
+    if (streamTaskShouldStop(&pTask->status)) {
       if (pInput) {
         streamFreeQitem(pInput);
       }
@@ -301,7 +306,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
              ", checkPoint id:%" PRId64 " -> %" PRId64,
              pTask->id.idStr, pTask->chkInfo.version, dataVer, pTask->chkInfo.id, ckId);
 
-      pTask->chkInfo = (SCheckpointInfo) {.version = dataVer, .id = ckId};
+      pTask->chkInfo = (SCheckpointInfo) {.version = dataVer, .id = ckId, .currentVer = pTask->chkInfo.currentVer};
 
       taosWLockLatch(&pTask->pMeta->lock);
       streamMetaSaveTask(pTask->pMeta, pTask);
@@ -368,7 +373,7 @@ int32_t streamTryExec(SStreamTask* pTask) {
     atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
     qDebug("s-task:%s exec completed", pTask->id.idStr);
 
-    if (!taosQueueEmpty(pTask->inputQueue->queue)) {
+    if (!taosQueueEmpty(pTask->inputQueue->queue) && (!streamTaskShouldStop(&pTask->status))) {
       streamSchedExec(pTask);
     }
   }

source/libs/stream/src/streamMeta.c

@@ -191,11 +191,13 @@ SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int32_t taskId) {
   taosRLockLatch(&pMeta->lock);
 
   SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, &taskId, sizeof(int32_t));
-  if (ppTask != NULL && (atomic_load_8(&((*ppTask)->status.taskStatus)) != TASK_STATUS__DROPPING)) {
+  if (ppTask != NULL) {
+    if (!streamTaskShouldStop(&(*ppTask)->status)) {
       atomic_add_fetch_32(&(*ppTask)->refCnt, 1);
       taosRUnLockLatch(&pMeta->lock);
       return *ppTask;
     }
+  }
 
   taosRUnLockLatch(&pMeta->lock);
   return NULL;
@@ -205,7 +207,7 @@ void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask) {
   int32_t left = atomic_sub_fetch_32(&pTask->refCnt, 1);
   ASSERT(left >= 0);
   if (left == 0) {
-    ASSERT(atomic_load_8(&pTask->status.taskStatus) == TASK_STATUS__DROPPING);
+    ASSERT(streamTaskShouldStop(&pTask->status));
     tFreeStreamTask(pTask);
   }
 }
@@ -216,11 +218,8 @@ void streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId) {
     SStreamTask* pTask = *ppTask;
     taosHashRemove(pMeta->pTasks, &taskId, sizeof(int32_t));
     tdbTbDelete(pMeta->pTaskDb, &taskId, sizeof(int32_t), pMeta->txn);
-    /*if (pTask->timer) {
-     * taosTmrStop(pTask->timer);*/
-    /*pTask->timer = NULL;*/
-    /*}*/
-    atomic_store_8(&pTask->status.taskStatus, TASK_STATUS__DROPPING);
+
+    atomic_store_8(&pTask->status.taskStatus, TASK_STATUS__STOP);
 
     taosWLockLatch(&pMeta->lock);
     streamMetaReleaseTask(pMeta, pTask);

tests/develop-test/5-taos-tools/taosbenchmark/json/rest_auto_create_table.json

@@ -11,6 +11,7 @@
   "confirm_parameter_prompt": "no",
   "prepared_rand": 100,
   "chinese": "no",
+  "escape_character": "yes",
   "insert_interval": 0,
   "num_of_records_per_req": 10,
   "databases": [{
@@ -29,7 +30,6 @@
       "child_table_exists":"no",
       "childtable_count": 8,
       "childtable_prefix": "stb_",
-      "escape_character": "yes",
       "auto_create_table": "yes",
       "batch_create_tbl_num": 10,
       "data_source": "rand",
@@ -54,7 +54,6 @@
       "child_table_exists":"no",
       "childtable_count": 8,
       "childtable_prefix": "stb3-2_",
-      "escape_character": "yes",
       "auto_create_table": "yes",
       "batch_create_tbl_num": 10,
       "data_source": "rand",

tests/develop-test/5-taos-tools/taosbenchmark/json/sml_auto_create_table.json

@@ -11,6 +11,7 @@
   "confirm_parameter_prompt": "no",
   "prepared_rand": 100,
   "chinese": "no",
+  "escape_character": "yes",
   "insert_interval": 0,
   "num_of_records_per_req": 10,
   "databases": [{
@@ -29,7 +30,6 @@
       "child_table_exists":"no",
       "childtable_count": 8,
       "childtable_prefix": "stb_",
-      "escape_character": "yes",
       "auto_create_table": "yes",
       "batch_create_tbl_num": 10,
       "data_source": "rand",
@@ -54,7 +54,6 @@
       "child_table_exists":"no",
       "childtable_count": 8,
       "childtable_prefix": "stb4-2_",
-      "escape_character": "yes",
       "auto_create_table": "yes",
       "batch_create_tbl_num": 10,
       "data_source": "rand",

tests/develop-test/5-taos-tools/taosbenchmark/json/stmt_auto_create_table.json

@@ -11,6 +11,7 @@
   "confirm_parameter_prompt": "no",
   "prepared_rand": 100,
   "chinese": "no",
+  "escape_character": "yes",
   "insert_interval": 0,
   "num_of_records_per_req": 10,
   "databases": [{
@@ -29,7 +30,6 @@
       "child_table_exists":"no",
       "childtable_count": 8,
       "childtable_prefix": "stb_",
-      "escape_character": "yes",
       "auto_create_table": "yes",
       "batch_create_tbl_num": 10,
       "data_source": "rand",
@@ -54,7 +54,6 @@
       "child_table_exists":"no",
       "childtable_count": 8,
       "childtable_prefix": "stb2-2_",
-      "escape_character": "yes",
       "auto_create_table": "yes",
       "batch_create_tbl_num": 10,
       "data_source": "rand",

tests/develop-test/5-taos-tools/taosbenchmark/json/taosc_auto_create_table.json

@@ -11,6 +11,7 @@
   "confirm_parameter_prompt": "no",
   "prepared_rand": 100,
   "chinese": "no",
+  "escape_character": "yes",
   "insert_interval": 0,
   "num_of_records_per_req": 10,
   "databases": [{
@@ -29,7 +30,6 @@
       "child_table_exists":"no",
       "childtable_count": 8,
       "childtable_prefix": "stb_",
-      "escape_character": "yes",
       "auto_create_table": "yes",
       "batch_create_tbl_num": 10,
       "data_source": "rand",
@@ -55,7 +55,6 @@
       "child_table_exists":"no",
       "childtable_count": 8,
       "childtable_prefix": "stb1-2_",
-      "escape_character": "yes",
       "auto_create_table": "yes",
       "batch_create_tbl_num": 10,
       "data_source": "rand",

tests/parallel_test/cases.task

@@ -747,6 +747,7 @@
 ,,y,script,./test.sh -f tsim/db/show_create_table.sim
 ,,y,script,./test.sh -f tsim/db/tables.sim
 ,,y,script,./test.sh -f tsim/db/taosdlog.sim
+,,y,script,./test.sh -f tsim/db/table_prefix_suffix.sim
 ,,y,script,./test.sh -f tsim/dnode/balance_replica1.sim
 ,,y,script,./test.sh -f tsim/dnode/balance_replica3.sim
 ,,y,script,./test.sh -f tsim/dnode/balance1.sim

tests/script/tsim/db/table_prefix_suffix.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sql connect
+
+sql drop database if exists db1;
+sql create database db1 vgroups 5 TABLE_PREFIX 1 TABLE_SUFFIX 2;
+sql use db1;
+sql create table atb1aa (ts timestamp, f1 int);
+sql create table btb1bb (ts timestamp, f1 int);
+sql create table ctb1cc (ts timestamp, f1 int);
+sql create table dtb1dd (ts timestamp, f1 int);
+sql create table atb2aa (ts timestamp, f1 int);
+sql create table btb2bb (ts timestamp, f1 int);
+sql create table ctb2cc (ts timestamp, f1 int);
+sql create table dtb2dd (ts timestamp, f1 int);
+sql create table etb2ee (ts timestamp, f1 int);
+sql show create database db1;
+sql select count(*) a from information_schema.ins_tables where db_name='db1' group by vgroup_id having(count(*) > 0) order by a;
+if $rows != 2 then
+  return -1
+endi
+if $data00 != 4 then
+  return -1
+endi
+if $data10 != 5 then
+  return -1
+endi
+sql drop database if exists db1;
+ 
+sql drop database if exists db2;
+sql create database db2 vgroups 5 TABLE_PREFIX -1 TABLE_SUFFIX -2;
+sql use db2;
+sql create table taaa11 (ts timestamp, f1 int);
+sql create table tbbb11 (ts timestamp, f1 int);
+sql create table tccc11 (ts timestamp, f1 int);
+sql create table tddd11 (ts timestamp, f1 int);
+sql create table taaa22 (ts timestamp, f1 int);
+sql create table tbbb22 (ts timestamp, f1 int);
+sql create table tccc22 (ts timestamp, f1 int);
+sql create table tddd22 (ts timestamp, f1 int);
+sql create table teee22 (ts timestamp, f1 int);
+sql show create database db2;
+sql select count(*) a from information_schema.ins_tables where db_name='db2' group by vgroup_id having(count(*) > 0) order by a;
+if $rows != 2 then
+  return -1
+endi
+if $data00 != 4 then
+  return -1
+endi
+if $data10 != 5 then
+  return -1
+endi
+sql drop database if exists db2;
+ 
+sql drop database if exists db3;
+sql create database db3 vgroups 5 TABLE_PREFIX -1;
+sql use db3;
+sql create table taaa11 (ts timestamp, f1 int);
+sql create table tbbb11 (ts timestamp, f1 int);
+sql create table tccc11 (ts timestamp, f1 int);
+sql create table tddd11 (ts timestamp, f1 int);
+sql create table zaaa22 (ts timestamp, f1 int);
+sql create table zbbb22 (ts timestamp, f1 int);
+sql create table zccc22 (ts timestamp, f1 int);
+sql create table zddd22 (ts timestamp, f1 int);
+sql create table zeee22 (ts timestamp, f1 int);
+sql show create database db3;
+sql select count(*) a from information_schema.ins_tables where db_name='db3' group by vgroup_id having(count(*) > 0) order by a;
+if $rows != 2 then
+  return -1
+endi
+if $data00 != 4 then
+  return -1
+endi
+if $data10 != 5 then
+  return -1
+endi
+sql drop database if exists db3;
+ 
+sql drop database if exists db4;
+sql create database db4 vgroups 5 TABLE_SUFFIX -2;
+sql use db4;
+sql create table taaa11 (ts timestamp, f1 int);
+sql create table tbbb11 (ts timestamp, f1 int);
+sql create table tccc11 (ts timestamp, f1 int);
+sql create table tddd11 (ts timestamp, f1 int);
+sql create table zaaa22 (ts timestamp, f1 int);
+sql create table zbbb22 (ts timestamp, f1 int);
+sql create table zccc22 (ts timestamp, f1 int);
+sql create table zddd22 (ts timestamp, f1 int);
+sql create table zeee22 (ts timestamp, f1 int);
+sql show create database db4;
+sql select count(*) a from information_schema.ins_tables where db_name='db4' group by vgroup_id having(count(*) > 0) order by a;
+if $rows != 2 then
+  return -1
+endi
+if $data00 != 4 then
+  return -1
+endi
+if $data10 != 5 then
+  return -1
+endi
+sql drop database if exists db4;
+ 
+sql drop database if exists db5;
+sql create database db5 vgroups 5 TABLE_PREFIX 1;
+sql use db5;
+sql create table taaa11 (ts timestamp, f1 int);
+sql create table baaa11 (ts timestamp, f1 int);
+sql create table caaa11 (ts timestamp, f1 int);
+sql create table daaa11 (ts timestamp, f1 int);
+sql create table faaa11 (ts timestamp, f1 int);
+sql create table gbbb11 (ts timestamp, f1 int);
+sql create table hbbb11 (ts timestamp, f1 int);
+sql create table ibbb11 (ts timestamp, f1 int);
+sql create table jbbb11 (ts timestamp, f1 int);
+sql show create database db5;
+sql select count(*) a from information_schema.ins_tables where db_name='db5' group by vgroup_id having(count(*) > 0) order by a;
+if $rows != 2 then
+  return -1
+endi
+if $data00 != 4 then
+  return -1
+endi
+if $data10 != 5 then
+  return -1
+endi
+sql drop database if exists db5;
+ 
+sql drop database if exists db6;
+sql create database db6 vgroups 5 TABLE_SUFFIX 2;
+sql use db6;
+sql create table taaa11 (ts timestamp, f1 int);
+sql create table taaa12 (ts timestamp, f1 int);
+sql create table taaa13 (ts timestamp, f1 int);
+sql create table taaa14 (ts timestamp, f1 int);
+sql create table tbbb23 (ts timestamp, f1 int);
+sql create table tbbb24 (ts timestamp, f1 int);
+sql create table tbbb31 (ts timestamp, f1 int);
+sql create table tbbb32 (ts timestamp, f1 int);
+sql create table tbbb33 (ts timestamp, f1 int);
+sql show create database db6;
+sql select count(*) a from information_schema.ins_tables where db_name='db6' group by vgroup_id having(count(*) > 0) order by a;
+if $rows != 2 then
+  return -1
+endi
+if $data00 != 4 then
+  return -1
+endi
+if $data10 != 5 then
+  return -1
+endi
+sql drop database if exists db6;
+ 
+sql drop database if exists db7;
+sql create database db7 vgroups 5 TABLE_PREFIX -100 TABLE_SUFFIX -92;
+sql use db7;
+sql create table taaa11 (ts timestamp, f1 int);
+sql create table taaa12 (ts timestamp, f1 int);
+sql create table taaa13 (ts timestamp, f1 int);
+sql create table tbbb21 (ts timestamp, f1 int);
+sql create table tbbb22 (ts timestamp, f1 int);
+sql create table tbbb23 (ts timestamp, f1 int);
+sql create table tbbb24 (ts timestamp, f1 int);
+sql create table tccc31 (ts timestamp, f1 int);
+sql create table tccc32 (ts timestamp, f1 int);
+sql create table tccc33 (ts timestamp, f1 int);
+sql create table tddd24 (ts timestamp, f1 int);
+sql create table tddd31 (ts timestamp, f1 int);
+sql create table tddd32 (ts timestamp, f1 int);
+sql create table tddd33 (ts timestamp, f1 int);
+sql show create database db7;
+sql select count(*) a from information_schema.ins_tables where db_name='db7' group by vgroup_id having(count(*) > 0) order by a;
+sql drop database if exists db7;
+
+sql_error create database db8 vgroups 5 TABLE_PREFIX -1 TABLE_SUFFIX 2;
+sql_error create database db8 vgroups 5 TABLE_PREFIX 191 TABLE_SUFFIX 192;
+sql_error create database db8 vgroups 5 TABLE_PREFIX -192 TABLE_SUFFIX -191;
+sql_error create database db8 vgroups 5 TABLE_PREFIX 100 TABLE_SUFFIX 92;
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/query/sys_tbname.sim

@@ -109,5 +109,26 @@ if $rows != 5000 then
         return -1
 endi
 
-
+sql create database d1;
+sql create stable d1.st1 (ts timestamp, f int) tags(t int);
+sql create stable d1.st2 (ts timestamp, f int) tags(t int);
+sql create table d1.ct1 using d1.st1 tags(1);
+sql create table d1.ct2 using d1.st2 tags(2);
+
+sql create database d2;
+sql create stable d2.st1(ts timestamp, f int) tags(t int);
+sql create stable d2.st2(ts timestamp, f int) tags(t int);
+sql create table d2.ct1 using d2.st1 tags(1);
+sql create table d2.ct2 using d2.st2 tags(2);
+
+sql create database d3;
+sql create stable d3.st1(ts timestamp, f int) tags(t int);
+sql create stable d3.st2(ts timestamp, f int) tags(t int);
+sql create table d3.ct1 using d3.st1 tags(1);
+sql create table d3.ct2 using d3.st2 tags(2);
+sql select count(*), stable_name, db_name from information_schema.ins_tables where db_name != 'd2' group by stable_name,db_name
+print $rows
+if $rows != 9 then
+  return -1
+endi
 #system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/query/tableCount.sim

@@ -104,4 +104,9 @@ if $data62 != 5 then
   return -1
 endi
 
+sql select distinct db_name from information_schema.ins_tables;
+print $rows
+if $rows != 4 then
+  return -1
+endi
 system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/stream/basic1.sim

@@ -37,7 +37,7 @@ if $loop_count == 20 then
 endi
 
 if $rows != 4 then
-  print =====rows=$rows, expect 4
+  print =====rows=$rows  expect 4
   goto loop0
 endi
 

tests/system-test/0-others/udf_create.py

@@ -47,17 +47,27 @@ class TDTestCase:
 
         if platform.system().lower() == 'windows':
             self.libudf1 = subprocess.Popen('(for /r %s %%i in ("udf1.d*") do @echo %%i)|grep lib|head -n1'%projPath , shell=True, stdout=subprocess.PIPE,stderr=subprocess.STDOUT).stdout.read().decode("utf-8")
+            self.libudf1_dup = subprocess.Popen('(for /r %s %%i in ("udf1_dup.d*") do @echo %%i)|grep lib|head -n1'%projPath , shell=True, stdout=subprocess.PIPE,stderr=subprocess.STDOUT).stdout.read().decode("utf-8")
             self.libudf2 = subprocess.Popen('(for /r %s %%i in ("udf2.d*") do @echo %%i)|grep lib|head -n1'%projPath , shell=True, stdout=subprocess.PIPE,stderr=subprocess.STDOUT).stdout.read().decode("utf-8")
+            self.libudf2_dup = subprocess.Popen('(for /r %s %%i in ("udf2_dup.d*") do @echo %%i)|grep lib|head -n1'%projPath , shell=True, stdout=subprocess.PIPE,stderr=subprocess.STDOUT).stdout.read().decode("utf-8")
             if (not tdDnodes.dnodes[0].remoteIP == ""):
                 tdDnodes.dnodes[0].remote_conn.get(tdDnodes.dnodes[0].config["path"]+'/debug/build/lib/libudf1.so',projPath+"\\debug\\build\\lib\\")
+                tdDnodes.dnodes[0].remote_conn.get(tdDnodes.dnodes[0].config["path"]+'/debug/build/lib/libudf1_dup.so',projPath+"\\debug\\build\\lib\\")
                 tdDnodes.dnodes[0].remote_conn.get(tdDnodes.dnodes[0].config["path"]+'/debug/build/lib/libudf2.so',projPath+"\\debug\\build\\lib\\")
+                tdDnodes.dnodes[0].remote_conn.get(tdDnodes.dnodes[0].config["path"]+'/debug/build/lib/libudf2_dup.so',projPath+"\\debug\\build\\lib\\")
                 self.libudf1 = self.libudf1.replace('udf1.dll','libudf1.so')
+                self.libudf1_dup = self.libudf1_dup.replace('udf1_dup.dll','libudf1_dup.so')
                 self.libudf2 = self.libudf2.replace('udf2.dll','libudf2.so')
+                self.libudf2_dup = self.libudf2_dup.replace('udf2_dup.dll','libudf2_dup.so')
         else:
             self.libudf1 = subprocess.Popen('find %s -name "libudf1.so"|grep lib|head -n1'%projPath , shell=True, stdout=subprocess.PIPE,stderr=subprocess.STDOUT).stdout.read().decode("utf-8")
+            self.libudf1_dup = subprocess.Popen('find %s -name "libudf1_dup.so"|grep lib|head -n1'%projPath , shell=True, stdout=subprocess.PIPE,stderr=subprocess.STDOUT).stdout.read().decode("utf-8")
             self.libudf2 = subprocess.Popen('find %s -name "libudf2.so"|grep lib|head -n1'%projPath , shell=True, stdout=subprocess.PIPE,stderr=subprocess.STDOUT).stdout.read().decode("utf-8")
+            self.libudf2_dup = subprocess.Popen('find %s -name "libudf2_dup.so"|grep lib|head -n1'%projPath , shell=True, stdout=subprocess.PIPE,stderr=subprocess.STDOUT).stdout.read().decode("utf-8")
         self.libudf1 = self.libudf1.replace('\r','').replace('\n','')
+        self.libudf1_dup = self.libudf1_dup.replace('\r','').replace('\n','')
         self.libudf2 = self.libudf2.replace('\r','').replace('\n','')
+        self.libudf2_dup = self.libudf2_dup.replace('\r','').replace('\n','')
 
 
     def prepare_data(self):
@@ -174,10 +184,12 @@ class TDTestCase:
 
         # create  scalar functions
         tdSql.execute("create function udf1 as '%s' outputtype int;"%self.libudf1)
+        tdSql.execute("create function udf1_dup as '%s' outputtype int;"%self.libudf1_dup)
 
         # create aggregate functions
 
         tdSql.execute("create aggregate function udf2 as '%s' outputtype double bufSize 8;"%self.libudf2)
+        tdSql.execute("create aggregate function udf2_dup as '%s' outputtype double bufSize 8;"%self.libudf2_dup)
 
         functions = tdSql.getResult("show functions")
         function_nums = len(functions)
@@ -188,6 +200,13 @@ class TDTestCase:
 
         # scalar functions
 
+        # udf1_dup
+        tdSql.query("select udf1(num1) ,udf1_dup(num1) from tb")
+        tdSql.checkData(1,0,1)
+        tdSql.checkData(1,1,2)
+        tdSql.checkData(2,0,1)
+        tdSql.checkData(2,1,2)
+
         tdSql.execute("use db ")
         tdSql.query("select num1 , udf1(num1) ,num2 ,udf1(num2),num3 ,udf1(num3),num4 ,udf1(num4) from tb")
         tdSql.checkData(0,0,None)
@@ -238,6 +257,10 @@ class TDTestCase:
 
 
         # aggregate functions
+        tdSql.query("select udf2(num1) ,udf2_dup(num1) from tb")
+        val = tdSql.queryResult[0][0] + 100
+        tdSql.checkData(0,1,val)
+        
         tdSql.query("select udf2(num1) ,udf2(num2), udf2(num3) from tb")
         tdSql.checkData(0,0,15.362291496)
         tdSql.checkData(0,1,10000949.553189287)

tests/system-test/7-tmq/stbTagFilter-1ctb.py

@@ -111,7 +111,7 @@ class TDTestCase:
         topicFromStb1 = 'topic_UpperCase_stb1'
         # queryString = "select ts, c1, c2 from %s.%s where t4 == 'shanghai' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
         queryString = "select ts, c1, c2, t4 from %s.%s where t4 == 'shanghai' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
-        sqlString = "create topic %s as %s" %(topicFromStb1, queryString)
+        sqlString = "create topic `%s` as %s" %(topicFromStb1, queryString)
         tdLog.info("create topic sql: %s"%sqlString)
         tdSql.execute(sqlString)
 
@@ -148,7 +148,7 @@ class TDTestCase:
 
         tmqCom.checkFileContent(consumerId, queryString)
 
-        tdSql.query("drop topic %s"%topicFromStb1)
+        tdSql.query("drop topic `%s`"%topicFromStb1)
         tdLog.printNoPrefix("======== test case 1 end ...... ")
 
     def tmqCase2(self):
@@ -196,7 +196,7 @@ class TDTestCase:
         topicFromStb1 = 'topic_UpperCase_stb1'
         queryString = "select ts, c1, c2 from %s.%s where t4 == 'shanghai' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
         # queryString = "select ts, c1, c2, t4 from %s.%s where t4 == 'shanghai' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
-        sqlString = "create topic %s as %s" %(topicFromStb1, queryString)
+        sqlString = "create topic `%s` as %s" %(topicFromStb1, queryString)
         tdLog.info("create topic sql: %s"%sqlString)
         tdSql.execute(sqlString)
 
@@ -242,7 +242,7 @@ class TDTestCase:
 
         # tmqCom.checkFileContent(consumerId, queryString)
 
-        tdSql.query("drop topic %s"%topicFromStb1)
+        tdSql.query("drop topic `%s`"%topicFromStb1)
 
         tdLog.printNoPrefix("======== test case 2 end ...... ")
 

tests/system-test/7-tmq/stbTagFilter-multiCtb.py

@@ -111,7 +111,7 @@ class TDTestCase:
         topicFromStb1 = 'topic_UpperCase_stb1'
         queryString = "select ts, c1, c2 from %s.%s where t4 == 'beijing' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
         # queryString = "select ts, c1, c2, t4 from %s.%s where t4 == 'beijing' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
-        sqlString = "create topic %s as %s" %(topicFromStb1, queryString)
+        sqlString = "create topic `%s` as %s" %(topicFromStb1, queryString)
         tdLog.info("create topic sql: %s"%sqlString)
         tdSql.execute(sqlString)
 
@@ -148,7 +148,7 @@ class TDTestCase:
 
         # tmqCom.checkFileContent(consumerId, queryString)
 
-        tdSql.query("drop topic %s"%topicFromStb1)
+        tdSql.query("drop topic `%s`"%topicFromStb1)
         tdLog.printNoPrefix("======== test case 1 end ...... ")
 
     def tmqCase2(self):
@@ -196,7 +196,7 @@ class TDTestCase:
         topicFromStb1 = 'topic_UpperCase_stb1'
         # queryString = "select ts, c1, c2 from %s.%s where t4 == 'beijing' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
         queryString = "select ts, c1, c2, t4 from %s.%s where t4 == 'beijing' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
-        sqlString = "create topic %s as %s" %(topicFromStb1, queryString)
+        sqlString = "create topic `%s` as %s" %(topicFromStb1, queryString)
         tdLog.info("create topic sql: %s"%sqlString)
         tdSql.execute(sqlString)
 
@@ -244,7 +244,7 @@ class TDTestCase:
 
         # tmqCom.checkFileContent(consumerId, queryString)
 
-        tdSql.query("drop topic %s"%topicFromStb1)
+        tdSql.query("drop topic `%s`"%topicFromStb1)
 
         tdLog.printNoPrefix("======== test case 2 end ...... ")
 

tests/system-test/7-tmq/subscribeDb3.py

@@ -82,7 +82,7 @@ class TDTestCase:
             tdSql.query("select * from %s.notifyinfo"%cdbName)
             #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
             if tdSql.getRows() == 2 :
-                print(tdSql.getData(0, 1), tdSql.getData(1, 1))
+                tdLog.info("row[0][1]: %d, row[1][1]: %d"%(tdSql.getData(0, 1), tdSql.getData(1, 1)))
                 if tdSql.getData(1, 1) == 1:
                     break
             time.sleep(0.1)
@@ -122,6 +122,7 @@ class TDTestCase:
         os.system(shellCmd)
 
     def create_tables(self,tsql, dbName,vgroups,stbName,ctbNum,rowsPerTbl):
+        tdLog.info("start create tables......")
         tsql.execute("create database if not exists %s vgroups %d wal_retention_period 3600"%(dbName, vgroups))
         tsql.execute("use %s" %dbName)
         tsql.execute("create table  if not exists %s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"%stbName)
@@ -137,11 +138,11 @@ class TDTestCase:
             tsql.execute(sql)
 
         event.set()
-        tdLog.debug("complete to create database[%s], stable[%s] and %d child tables" %(dbName, stbName, ctbNum))
+        tdLog.info("complete to create database[%s], stable[%s] and %d child tables" %(dbName, stbName, ctbNum))
         return
 
     def insert_data(self,tsql,dbName,stbName,ctbNum,rowsPerTbl,batchNum,startTs):
-        tdLog.debug("start to insert data ............")
+        tdLog.info("start to insert data ............")
         tsql.execute("use %s" %dbName)
         pre_insert = "insert into "
         sql = pre_insert
@@ -163,7 +164,7 @@ class TDTestCase:
         if sql != pre_insert:
             #print("insert sql:%s"%sql)
             tsql.execute(sql)
-        tdLog.debug("insert data ............ [OK]")
+        tdLog.info("insert data ............ [OK]")
         return
 
     def prepareEnv(self, **parameterDict):
@@ -286,7 +287,7 @@ class TDTestCase:
         prepareEnvThread.start()
 
         tdLog.info("create topics from db")
-        topicName1 = 'topic_db1'
+        topicName1 = 'topic_db11'
 
         tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0

tests/system-test/7-tmq/tmqDropStb.py

@@ -82,7 +82,7 @@ class TDTestCase:
 
         tdLog.info("create topics from db")
         topicName1 = 'UpperCasetopic_%s'%(self.paraDict['dbName'])
-        tdSql.execute("create topic %s as database %s" %(topicName1, self.paraDict['dbName']))
+        tdSql.execute("create topic `%s` as database %s" %(topicName1, self.paraDict['dbName']))
 
         topicList = topicName1 + ',' +topicName1
         keyList = '%s,%s,%s,%s'%(self.groupId,self.autoCommit,self.autoCommitInterval,self.autoOffset)
@@ -113,7 +113,7 @@ class TDTestCase:
             tdLog.exit("tmq consume rows error!")
 
         time.sleep(10)
-        tdSql.query("drop topic %s"%topicName1)
+        tdSql.query("drop topic `%s`"%topicName1)
 
         tdLog.printNoPrefix("======== test case 1 end ...... ")
 

utils/test/c/sml_test.c

@@ -895,6 +895,63 @@ int smlProcess_18784_Test() {
   return code;
 }
 
+int sml_escape_Test() {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+
+  TAOS_RES *pRes = taos_query(taos, "create database if not exists db_escape");
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "use db_escape");
+  taos_free_result(pRes);
+
+  const char *sql[] = {
+      "d\\,i=\\ s\\k\",dev\"i\\,\\=\\ ce=s\"i\\,\\=\\ dc inode\"i\\,\\=\\ s_used=176059i,total=1076048383523889174i 1661943960000000000",
+      "d\\,i=\\ s\\k\",dev\"i\\,\\=\\ ce=s\"i\\,\\=\\ dc inode\"i\\,\\=\\ s_f\\\\ree=\"\\\"id,= ei\\\\\\f\" 1661943960000000000",
+  };
+  pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_LINE_PROTOCOL, 0);
+  printf("%s result:%s, rows:%d\n", __FUNCTION__, taos_errstr(pRes), taos_affected_rows(pRes));
+  int code = taos_errno(pRes);
+  ASSERT(!code);
+  ASSERT(taos_affected_rows(pRes) == 1);
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "select * from `d,i= s\\k\"`");     //check stable name
+  ASSERT(pRes);
+  int fieldNum = taos_field_count(pRes);
+  ASSERT(fieldNum == 5);
+  printf("fieldNum:%d\n", fieldNum);
+
+  int         numFields = taos_num_fields(pRes);
+  TAOS_FIELD *fields = taos_fetch_fields(pRes);
+  ASSERT(numFields == 5);
+  ASSERT(strncmp(fields[1].name, "inode\"i,= s_used", sizeof("inode\"i,= s_used") - 1) == 0);
+  ASSERT(strncmp(fields[2].name, "total", sizeof("total") - 1) == 0);
+  ASSERT(strncmp(fields[3].name, "inode\"i,= s_f\\\\ree", sizeof("inode\"i,= s_f\\\\ree") - 1) == 0);
+  ASSERT(strncmp(fields[4].name, "dev\"i,= ce", sizeof("dev\"i,= ce") - 1) == 0);
+
+  TAOS_ROW row = NULL;
+  int32_t  rowIndex = 0;
+  while ((row = taos_fetch_row(pRes)) != NULL) {
+    int64_t ts = *(int64_t *)row[0];
+    int64_t used = *(int64_t *)row[1];
+    int64_t total = *(int64_t *)row[2];
+
+    if (rowIndex == 0) {
+      ASSERT(ts == 1661943960000);
+      ASSERT(used == 176059);
+      ASSERT(total == 1076048383523889174);
+      ASSERT(strncmp(row[3], "\"id,= ei\\\\f", sizeof("\"id,= ei\\\\f") - 1) == 0);
+      ASSERT(strncmp(row[4], "s\"i,= dc", sizeof("s\"i,= dc") - 1) == 0);
+
+    }
+    rowIndex++;
+  }
+  taos_free_result(pRes);
+  taos_close(taos);
+
+  return code;
+}
+
 int sml_19221_Test() {
   TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
 
@@ -961,6 +1018,55 @@ int sml_ts2164_Test() {
   return code;
 }
 
+
+int sml_ts3116_Test() {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+
+  TAOS_RES *pRes =
+      taos_query(taos, "DROP DATABASE IF EXISTS ts3116");
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "CREATE DATABASE IF NOT EXISTS ts3116  BUFFER 384  MINROWS 1000  PAGES 256 PRECISION 'ns'");
+  taos_free_result(pRes);
+
+  char *sql = {
+      "meters,location=la,groupid=ca current=11.8,voltage=221",
+  };
+
+  pRes = taos_query(taos, "use ts3116");
+  taos_free_result(pRes);
+  int32_t totalRows = 0;
+  char *tmp = (char *)taosMemoryCalloc(1024, 1);
+  memcpy(tmp, sql, strlen(sql));
+  totalRows = 0;
+  pRes = taos_schemaless_insert_raw(taos, tmp, strlen(tmp), &totalRows, TSDB_SML_LINE_PROTOCOL,
+                                    TSDB_SML_TIMESTAMP_MILLI_SECONDS);
+  taosMemoryFree(tmp);
+  printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
+  int code = taos_errno(pRes);
+  taos_free_result(pRes);
+
+  char *sql1 = {
+      "meters,location=la,groupid=ca\\=3 current=11.8,voltage=221\nmeters,location=la,groupid=ca current=11.8,voltage=221,phase=0.27",
+  };
+
+  pRes = taos_query(taos, "use ts3116");
+  taos_free_result(pRes);
+
+  tmp = (char *)taosMemoryCalloc(1024, 1);
+  memcpy(tmp, sql1, strlen(sql1));
+  totalRows = 0;
+  pRes = taos_schemaless_insert_raw(taos, tmp, strlen(tmp), &totalRows, TSDB_SML_LINE_PROTOCOL,
+                                TSDB_SML_TIMESTAMP_MILLI_SECONDS);
+  taosMemoryFree(tmp);
+  printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
+  code = taos_errno(pRes);
+  taos_free_result(pRes);
+  taos_close(taos);
+
+  return code;
+}
+
 int sml_td22898_Test() {
   TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
 
@@ -1195,6 +1301,10 @@ int main(int argc, char *argv[]) {
   }
 
   int ret = 0;
+  ret = sml_escape_Test();
+  ASSERT(!ret);
+  ret = sml_ts3116_Test();
+  ASSERT(!ret);
   ret = sml_ts2385_Test();    // this test case need config sml table name using ./sml_test config_file
   ASSERT(!ret);
   //  for(int i = 0; i < sizeof(str)/sizeof(str[0]); i++){