fix: tq mem leak

cmake/cmake.define

@@ -46,11 +46,17 @@ ENDIF ()
 
 IF (TD_WINDOWS)
     MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}")
-    SET(COMMON_FLAGS "/W3 /D_WIN32")
+    SET(COMMON_FLAGS "/w /D_WIN32")
     SET(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} /MANIFEST:NO")
     # IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900))
     #  SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18")
     # ENDIF ()
+    IF (CMAKE_DEPFILE_FLAGS_C)
+        SET(CMAKE_DEPFILE_FLAGS_C "")
+    ENDIF ()
+    IF (CMAKE_DEPFILE_FLAGS_CXX)
+        SET(CMAKE_DEPFILE_FLAGS_CXX "")
+    ENDIF ()
 
     SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${COMMON_FLAGS}")
     SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${COMMON_FLAGS}")

cmake/cmake.options

@@ -46,6 +46,18 @@ IF(${TD_WINDOWS})
             ON
         )
 
+    option(
+        BUILD_TEST 
+        "If build unit tests using googletest" 
+        OFF
+    )
+ELSE ()
+
+    option(
+        BUILD_TEST 
+        "If build unit tests using googletest" 
+        ON
+    )
 ENDIF ()
 
 option(
@@ -54,12 +66,6 @@ option(
     OFF
     )
 
-option(
-    BUILD_TEST 
-    "If build unit tests using googletest" 
-    ON
-)
-
 option(
     BUILD_WITH_LEVELDB 
     "If build with leveldb" 

include/libs/nodes/querynodes.h

@@ -351,9 +351,6 @@ bool nodesIsComparisonOp(const SOperatorNode* pOp);
 bool nodesIsJsonOp(const SOperatorNode* pOp);
 bool nodesIsRegularOp(const SOperatorNode* pOp);
 
-bool nodesIsTimeorderQuery(const SNode* pQuery);
-bool nodesIsTimelineQuery(const SNode* pQuery);
-
 void*   nodesGetValueFromNode(SValueNode* pNode);
 int32_t nodesSetValueNodeValue(SValueNode* pNode, void* value);
 char*   nodesGetStrValueFromNode(SValueNode* pNode);

include/util/taoserror.h

@@ -650,6 +650,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_FUNCTION_NAME     TAOS_DEF_ERROR_CODE(0, 0x264D)
 #define TSDB_CODE_PAR_COMMENT_TOO_LONG          TAOS_DEF_ERROR_CODE(0, 0x264E)
 #define TSDB_CODE_PAR_NOT_ALLOWED_FUNC          TAOS_DEF_ERROR_CODE(0, 0x264F)
+#define TSDB_CODE_PAR_NOT_ALLOWED_WIN_QUERY     TAOS_DEF_ERROR_CODE(0, 0x2650)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

source/client/src/clientImpl.c

@@ -23,6 +23,8 @@
 #include "tmsgtype.h"
 #include "tpagedbuf.h"
 #include "tref.h"
+#include "cJSON.h"
+#include "tdataformat.h"
 
 static int32_t       initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet);
 static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest);
@@ -268,7 +270,7 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
 
     if (pSchema[i].type == TSDB_DATA_TYPE_VARCHAR) {
       pResInfo->userFields[i].bytes -= VARSTR_HEADER_SIZE;
-    } else if (pSchema[i].type == TSDB_DATA_TYPE_NCHAR) {
+    } else if (pSchema[i].type == TSDB_DATA_TYPE_NCHAR || pSchema[i].type == TSDB_DATA_TYPE_JSON) {
       pResInfo->userFields[i].bytes = (pResInfo->userFields[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
     }
 
@@ -803,6 +805,101 @@ static int32_t doPrepareResPtr(SReqResultInfo* pResInfo) {
   return TSDB_CODE_SUCCESS;
 }
 
+static char* parseTagDatatoJson(void *p){
+  char* string = NULL;
+  cJSON *json = cJSON_CreateObject();
+  if (json == NULL)
+  {
+    goto end;
+  }
+
+  int16_t nCols = kvRowNCols(p);
+  char tagJsonKey[256] = {0};
+  for (int j = 0; j < nCols; ++j) {
+    SColIdx * pColIdx = kvRowColIdxAt(p, j);
+    char* val = (char*)(kvRowColVal(p, pColIdx));
+    if (j == 0){
+      if(*val == TSDB_DATA_TYPE_NULL){
+        string = taosMemoryCalloc(1, 8);
+        sprintf(varDataVal(string), "%s", TSDB_DATA_NULL_STR_L);
+        varDataSetLen(string, strlen(varDataVal(string)));
+        goto end;
+      }
+      continue;
+    }
+
+    // json key  encode by binary
+    memset(tagJsonKey, 0, sizeof(tagJsonKey));
+    memcpy(tagJsonKey, varDataVal(val), varDataLen(val));
+    // json value
+    val += varDataTLen(val);
+    char* realData = POINTER_SHIFT(val, CHAR_BYTES);
+    char type = *val;
+    if(type == TSDB_DATA_TYPE_NULL) {
+      cJSON* value = cJSON_CreateNull();
+      if (value == NULL)
+      {
+        goto end;
+      }
+      cJSON_AddItemToObject(json, tagJsonKey, value);
+    }else if(type == TSDB_DATA_TYPE_NCHAR) {
+      cJSON* value = NULL;
+      if (varDataLen(realData) > 0){
+        char *tagJsonValue = taosMemoryCalloc(varDataLen(realData), 1);
+        int32_t length = taosUcs4ToMbs((TdUcs4 *)varDataVal(realData), varDataLen(realData), tagJsonValue);
+        if (length < 0) {
+          tscError("charset:%s to %s. val:%s convert json value failed.", DEFAULT_UNICODE_ENCODEC, tsCharset, val);
+          taosMemoryFree(tagJsonValue);
+          goto end;
+        }
+        value = cJSON_CreateString(tagJsonValue);
+        taosMemoryFree(tagJsonValue);
+        if (value == NULL)
+        {
+          goto end;
+        }
+      }else if(varDataLen(realData) == 0){
+        value = cJSON_CreateString("");
+      }else{
+        ASSERT(0);
+      }
+
+      cJSON_AddItemToObject(json, tagJsonKey, value);
+    }else if(type == TSDB_DATA_TYPE_DOUBLE){
+      double jsonVd = *(double*)(realData);
+      cJSON* value = cJSON_CreateNumber(jsonVd);
+      if (value == NULL)
+      {
+        goto end;
+      }
+      cJSON_AddItemToObject(json, tagJsonKey, value);
+//    }else if(type == TSDB_DATA_TYPE_BIGINT){
+//      int64_t jsonVd = *(int64_t*)(realData);
+//      cJSON* value = cJSON_CreateNumber((double)jsonVd);
+//      if (value == NULL)
+//      {
+//        goto end;
+//      }
+//      cJSON_AddItemToObject(json, tagJsonKey, value);
+    }else if (type == TSDB_DATA_TYPE_BOOL) {
+      char jsonVd = *(char*)(realData);
+      cJSON* value = cJSON_CreateBool(jsonVd);
+      if (value == NULL)
+      {
+        goto end;
+      }
+      cJSON_AddItemToObject(json, tagJsonKey, value);
+    }else{
+      ASSERT(0);
+    }
+
+  }
+  string = cJSON_PrintUnformatted(json);
+end:
+  cJSON_Delete(json);
+  return string;
+}
+
 static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int32_t numOfCols, int32_t* colLength) {
   for (int32_t i = 0; i < numOfCols; ++i) {
     int32_t type = pResultInfo->fields[i].type;
@@ -833,9 +930,7 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
 
       pResultInfo->pCol[i].pData = pResultInfo->convertBuf[i];
       pResultInfo->row[i] = pResultInfo->pCol[i].pData;
-    }
-
-    if (type == TSDB_DATA_TYPE_JSON) {
+    }else if (type == TSDB_DATA_TYPE_JSON && colLength[i] > 0) {
       char* p = taosMemoryRealloc(pResultInfo->convertBuf[i], colLength[i]);
       if (p == NULL) {
         return TSDB_CODE_OUT_OF_MEMORY;
@@ -848,6 +943,7 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
         if (pCol->offset[j] != -1) {
           char* pStart = pCol->offset[j] + pCol->pData;
 
+
           int32_t jsonInnerType = *pStart;
           char*   jsonInnerData = pStart + CHAR_BYTES;
           char    dst[TSDB_MAX_JSON_TAG_LEN] = {0};
@@ -855,15 +951,9 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
             sprintf(varDataVal(dst), "%s", TSDB_DATA_NULL_STR_L);
             varDataSetLen(dst, strlen(varDataVal(dst)));
           } else if (jsonInnerType == TSDB_DATA_TYPE_JSON) {
-            int32_t length =
-                taosUcs4ToMbs((TdUcs4*)varDataVal(jsonInnerData), varDataLen(jsonInnerData), varDataVal(dst));
-
-            if (length <= 0) {
-              tscError("charset:%s to %s. val:%s convert failed.", DEFAULT_UNICODE_ENCODEC, tsCharset,
-                       varDataVal(jsonInnerData));
-              length = 0;
-            }
-            varDataSetLen(dst, length);
+            char *jsonString = parseTagDatatoJson(jsonInnerData);
+            STR_TO_VARSTR(dst, jsonString);
+            taosMemoryFree(jsonString);
           } else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) {  // value -> "value"
             *(char*)varDataVal(dst) = '\"';
             int32_t length = taosUcs4ToMbs((TdUcs4*)varDataVal(jsonInnerData), varDataLen(jsonInnerData),

source/client/src/clientSml.c

@@ -67,8 +67,10 @@ for (int i = 1; i < keyLen; ++i) {      \
 
 #define TS              "_ts"
 #define TS_LEN          3
-#define TAG       "_tagNone"
-#define TAG_LEN   8
+#define TAG             "_tag"
+#define TAG_LEN         4
+#define TAG_VALUE       "NULL"
+#define TAG_VALUE_LEN   4
 #define VALUE           "value"
 #define VALUE_LEN       5
 
@@ -598,25 +600,33 @@ static bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg){
     kvVal->type = TSDB_DATA_TYPE_FLOAT;
     kvVal->f = (float)result;
   }else if ((left == 1 && *endptr == 'i') || (left == 3 && strncasecmp(endptr, "i64", left) == 0)){
-    if(result >= (double)INT64_MAX){
-      kvVal->i = INT64_MAX;
-    }else if(result <= (double)INT64_MIN){
-      kvVal->i = INT64_MIN;
-    }else{
-      kvVal->i = result;
+    if(smlDoubleToInt64OverFlow(result)){
+      errno = 0;
+      int64_t tmp = taosStr2Int64(pVal, &endptr, 10);
+      if(errno == ERANGE){
+        smlBuildInvalidDataMsg(msg, "big int out of range[-9223372036854775808,9223372036854775807]", pVal);
+        return false;
+      }
+      kvVal->type = TSDB_DATA_TYPE_BIGINT;
+      kvVal->i = tmp;
+      return true;
     }
     kvVal->type = TSDB_DATA_TYPE_BIGINT;
+    kvVal->i = (int64_t)result;
   }else if ((left == 3 && strncasecmp(endptr, "u64", left) == 0)){
-    if(result < 0){
-      smlBuildInvalidDataMsg(msg, "unsigned big int is too large, out of precision", pVal);
+    if(result >= (double)UINT64_MAX || result < 0){
+      errno = 0;
+      uint64_t tmp = taosStr2UInt64(pVal, &endptr, 10);
+      if(errno == ERANGE || result < 0){
+        smlBuildInvalidDataMsg(msg, "unsigned big int out of range[0,18446744073709551615]", pVal);
         return false;
       }
-    if(result >= (double)UINT64_MAX){
-      kvVal->u = UINT64_MAX;
-    }else{
-      kvVal->u = result;
+      kvVal->type = TSDB_DATA_TYPE_UBIGINT;
+      kvVal->u = tmp;
+      return true;
     }
     kvVal->type = TSDB_DATA_TYPE_UBIGINT;
+    kvVal->u = result;
   }else if (left == 3 && strncasecmp(endptr, "i32", left) == 0){
     if(!IS_VALID_INT(result)){
       smlBuildInvalidDataMsg(msg, "int out of range[-2147483648,2147483647]", pVal);
@@ -1103,8 +1113,7 @@ static int32_t smlParseTelnetString(SSmlHandle *info, const char* sql, SSmlTable
   kv->keyLen = VALUE_LEN;
   kv->value = value;
   kv->length = valueLen;
-  if(!smlParseValue(kv, &info->msgBuf) || kv->type == TSDB_DATA_TYPE_BINARY
-      || kv->type == TSDB_DATA_TYPE_NCHAR || kv->type == TSDB_DATA_TYPE_BOOL){
+  if(!smlParseValue(kv, &info->msgBuf)){
     return TSDB_CODE_SML_INVALID_DATA;
   }
 
@@ -1124,8 +1133,8 @@ static int32_t smlParseCols(const char* data, int32_t len, SArray *cols, char *c
     if(!kv) return TSDB_CODE_OUT_OF_MEMORY;
     kv->key = TAG;
     kv->keyLen = TAG_LEN;
-    kv->value = TAG;
-    kv->length = TAG_LEN;
+    kv->value = TAG_VALUE;
+    kv->length = TAG_VALUE_LEN;
     kv->type = TSDB_DATA_TYPE_NCHAR;
     if(cols) taosArrayPush(cols, &kv);
     return TSDB_CODE_SUCCESS;
@@ -2264,6 +2273,7 @@ static int32_t smlParseLine(SSmlHandle *info, char* lines[], int numLines){
       uError("SML:0x%" PRIx64 " smlParseJSON failed:%s", info->id, *lines);
       return code;
     }
+    return code;
   }
 
   for (int32_t i = 0; i < numLines; ++i) {

source/client/test/smlTest.cpp

@@ -208,6 +208,7 @@ TEST(testCase, smlParseCols_Error_Test) {
     memcpy(sql, data[i], len + 1);
     SArray *cols = taosArrayInit(8, POINTER_BYTES);
     int32_t ret = smlParseCols(sql, len, cols, NULL, false, dumplicateKey, &msgBuf);
+    printf("i:%d\n",i);
     ASSERT_NE(ret, TSDB_CODE_SUCCESS);
     taosHashClear(dumplicateKey);
     taosMemoryFree(sql);
@@ -272,11 +273,11 @@ TEST(testCase, smlParseCols_tag_Test) {
 
   // nchar
   kv = (SSmlKv *)taosArrayGetP(cols, 0);
-  ASSERT_EQ(strncasecmp(kv->key, TAG, strlen(TAG)), 0);
-  ASSERT_EQ(kv->keyLen, strlen(TAG));
+  ASSERT_EQ(strncasecmp(kv->key, TAG, TAG_LEN), 0);
+  ASSERT_EQ(kv->keyLen, TAG_LEN);
   ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
-  ASSERT_EQ(kv->length, strlen(TAG));
-  ASSERT_EQ(strncasecmp(kv->value, TAG, strlen(TAG)), 0);
+  ASSERT_EQ(kv->length, TAG_LEN);
+  ASSERT_EQ(strncasecmp(kv->value, TAG_VALUE, TAG_VALUE_LEN), 0);
   taosMemoryFree(kv);
 
   taosArrayDestroy(cols);
@@ -506,7 +507,7 @@ TEST(testCase, smlProcess_influx_Test) {
     "readings,name=truck_0,fleet=South,driver=Trish,model=H-2,device_version=v2.3 load_capacity=1500,fuel_capacity=150,nominal_fuel_consumption=12,latitude=52.31854,longitude=4.72037,elevation=124,heading=221,grade=0,fuel_consumption=25 1451608403000000000",
     "readings,name=truck_0,fleet=South,driver=Trish,model=H-2,device_version=v2.3 fuel_capacity=150,nominal_fuel_consumption=12,latitude=52.31854,longitude=4.72037,elevation=124,velocity=0,heading=221,grade=0,fuel_consumption=25 1451609404000000000",
     "readings,name=truck_0,fleet=South,driver=Trish,model=H-2,device_version=v2.3 fuel_consumption=25,grade=0 1451619405000000000",
-    "readings,name=truck_1,fleet=South,driver=Albert,model=F-150,device_version=v1.5 load_capacity=2000,fuel_capacity=200,nominal_fuel_consumption=15,latitude=72.45258,longitude=68.83761,elevation=255,velocity=0,heading=181,grade=0,fuel_consumption=25 145160640600000000",
+    "readings,name=truck_1,fleet=South,driver=Albert,model=F-150,device_version=v1.5 load_capacity=2000,fuel_capacity=200,nominal_fuel_consumption=15,latitude=72.45258,longitude=68.83761,elevation=255,velocity=0,heading=181,grade=0,fuel_consumption=25 1451606406000000000",
     "readings,name=truck_2,driver=Derek,model=F-150,device_version=v1.5 load_capacity=2000,fuel_capacity=200,nominal_fuel_consumption=15,latitude=24.5208,longitude=28.09377,elevation=428,velocity=0,heading=304,grade=0,fuel_consumption=25 1451606407000000000",
     "readings,name=truck_2,fleet=North,driver=Derek,model=F-150 load_capacity=2000,fuel_capacity=200,nominal_fuel_consumption=15,latitude=24.5208,longitude=28.09377,elevation=428,velocity=0,heading=304,grade=0,fuel_consumption=25 1451609408000000000",
     "readings,fleet=South,name=truck_0,driver=Trish,model=H-2,device_version=v2.3 fuel_consumption=25,grade=0 1451629409000000000",
@@ -745,7 +746,7 @@ TEST(testCase, smlProcess_json1_Test) {
      "        }\n"
      "    }\n"
      "]";
-  int ret = smlProcess(info, (char **)(&sql), -1);
+  int ret = smlProcess(info, (char **)(&sql), 1);
   ASSERT_EQ(ret, 0);
 
   // case 1
@@ -1221,3 +1222,55 @@ TEST(testCase, sml_TD15662_Test) {
 
   taos_free_result(res);
 }
+
+TEST(testCase, sml_TD15735_Test) {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(taos, nullptr);
+
+  TAOS_RES* pRes = taos_query(taos, "create database if not exists sml_db");
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "use sml_db");
+  taos_free_result(pRes);
+
+  SRequestObj *request = (SRequestObj *)createRequest((STscObj*)taos, NULL, NULL, TSDB_SQL_INSERT);
+  ASSERT_NE(request, nullptr);
+
+  SSmlHandle *info = smlBuildSmlInfo(taos, request, TSDB_SML_TELNET_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS);
+  ASSERT_NE(info, nullptr);
+
+  const char *sql[1] = {
+      "{'metric': 'pekoiw', 'timestamp': {'value': 1626006833639000000, 'type': 'ns'}, 'value': {'value': False, 'type': 'bool'}, 'tags': {'t0': {'value': True, 'type': 'bool'}, 't1': {'value': 127, 'type': 'tinyint'}, 't2': {'value': 32767, 'type': 'smallint'}, 't3': {'value': 2147483647, 'type': 'int'}, 't4': {'value': 9223372036854775807, 'type': 'bigint'}, 't5': {'value': 11.12345027923584, 'type': 'float'}, 't6': {'value': 22.123456789, 'type': 'double'}, 't7': {'value': 'binaryTagValue', 'type': 'binary'}, 't8': {'value': 'ncharTagValue', 'type': 'nchar'}}}",
+  };
+  int32_t ret = smlProcess(info, (char**)sql, sizeof(sql)/sizeof(sql[0]));
+  ASSERT_NE(ret, 0);
+
+  destroyRequest(request);
+  smlDestroyInfo(info);
+}
+
+TEST(testCase, sml_TD15742_Test) {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(taos, nullptr);
+
+  TAOS_RES* pRes = taos_query(taos, "create database if not exists TD15742");
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "use TD15742");
+  taos_free_result(pRes);
+
+  SRequestObj *request = (SRequestObj *)createRequest((STscObj*)taos, NULL, NULL, TSDB_SQL_INSERT);
+  ASSERT_NE(request, nullptr);
+
+  SSmlHandle *info = smlBuildSmlInfo(taos, request, TSDB_SML_TELNET_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS);
+  ASSERT_NE(info, nullptr);
+
+  const char *sql[] = {
+      "zgzbix 1626006833641 False id=zgzbix_992_38861 t0=t t1=127i8 t2=32767i16 t3=2147483647i32 t4=9223372036854775807i64 t5=11.12345f32 t6=22.123456789f64 t7=\"binaryTagValue\" t8=L\"ncharTagValue\"",
+  };
+  int ret = smlProcess(info, (char**)sql, sizeof(sql)/sizeof(sql[0]));
+  ASSERT_EQ(ret, 0);
+
+  destroyRequest(request);
+  smlDestroyInfo(info);
+}
\ No newline at end of file

source/common/src/tdatablock.c

@@ -122,10 +122,14 @@ int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, con
         dataLen = 0;
       } else if (*pData == TSDB_DATA_TYPE_NCHAR) {
         dataLen = varDataTLen(pData + CHAR_BYTES);
-      } else if (*pData == TSDB_DATA_TYPE_BIGINT || *pData == TSDB_DATA_TYPE_DOUBLE) {
-        dataLen = LONG_BYTES;
+      } else if (*pData == TSDB_DATA_TYPE_DOUBLE) {
+        dataLen = DOUBLE_BYTES;
       } else if (*pData == TSDB_DATA_TYPE_BOOL) {
         dataLen = CHAR_BYTES;
+      } else if (*pData == TSDB_DATA_TYPE_JSON) {
+        dataLen = kvRowLen(pData + CHAR_BYTES);
+      } else {
+        ASSERT(0);
       }
       dataLen += CHAR_BYTES;
     }

source/common/src/tglobal.c

@@ -40,11 +40,11 @@ bool    tsPrintAuth = false;
 
 // multi process
 int32_t tsMultiProcess = 0;
-int32_t tsMnodeShmSize = TSDB_MAX_WAL_SIZE * 2 + 128;
-int32_t tsVnodeShmSize = TSDB_MAX_WAL_SIZE * 10 + 128;
-int32_t tsQnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 128;
-int32_t tsSnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 128;
-int32_t tsBnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 128;
+int32_t tsMnodeShmSize = TSDB_MAX_WAL_SIZE * 2 + 1024;
+int32_t tsVnodeShmSize = TSDB_MAX_WAL_SIZE * 10 + 1024;
+int32_t tsQnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 1024;
+int32_t tsSnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 1024;
+int32_t tsBnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 1024;
 int32_t tsNumOfShmThreads = 1;
 
 // queue & threads
@@ -380,11 +380,11 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddBool(pCfg, "deadLockKillQuery", tsDeadLockKillQuery, 0) != 0) return -1;
 
   if (cfgAddInt32(pCfg, "multiProcess", tsMultiProcess, 0, 2, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "mnodeShmSize", tsMnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "vnodeShmSize", tsVnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "qnodeShmSize", tsQnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "snodeShmSize", tsSnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "bnodeShmSize", tsBnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "mnodeShmSize", tsMnodeShmSize, TSDB_MAX_WAL_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "vnodeShmSize", tsVnodeShmSize, TSDB_MAX_WAL_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "qnodeShmSize", tsQnodeShmSize, TSDB_MAX_WAL_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "snodeShmSize", tsSnodeShmSize, TSDB_MAX_WAL_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "bnodeShmSize", tsBnodeShmSize, TSDB_MAX_WAL_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "mumOfShmThreads", tsNumOfShmThreads, 1, 1024, 0) != 0) return -1;
 
   tsNumOfRpcThreads = tsNumOfCores / 2;

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

@@ -22,21 +22,19 @@
 static inline void vmSendRsp(SRpcMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {
       .code = code,
-      .info = pMsg->info,
       .pCont = pMsg->info.rsp,
       .contLen = pMsg->info.rspLen,
+      .info = pMsg->info,
   };
   tmsgSendRsp(&rsp);
 }
 
 static void vmProcessMgmtMonitorQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   SVnodeMgmt *pMgmt = pInfo->ahandle;
-
   int32_t     code = -1;
-  tmsg_t  msgType = pMsg->msgType;
-  dTrace("msg:%p, get from vnode queue, type:%s", pMsg, TMSG_INFO(msgType));
+  dTrace("msg:%p, get from vnode queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
 
-  switch (msgType) {
+  switch (pMsg->msgType) {
     case TDMT_MON_VM_INFO:
       code = vmProcessGetMonitorInfoReq(pMgmt, pMsg);
       break;
@@ -54,7 +52,7 @@ static void vmProcessMgmtMonitorQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
       dError("msg:%p, not processed in vnode queue", pMsg);
   }
 
-  if (msgType & 1u) {
+  if (IsReq(pMsg)) {
     if (code != 0 && terrno != 0) code = terrno;
     vmSendRsp(pMsg, code);
   }
@@ -94,8 +92,7 @@ static void vmProcessFetchQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
 
 static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-
-  SArray *pArray = taosArrayInit(numOfMsgs, sizeof(SRpcMsg *));
+  SArray *   pArray = taosArrayInit(numOfMsgs, sizeof(SRpcMsg *));
   if (pArray == NULL) {
     dError("failed to process %d msgs in write-queue since %s", numOfMsgs, terrstr());
     return;
@@ -114,7 +111,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
   for (int i = 0; i < taosArrayGetSize(pArray); i++) {
     SRpcMsg *pMsg = *(SRpcMsg **)taosArrayGet(pArray, i);
-    SRpcMsg  rsp = {.info = pMsg->info, .pCont = NULL, .contLen = 0};
+    SRpcMsg  rsp = {.info = pMsg->info};
 
     int32_t ret = syncPropose(vnodeGetSyncHandle(pVnode->pImpl), pMsg, false);
     if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) {
@@ -128,7 +125,6 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
       rsp.code = TSDB_CODE_SYN_INTERNAL_ERROR;
       tmsgSendRsp(&rsp);
     } else if (ret == TAOS_SYNC_PROPOSE_SUCCESS) {
-      // ok
       // send response in applyQ
     } else {
       assert(0);
@@ -147,16 +143,13 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SRpcMsg *  pMsg = NULL;
-  SRpcMsg    rsp;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
+    SRpcMsg *pMsg = NULL;
     taosGetQitem(qall, (void **)&pMsg);
 
     // init response rpc msg
-    rsp.code = 0;
-    rsp.pCont = NULL;
-    rsp.contLen = 0;
+    SRpcMsg rsp = {0};
 
     // get original rpc msg
     assert(pMsg->msgType == TDMT_VND_SYNC_APPLY_MSG);
@@ -175,7 +168,6 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     rpcFreeCont(originalRpcMsg.pCont);
 
     // if leader, send response
-    // if (pMsg->rpcMsg.handle != NULL && pMsg->rpcMsg.ahandle != NULL) {
     if (pMsg->info.handle != NULL) {
       rsp.info = pMsg->info;
       tmsgSendRsp(&rsp);
@@ -188,21 +180,19 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SRpcMsg *  pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
+    SRpcMsg *pMsg = NULL;
     taosGetQitem(qall, (void **)&pMsg);
 
-    // todo
-    SRpcMsg *pRsp = NULL;
-    int32_t  ret = vnodeProcessSyncReq(pVnode->pImpl, pMsg, &pRsp);
-    if (ret != 0) {
-      // if leader, send response
+    int32_t code = vnodeProcessSyncReq(pVnode->pImpl, pMsg, NULL);
+    if (code != 0) {
       if (pMsg->info.handle != NULL) {
-        SRpcMsg rsp = {0};
-        rsp.code = terrno;
-        rsp.info = pMsg->info;
-        dTrace("msg:%p, process sync queue error since code:%s", pMsg, terrstr());
+        SRpcMsg rsp = {
+            .code = (terrno < 0) ? terrno : code,
+            .info = pMsg->info,
+        };
+        dTrace("msg:%p, failed to process sync queue since %s", pMsg, terrstr());
         tmsgSendRsp(&rsp);
       }
     }
@@ -214,9 +204,9 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
 
 static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SRpcMsg *  pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
+    SRpcMsg *pMsg = NULL;
     taosGetQitem(qall, (void **)&pMsg);
 
     dTrace("msg:%p, get from vnode-merge queue", pMsg);
@@ -224,12 +214,11 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     if (code != 0) {
       if (terrno != 0) code = terrno;
       vmSendRsp(pMsg, code);
-
+    }
     dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
     rpcFreeCont(pMsg->pCont);
     taosFreeQitem(pMsg);
   }
-  }
 }
 
 static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) {
@@ -306,22 +295,24 @@ int32_t vmPutNodeMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 
 int32_t vmPutNodeMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SSingleWorker *pWorker = &pMgmt->monitorWorker;
-
   dTrace("msg:%p, put into vnode-monitor worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
   taosWriteQitem(pWorker->queue, pMsg);
   return 0;
 }
 
 static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType qtype) {
-  SMsgHead *pHead = pRpc->pCont;
-
+  SMsgHead * pHead = pRpc->pCont;
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) return -1;
 
   SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM);
   int32_t  code = 0;
 
-  if (pMsg != NULL) {
+  if (pMsg == NULL) {
+    rpcFreeCont(pRpc->pCont);
+    pRpc->pCont = NULL;
+    code = -1;
+  } else {
     memcpy(pMsg, pRpc, sizeof(SRpcMsg));
     switch (qtype) {
       case WRITE_QUEUE:
@@ -426,7 +417,7 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
     return -1;
   }
 
-  dDebug("vgId:%d, vnode queue is alloced", pVnode->vgId);
+  dDebug("vgId:%d, queue is alloced", pVnode->vgId);
   return 0;
 }
 
@@ -443,7 +434,7 @@ void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
   pVnode->pQueryQ = NULL;
   pVnode->pFetchQ = NULL;
   pVnode->pMergeQ = NULL;
-  dDebug("vgId:%d, vnode queue is freed", pVnode->vgId);
+  dDebug("vgId:%d, queue is freed", pVnode->vgId);
 }
 
 int32_t vmStartWorker(SVnodeMgmt *pMgmt) {
@@ -494,7 +485,7 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) {
       .param = pMgmt,
   };
   if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {
-    dError("failed to start mnode vnode-monitor worker since %s", terrstr());
+    dError("failed to start vnode-monitor worker since %s", terrstr());
     return -1;
   }
 

source/dnode/mgmt/node_mgmt/src/dmMonitor.c

@@ -161,10 +161,12 @@ void dmGetVnodeLoads(SMonVloadInfo *pInfo) {
 void dmGetMnodeLoads(SMonMloadInfo *pInfo) {
   SDnode       *pDnode = dmInstance();
   SMgmtWrapper *pWrapper = &pDnode->wrappers[MNODE];
+  if (dmMarkWrapper(pWrapper) == 0) {
     if (tsMultiProcess) {
       dmSendLocalRecv(pDnode, TDMT_MON_MM_LOAD, tDeserializeSMonMloadInfo, pInfo);
     } else if (pWrapper->pMgmt != NULL) {
       mmGetMnodeLoads(pWrapper->pMgmt, pInfo);
     }
     dmReleaseWrapper(pWrapper);
+  }
 }

source/dnode/vnode/src/meta/metaEntry.c

@@ -56,8 +56,8 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
   if (tDecodeCStr(pCoder, &pME->name) < 0) return -1;
 
   if (pME->type == TSDB_SUPER_TABLE) {
-    if (tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schema) < 0) return -1;
-    if (tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schemaTag) < 0) return -1;
+    if (tDecodeSSchemaWrapper(pCoder, &pME->stbEntry.schema) < 0) return -1;
+    if (tDecodeSSchemaWrapper(pCoder, &pME->stbEntry.schemaTag) < 0) return -1;
   } else if (pME->type == TSDB_CHILD_TABLE) {
     if (tDecodeI64(pCoder, &pME->ctbEntry.ctime) < 0) return -1;
     if (tDecodeI32(pCoder, &pME->ctbEntry.ttlDays) < 0) return -1;
@@ -67,10 +67,10 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
     if (tDecodeI64(pCoder, &pME->ntbEntry.ctime) < 0) return -1;
     if (tDecodeI32(pCoder, &pME->ntbEntry.ttlDays) < 0) return -1;
     if (tDecodeI32v(pCoder, &pME->ntbEntry.ncid) < 0) return -1;
-    if (tDecodeSSchemaWrapperEx(pCoder, &pME->ntbEntry.schema) < 0) return -1;
+    if (tDecodeSSchemaWrapper(pCoder, &pME->ntbEntry.schema) < 0) return -1;
   } else if (pME->type == TSDB_TSMA_TABLE) {
     pME->smaEntry.tsma = tDecoderMalloc(pCoder, sizeof(STSma));
-    if(!pME->smaEntry.tsma) {
+    if (!pME->smaEntry.tsma) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return -1;
     }

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -425,6 +425,12 @@ static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond*
       rowLen += pCond->colList[i].bytes;
     }
 
+    // make sure the output SSDataBlock size be less than 2MB.
+    int32_t TWOMB = 2 * 1024 * 1024;
+    if (pReadHandle->outputCapacity * rowLen > TWOMB) {
+      pReadHandle->outputCapacity = TWOMB / rowLen;
+    }
+
     // allocate buffer in order to load data blocks from file
     pReadHandle->suppInfo.pstatis = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnDataAgg));
     if (pReadHandle->suppInfo.pstatis == NULL) {
@@ -1302,20 +1308,22 @@ static int32_t handleDataMergeIfNeeded(STsdbReadHandle* pTsdbReadHandle, SBlock*
 
   if ((ascScan && (key != TSKEY_INITIAL_VAL && key <= binfo.window.ekey)) ||
       (!ascScan && (key != TSKEY_INITIAL_VAL && key >= binfo.window.skey))) {
-    if ((ascScan && (key != TSKEY_INITIAL_VAL && key < binfo.window.skey)) ||
-        (!ascScan && (key != TSKEY_INITIAL_VAL && key > binfo.window.ekey))) {
+
+    bool cacheDataInFileBlockHole = (ascScan && (key != TSKEY_INITIAL_VAL && key < binfo.window.skey)) ||
+                                    (!ascScan && (key != TSKEY_INITIAL_VAL && key > binfo.window.ekey));
+    if (cacheDataInFileBlockHole) {
       // do not load file block into buffer
       int32_t step = ascScan ? 1 : -1;
 
-      TSKEY maxKey =
-          ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? (binfo.window.skey - step) : (binfo.window.ekey - step);
+      TSKEY maxKey = ascScan ? (binfo.window.skey - step) : (binfo.window.ekey - step);
       cur->rows =
           tsdbReadRowsFromCache(pCheckInfo, maxKey, pTsdbReadHandle->outputCapacity, &cur->win, pTsdbReadHandle);
       pTsdbReadHandle->realNumOfRows = cur->rows;
 
       // update the last key value
       pCheckInfo->lastKey = cur->win.ekey + step;
-      if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
+
+      if (!ascScan) {
         TSWAP(cur->win.skey, cur->win.ekey);
       }
 
@@ -1334,14 +1342,12 @@ static int32_t handleDataMergeIfNeeded(STsdbReadHandle* pTsdbReadHandle, SBlock*
     /*
      * no data in cache, only load data from file
      * during the query processing, data in cache will not be checked anymore.
-     *
      * Here the buffer is not enough, so only part of file block can be loaded into memory buffer
      */
-    assert(pTsdbReadHandle->outputCapacity >= binfo.rows);
     int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &binfo);
 
-    if ((cur->pos == 0 && endPos == binfo.rows - 1 && ascScan) ||
-        (cur->pos == (binfo.rows - 1) && endPos == 0 && (!ascScan))) {
+    bool wholeBlockReturned = ((abs(cur->pos - endPos) + 1) == binfo.rows);
+    if (wholeBlockReturned) {
       pTsdbReadHandle->realNumOfRows = binfo.rows;
 
       cur->rows = binfo.rows;
@@ -1357,11 +1363,23 @@ static int32_t handleDataMergeIfNeeded(STsdbReadHandle* pTsdbReadHandle, SBlock*
         cur->pos = -1;
       }
     } else { // partially copy to dest buffer
+      // make sure to only load once
+      bool firstTimeExtract = ((cur->pos == 0 && ascScan) || (cur->pos == binfo.rows -1 && (!ascScan)));
+      if (pTsdbReadHandle->outputCapacity < binfo.rows && firstTimeExtract) {
+        code = doLoadFileDataBlock(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
+      }
+
       copyAllRemainRowsFromFileBlock(pTsdbReadHandle, pCheckInfo, &binfo, endPos);
       cur->mixBlock = true;
     }
 
-    assert(cur->blockCompleted);
+    if (pTsdbReadHandle->outputCapacity >= binfo.rows) {
+      ASSERT(cur->blockCompleted);
+    }
+
     if (cur->rows == binfo.rows) {
       tsdbDebug("%p whole file block qualified, brange:%" PRId64 "-%" PRId64 ", rows:%d, lastKey:%" PRId64 ", %s",
                 pTsdbReadHandle, cur->win.skey, cur->win.ekey, cur->rows, cur->lastKey, pTsdbReadHandle->idStr);
@@ -1858,15 +1876,14 @@ static void copyAllRemainRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, STa
   SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
   TSKEY*     tsArray = pCols->cols[0].pData;
 
-  int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
-  int32_t numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pTsdbReadHandle));
+  bool ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
 
-  int32_t pos = cur->pos;
+  int32_t step = ascScan? 1 : -1;
 
   int32_t start = cur->pos;
   int32_t end = endPos;
 
-  if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
+  if (!ascScan) {
     TSWAP(start, end);
   }
 
@@ -1877,10 +1894,10 @@ static void copyAllRemainRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, STa
   cur->win = (STimeWindow){.skey = tsArray[start], .ekey = tsArray[end]};
   cur->mixBlock = (numOfRows != pBlockInfo->rows);
   cur->lastKey  = tsArray[endPos] + step;
-  cur->blockCompleted = true;
+  cur->blockCompleted = (ascScan? (endPos == pBlockInfo->rows - 1):(endPos == 0));
 
   // The value of pos may be -1 or pBlockInfo->rows, and it is invalid in both cases.
-  pos = endPos + step;
+  int32_t pos = endPos + step;
   updateInfoAfterMerge(pTsdbReadHandle, pCheckInfo, numOfRows, pos);
   doCheckGeneratedBlockRange(pTsdbReadHandle);
 
@@ -1892,15 +1909,17 @@ static void copyAllRemainRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, STa
 int32_t getEndPosInDataBlock(STsdbReadHandle* pTsdbReadHandle, SDataBlockInfo* pBlockInfo) {
   // NOTE: reverse the order to find the end position in data block
   int32_t endPos = -1;
-  int32_t order = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
+  bool    ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
+  int32_t order = ascScan? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
 
   SQueryFilePos* cur = &pTsdbReadHandle->cur;
   SDataCols*     pCols = pTsdbReadHandle->rhelper.pDCols[0];
 
-  if (ASCENDING_TRAVERSE(pTsdbReadHandle->order) && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
+  if (pTsdbReadHandle->outputCapacity >= pBlockInfo->rows) {
+    if (ascScan && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
       endPos = pBlockInfo->rows - 1;
       cur->mixBlock = (cur->pos != 0);
-  } else if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
+    } else if ((!ascScan) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
       endPos = 0;
       cur->mixBlock = (cur->pos != pBlockInfo->rows - 1);
     } else {
@@ -1908,6 +1927,28 @@ int32_t getEndPosInDataBlock(STsdbReadHandle* pTsdbReadHandle, SDataBlockInfo* p
       endPos = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, pTsdbReadHandle->window.ekey, order);
       cur->mixBlock = true;
     }
+  } else {
+    if (ascScan && pTsdbReadHandle->window.ekey >= pBlockInfo->window.ekey) {
+      endPos = TMIN(cur->pos + pTsdbReadHandle->outputCapacity - 1, pBlockInfo->rows - 1);
+    } else if ((!ascScan) && pTsdbReadHandle->window.ekey <= pBlockInfo->window.skey) {
+      endPos = TMAX(cur->pos - pTsdbReadHandle->outputCapacity + 1, 0);
+    } else {
+      ASSERT(pCols->numOfRows > 0);
+      endPos = doBinarySearchKey(pCols->cols[0].pData, pCols->numOfRows, pTsdbReadHandle->window.ekey, order);
+
+      // current data is more than the capacity
+      int32_t size = abs(cur->pos - endPos) + 1;
+      if (size > pTsdbReadHandle->outputCapacity) {
+        int32_t delta = size - pTsdbReadHandle->outputCapacity;
+        if (ascScan) {
+          endPos -= delta;
+        } else {
+          endPos += delta;
+        }
+      }
+    }
+    cur->mixBlock = true;
+  }
 
   return endPos;
 }
@@ -2369,7 +2410,7 @@ static int32_t createDataBlocksInfo(STsdbReadHandle* pTsdbReadHandle, int32_t nu
 
 static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exists);
 
-static int32_t getDataBlockRv(STsdbReadHandle* pTsdbReadHandle, STableBlockInfo* pNext, bool* exists) {
+static int32_t getDataBlock(STsdbReadHandle* pTsdbReadHandle, STableBlockInfo* pNext, bool* exists) {
   int32_t        step = ASCENDING_TRAVERSE(pTsdbReadHandle->order) ? 1 : -1;
   SQueryFilePos* cur = &pTsdbReadHandle->cur;
 
@@ -2478,7 +2519,7 @@ static int32_t getFirstFileDataBlock(STsdbReadHandle* pTsdbReadHandle, bool* exi
   cur->fid = pTsdbReadHandle->pFileGroup->fid;
 
   STableBlockInfo* pBlockInfo = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
-  return getDataBlockRv(pTsdbReadHandle, pBlockInfo, exists);
+  return getDataBlock(pTsdbReadHandle, pBlockInfo, exists);
 }
 
 static bool isEndFileDataBlock(SQueryFilePos* cur, int32_t numOfBlocks, bool ascTrav) {
@@ -2643,7 +2684,7 @@ static int32_t getDataBlocksInFiles(STsdbReadHandle* pTsdbReadHandle, bool* exis
     } else {
       moveToNextDataBlockInCurrentFile(pTsdbReadHandle);
       STableBlockInfo* pNext = &pTsdbReadHandle->pDataBlockInfo[cur->slot];
-      return getDataBlockRv(pTsdbReadHandle, pNext, exists);
+      return getDataBlock(pTsdbReadHandle, pNext, exists);
     }
   }
 }

source/libs/executor/src/executorimpl.c

@@ -3546,11 +3546,12 @@ _error:
 
 int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t *order, int32_t* scanFlag) {
   // todo add more information about exchange operation
-  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_EXCHANGE) {
+  int32_t type = pOperator->operatorType;
+  if (type == QUERY_NODE_PHYSICAL_PLAN_EXCHANGE || type == QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
     *order = TSDB_ORDER_ASC;
     *scanFlag = MAIN_SCAN;
     return TSDB_CODE_SUCCESS;
-  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
+  } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
     STableScanInfo* pTableScanInfo = pOperator->info;
     *order = pTableScanInfo->cond.order;
     *scanFlag = pTableScanInfo->scanFlag;
@@ -3910,6 +3911,9 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
 
     // the pDataBlock are always the same one, no need to call this again
     int32_t code = getTableScanInfo(pOperator->pDownstream[0], &order, &scanFlag);
+    if (code != TSDB_CODE_SUCCESS) {
+      longjmp(pTaskInfo->env, code);
+    }
 
     setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, scanFlag, false);
     blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
@@ -4203,7 +4207,7 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo*
   pInfo->pScalarExprInfo = pScalarExprInfo;
   pInfo->numOfScalarExpr = numOfScalarExpr;
   if (pInfo->pScalarExprInfo != NULL) {
-    pInfo->pScalarCtx = createSqlFunctionCtx(pScalarExprInfo, numOfCols, &pInfo->rowCellInfoOffset);
+    pInfo->pScalarCtx = createSqlFunctionCtx(pScalarExprInfo, numOfScalarExpr, &pInfo->rowCellInfoOffset);
   }
 
   pOperator->name = "TableAggregate";
@@ -4311,11 +4315,17 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* p
   int32_t numOfRows = 4096;
   size_t  keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
 
+  // Make sure the size of SSDataBlock will never exceed the size of 2MB.
+  int32_t TWOMB = 2 * 1024 * 1024;
+  if (numOfRows * pResBlock->info.rowSize > TWOMB) {
+    numOfRows = TWOMB / pResBlock->info.rowSize;
+  }
   initResultSizeInfo(pOperator, numOfRows);
+
   initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str);
   setFunctionResultOutput(&pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfCols, pTaskInfo);
-  pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pInfo->binfo.pCtx, numOfCols);
 
+  pInfo->pPseudoColInfo   = setRowTsColumnOutputInfo(pInfo->binfo.pCtx, numOfCols);
   pOperator->name         = "ProjectOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PROJECT;
   pOperator->blocking     = false;
@@ -4323,11 +4333,11 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* p
   pOperator->info         = pInfo;
   pOperator->pExpr        = pExprInfo;
   pOperator->numOfExprs   = num;
+  pOperator->pTaskInfo    = pTaskInfo;
 
   pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doProjectOperation, NULL, NULL,
                                          destroyProjectOperatorInfo, NULL, NULL, NULL);
 
-  pOperator->pTaskInfo = pTaskInfo;
   int32_t code = appendDownstream(pOperator, &downstream, 1);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;

source/libs/executor/src/scanoperator.c

@@ -300,10 +300,26 @@ void addTagPseudoColumnData(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock)
     if (fmIsScanPseudoColumnFunc(functionId)) {
       setTbNameColData(pTableScanInfo->readHandle.meta, pBlock, pColInfoData, functionId);
     } else {  // these are tags
-      const char* p = metaGetTableTagVal(&mr.me, pExpr->base.pParam[0].pCol->colId);
+      const char* p = NULL;
+      if(pColInfoData->info.type == TSDB_DATA_TYPE_JSON){
+        const uint8_t *tmp = mr.me.ctbEntry.pTags;
+        char *data = taosMemoryCalloc(kvRowLen(tmp) + 1, 1);
+        if(data == NULL){
+          qError("doTagScan calloc error:%d", kvRowLen(tmp) + 1);
+          return;
+        }
+        *data = TSDB_DATA_TYPE_JSON;
+        memcpy(data+1, tmp, kvRowLen(tmp));
+        p = data;
+      }else{
+        p = metaGetTableTagVal(&mr.me, pExpr->base.pParam[0].pCol->colId);
+      }
       for (int32_t i = 0; i < pBlock->info.rows; ++i) {
         colDataAppend(pColInfoData, i, p, (p == NULL));
       }
+      if(pColInfoData->info.type == TSDB_DATA_TYPE_JSON){
+        taosMemoryFree((void*)p);
+      }
     }
   }
 
@@ -1587,10 +1603,23 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
         STR_TO_VARSTR(str, mr.me.name);
         colDataAppend(pDst, count, str, false);
       } else { // it is a tag value
+        if(pDst->info.type == TSDB_DATA_TYPE_JSON){
+          const uint8_t *tmp = mr.me.ctbEntry.pTags;
+          char *data = taosMemoryCalloc(kvRowLen(tmp) + 1, 1);
+          if(data == NULL){
+            qError("doTagScan calloc error:%d", kvRowLen(tmp) + 1);
+            return NULL;
+          }
+          *data = TSDB_DATA_TYPE_JSON;
+          memcpy(data+1, tmp, kvRowLen(tmp));
+          colDataAppend(pDst, count, data, false);
+          taosMemoryFree(data);
+        }else{
           const char* p = metaGetTableTagVal(&mr.me, pExprInfo[j].base.pParam[0].pCol->colId);
           colDataAppend(pDst, count, p, (p == NULL));
         }
       }
+    }
 
     count += 1;
     if (++pInfo->curPos >= pInfo->pTableGroups->numOfTables) {

source/libs/function/inc/builtinsimpl.h

@@ -76,11 +76,6 @@ int32_t firstFunction(SqlFunctionCtx *pCtx);
 int32_t lastFunction(SqlFunctionCtx *pCtx);
 int32_t lastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
 
-bool getUniqueFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
-bool uniqueFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
-int32_t uniqueFunction(SqlFunctionCtx *pCtx);
-int32_t uniqueFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
-
 bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
 int32_t topFunction(SqlFunctionCtx *pCtx);
 int32_t bottomFunction(SqlFunctionCtx *pCtx);
@@ -125,7 +120,13 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx);
 bool getTailFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool tailFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t tailFunction(SqlFunctionCtx* pCtx);
-int32_t tailFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+//int32_t tailFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+
+bool getUniqueFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+bool uniqueFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
+int32_t uniqueFunction(SqlFunctionCtx *pCtx);
+//int32_t uniqueFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+
 
 bool getSelectivityFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 

source/libs/function/src/builtins.c

@@ -892,16 +892,6 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .processFunc  = lastFunction,
     .finalizeFunc = lastFinalize
   },
-  {
-    .name = "unique",
-    .type = FUNCTION_TYPE_UNIQUE,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC,
-    .translateFunc = translateUnique,
-    .getEnvFunc   = getUniqueFuncEnv,
-    .initFunc     = uniqueFunctionSetup,
-    .processFunc  = uniqueFunction,
-    .finalizeFunc = uniqueFinalize
-  },
   {
     .name = "histogram",
     .type = FUNCTION_TYPE_HISTOGRAM,
@@ -990,7 +980,17 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .getEnvFunc   = getTailFuncEnv,
     .initFunc     = tailFunctionSetup,
     .processFunc  = tailFunction,
-    .finalizeFunc = tailFinalize
+    .finalizeFunc = NULL
+  },
+  {
+    .name = "unique",
+    .type = FUNCTION_TYPE_UNIQUE,
+    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .translateFunc = translateUnique,
+    .getEnvFunc   = getUniqueFuncEnv,
+    .initFunc     = uniqueFunctionSetup,
+    .processFunc  = uniqueFunction,
+    .finalizeFunc = NULL
   },
   {
     .name = "abs",

source/libs/function/src/builtinsimpl.c

@@ -1979,99 +1979,6 @@ int32_t lastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   return pResInfo->numOfRes;
 }
 
-bool getUniqueFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
-  pEnv->calcMemSize = sizeof(SUniqueInfo) + UNIQUE_MAX_RESULT_SIZE;
-  return true;
-}
-
-bool uniqueFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
-  if (!functionSetup(pCtx, pResInfo)) {
-    return false;
-  }
-
-  SUniqueInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
-  pInfo->numOfPoints = 0;
-  pInfo->colType = pCtx->resDataInfo.type;
-  pInfo->colBytes = pCtx->resDataInfo.bytes;
-  if (pInfo->pHash != NULL) {
-    taosHashClear(pInfo->pHash);
-  } else {
-    pInfo->pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
-  }
-  return true;
-}
-
-static void doUniqueAdd(SUniqueInfo* pInfo, char *data, TSKEY ts, bool isNull) {
-  int32_t hashKeyBytes = IS_VAR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes;
-
-  SUniqueItem *pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes);
-  if (pHashItem == NULL) {
-    int32_t size = sizeof(SUniqueItem) + pInfo->colBytes;
-    SUniqueItem *pItem = (SUniqueItem *)(pInfo->pItems + pInfo->numOfPoints * size);
-    pItem->timestamp = ts;
-    memcpy(pItem->data, data, pInfo->colBytes);
-
-    taosHashPut(pInfo->pHash, data, hashKeyBytes, (char *)pItem, sizeof(SUniqueItem*));
-    pInfo->numOfPoints++;
-  } else if (pHashItem->timestamp > ts) {
-    pHashItem->timestamp = ts;
-  }
-
-}
-
-int32_t uniqueFunction(SqlFunctionCtx* pCtx) {
-  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
-  SUniqueInfo*         pInfo = GET_ROWCELL_INTERBUF(pResInfo);
-
-  SInputColumnInfoData* pInput = &pCtx->input;
-  TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
-
-  SColumnInfoData* pInputCol = pInput->pData[0];
-  SColumnInfoData* pTsOutput = pCtx->pTsOutput;
-  SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
-
-  int32_t startOffset = pCtx->offset;
-  for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
-    char* data = colDataGetData(pInputCol, i);
-    doUniqueAdd(pInfo, data, tsList[i], colDataIsNull_s(pInputCol, i));
-
-    if (sizeof(SUniqueInfo) + pInfo->numOfPoints * (sizeof(SUniqueItem) + pInfo->colBytes) >= UNIQUE_MAX_RESULT_SIZE) {
-      taosHashCleanup(pInfo->pHash);
-      return 0;
-    }
-  }
-
-  //taosqsort(pInfo->pItems, pInfo->numOfPoints, POINTER_BYTES, NULL, tailCompFn);
-
-  //for (int32_t i = 0; i < pInfo->numOfPoints; ++i) {
-  //  int32_t pos = startOffset + i;
-  //  STailItem *pItem = pInfo->pItems[i];
-  //  if (pItem->isNull) {
-  //    colDataAppendNULL(pOutput, pos);
-  //  } else {
-  //    colDataAppend(pOutput, pos, pItem->data, false);
-  //  }
-  //}
-
-  pResInfo->numOfRes = pInfo->numOfPoints;
-  return TSDB_CODE_SUCCESS;
-}
-
-int32_t uniqueFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
-  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
-  SUniqueInfo*    pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
-  int32_t        slotId = pCtx->pExpr->base.resSchema.slotId;
-  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
-
-  for (int32_t i = 0; i < pResInfo->numOfRes; ++i) {
-    SUniqueItem *pItem = (SUniqueItem *)(pInfo->pItems + i * (sizeof(SUniqueItem) + pInfo->colBytes));
-    colDataAppend(pCol, i, pItem->data, false);
-    //TODO: handle ts output
-  }
-
-  return pResInfo->numOfRes;
-}
-
 bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SDiffInfo);
   return true;
@@ -3659,3 +3566,92 @@ int32_t tailFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 
   return pEntryInfo->numOfRes;
 }
+
+bool getUniqueFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
+  pEnv->calcMemSize = sizeof(SUniqueInfo) + UNIQUE_MAX_RESULT_SIZE;
+  return true;
+}
+
+bool uniqueFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
+  if (!functionSetup(pCtx, pResInfo)) {
+    return false;
+  }
+
+  SUniqueInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+  pInfo->numOfPoints = 0;
+  pInfo->colType = pCtx->resDataInfo.type;
+  pInfo->colBytes = pCtx->resDataInfo.bytes;
+  if (pInfo->pHash != NULL) {
+    taosHashClear(pInfo->pHash);
+  } else {
+    pInfo->pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+  }
+  return true;
+}
+
+static void doUniqueAdd(SUniqueInfo* pInfo, char *data, TSKEY ts, bool isNull) {
+  int32_t hashKeyBytes = IS_VAR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes;
+
+  SUniqueItem *pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes);
+  if (pHashItem == NULL) {
+    int32_t size = sizeof(SUniqueItem) + pInfo->colBytes;
+    SUniqueItem *pItem = (SUniqueItem *)(pInfo->pItems + pInfo->numOfPoints * size);
+    pItem->timestamp = ts;
+    memcpy(pItem->data, data, pInfo->colBytes);
+
+    taosHashPut(pInfo->pHash, data, hashKeyBytes, (char *)pItem, sizeof(SUniqueItem*));
+    pInfo->numOfPoints++;
+  } else if (pHashItem->timestamp > ts) {
+    pHashItem->timestamp = ts;
+  }
+
+}
+
+int32_t uniqueFunction(SqlFunctionCtx* pCtx) {
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  SUniqueInfo*         pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+
+  SInputColumnInfoData* pInput = &pCtx->input;
+  TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
+
+  SColumnInfoData* pInputCol = pInput->pData[0];
+  SColumnInfoData* pTsOutput = pCtx->pTsOutput;
+  SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
+
+  int32_t startOffset = pCtx->offset;
+  for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
+    char* data = colDataGetData(pInputCol, i);
+    doUniqueAdd(pInfo, data, tsList[i], colDataIsNull_s(pInputCol, i));
+
+    if (sizeof(SUniqueInfo) + pInfo->numOfPoints * (sizeof(SUniqueItem) + pInfo->colBytes) >= UNIQUE_MAX_RESULT_SIZE) {
+      taosHashCleanup(pInfo->pHash);
+      return 0;
+    }
+  }
+
+  for (int32_t i = 0; i < pInfo->numOfPoints; ++i) {
+    SUniqueItem *pItem = (SUniqueItem *)(pInfo->pItems + i * (sizeof(SUniqueItem) + pInfo->colBytes));
+    colDataAppend(pOutput, i, pItem->data, false);
+    if (pTsOutput != NULL) {
+      colDataAppendInt64(pTsOutput, i, &pItem->timestamp);
+    }
+  }
+
+  return pInfo->numOfPoints;
+}
+
+int32_t uniqueFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  SUniqueInfo*    pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
+  int32_t        slotId = pCtx->pExpr->base.resSchema.slotId;
+  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
+
+  for (int32_t i = 0; i < pResInfo->numOfRes; ++i) {
+    SUniqueItem *pItem = (SUniqueItem *)(pInfo->pItems + i * (sizeof(SUniqueItem) + pInfo->colBytes));
+    colDataAppend(pCol, i, pItem->data, false);
+    //TODO: handle ts output
+  }
+
+  return pResInfo->numOfRes;
+}
+

source/libs/nodes/src/nodesCodeFuncs.c

@@ -1771,6 +1771,7 @@ static const char* jkSubplanId = "Id";
 static const char* jkSubplanType = "SubplanType";
 static const char* jkSubplanMsgType = "MsgType";
 static const char* jkSubplanLevel = "Level";
+static const char* jkSubplanDbFName = "DbFName";
 static const char* jkSubplanNodeAddr = "NodeAddr";
 static const char* jkSubplanRootNode = "RootNode";
 static const char* jkSubplanDataSink = "DataSink";
@@ -1788,6 +1789,9 @@ static int32_t subplanToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddIntegerToObject(pJson, jkSubplanLevel, pNode->level);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddStringToObject(pJson, jkSubplanDbFName, pNode->dbFName);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkSubplanNodeAddr, queryNodeAddrToJson, &pNode->execNode);
   }
@@ -1815,6 +1819,9 @@ static int32_t jsonToSubplan(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetIntValue(pJson, jkSubplanLevel, &pNode->level);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetStringValue(pJson, jkSubplanDbFName, pNode->dbFName);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonToObject(pJson, jkSubplanNodeAddr, jsonToQueryNodeAddr, &pNode->execNode);
   }

source/libs/nodes/src/nodesUtilFuncs.c

@@ -1137,10 +1137,6 @@ bool nodesIsRegularOp(const SOperatorNode* pOp) {
   return false;
 }
 
-bool nodesIsTimeorderQuery(const SNode* pQuery) { return false; }
-
-bool nodesIsTimelineQuery(const SNode* pQuery) { return false; }
-
 typedef struct SCollectColumnsCxt {
   int32_t         errCode;
   const char*     pTableAlias;

source/libs/parser/src/parTranslater.c

@@ -382,6 +382,35 @@ static bool isInternalPrimaryKey(const SColumnNode* pCol) {
   return PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && 0 == strcmp(pCol->colName, PK_TS_COL_INTERNAL_NAME);
 }
 
+static bool isTimeOrderQuery(SNode* pStmt) {
+  if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) {
+    return ((SSelectStmt*)pStmt)->isTimeOrderQuery;
+  } else {
+    return false;
+  }
+}
+
+static bool isPrimaryKeyImpl(STempTableNode* pTable, SNode* pExpr) {
+  if (QUERY_NODE_COLUMN == nodeType(pExpr)) {
+    return (PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)pExpr)->colId);
+  } else if (QUERY_NODE_FUNCTION == nodeType(pExpr)) {
+    SFunctionNode* pFunc = (SFunctionNode*)pExpr;
+    if (FUNCTION_TYPE_SELECT_VALUE == pFunc->funcType) {
+      return isPrimaryKeyImpl(pTable, nodesListGetNode(pFunc->pParameterList, 0));
+    } else if (FUNCTION_TYPE_WSTARTTS == pFunc->funcType || FUNCTION_TYPE_WENDTS == pFunc->funcType) {
+      return true;
+    }
+  }
+  return false;
+}
+
+static bool isPrimaryKey(STempTableNode* pTable, SNode* pExpr) {
+  if (!isTimeOrderQuery(pTable->pSubquery)) {
+    return false;
+  }
+  return isPrimaryKeyImpl(pTable, pExpr);
+}
+
 static bool findAndSetColumn(SColumnNode* pCol, const STableNode* pTable) {
   bool found = false;
   if (QUERY_NODE_REAL_TABLE == nodeType(pTable)) {
@@ -404,8 +433,7 @@ static bool findAndSetColumn(SColumnNode* pCol, const STableNode* pTable) {
     FOREACH(pNode, pProjectList) {
       SExprNode* pExpr = (SExprNode*)pNode;
       if (0 == strcmp(pCol->colName, pExpr->aliasName) ||
-          ((QUERY_NODE_COLUMN == nodeType(pExpr) && PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)pExpr)->colId) &&
-           isInternalPrimaryKey(pCol))) {
+          (isPrimaryKey((STempTableNode*)pTable, pNode) && isInternalPrimaryKey(pCol))) {
         setColumnInfoByExpr(pTable, pExpr, pCol);
         found = true;
         break;
@@ -454,6 +482,9 @@ static EDealRes translateColumnWithoutPrefix(STranslateContext* pCxt, SColumnNod
   }
   if (!found) {
     if (isInternalPk) {
+      if (NULL != pCxt->pCurrStmt->pWindow) {
+        return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_NOT_ALLOWED_WIN_QUERY);
+      }
       return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_INTERNAL_PK);
     } else {
       return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_COLUMN, pCol->colName);
@@ -781,7 +812,6 @@ static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode* pFunc)
     }
 
     pCxt->pCurrStmt->hasAggFuncs = true;
-    pCxt->pCurrStmt->isTimeOrderQuery = false;
     if (isCountStar(pFunc)) {
       pCxt->errCode = rewriteCountStar(pCxt, pFunc);
     }

source/libs/parser/src/parUtil.c

@@ -167,6 +167,8 @@ static char* getSyntaxErrFormat(int32_t errCode) {
     case TSDB_CODE_PAR_NOT_ALLOWED_FUNC:
       return "Some functions are allowed only in the SELECT list of a query. "
              "And, cannot be mixed with other non scalar functions or columns.";
+    case TSDB_CODE_PAR_NOT_ALLOWED_WIN_QUERY:
+      return "Window query not supported, since the result of subquery not include valid timestamp column";
     case TSDB_CODE_OUT_OF_MEMORY:
       return "Out of memory";
     default:
@@ -365,8 +367,8 @@ int parseJsontoTagData(const char* json, SKVRowBuilder* kvRowBuilder, SMsgBuf* p
     if (keyLen == 0 || taosHashGet(keyHash, jsonKey, keyLen) != NULL) {
       continue;
     }
-    // key: keyLen + VARSTR_HEADER_SIZE, value type: CHAR_BYTES, value reserved: LONG_BYTES
-    tagKV = taosMemoryCalloc(keyLen + VARSTR_HEADER_SIZE + CHAR_BYTES + LONG_BYTES, 1);
+    // key: keyLen + VARSTR_HEADER_SIZE, value type: CHAR_BYTES, value reserved: DOUBLE_BYTES
+    tagKV = taosMemoryCalloc(keyLen + VARSTR_HEADER_SIZE + CHAR_BYTES + DOUBLE_BYTES, 1);
     if (!tagKV) {
       retCode = TSDB_CODE_TSC_OUT_OF_MEMORY;
       goto end;
@@ -411,13 +413,9 @@ int parseJsontoTagData(const char* json, SKVRowBuilder* kvRowBuilder, SMsgBuf* p
       }
       char* valueType = POINTER_SHIFT(tagKV, keyLen + VARSTR_HEADER_SIZE);
       char* valueData = POINTER_SHIFT(tagKV, keyLen + VARSTR_HEADER_SIZE + CHAR_BYTES);
-      *valueType =
-          (item->valuedouble - (int64_t)(item->valuedouble) == 0) ? TSDB_DATA_TYPE_BIGINT : TSDB_DATA_TYPE_DOUBLE;
-      if (*valueType == TSDB_DATA_TYPE_DOUBLE)
+      *valueType = TSDB_DATA_TYPE_DOUBLE;
       *((double*)valueData) = item->valuedouble;
-      else if (*valueType == TSDB_DATA_TYPE_BIGINT)
-        *((int64_t*)valueData) = item->valueint;
-      tdAddColToKVRow(kvRowBuilder, jsonIndex++, tagKV, keyLen + VARSTR_HEADER_SIZE + CHAR_BYTES + LONG_BYTES);
+      tdAddColToKVRow(kvRowBuilder, jsonIndex++, tagKV, keyLen + VARSTR_HEADER_SIZE + CHAR_BYTES + DOUBLE_BYTES);
     } else if (item->type == cJSON_True || item->type == cJSON_False) {
       char* valueType = POINTER_SHIFT(tagKV, keyLen + VARSTR_HEADER_SIZE);
       char* valueData = POINTER_SHIFT(tagKV, keyLen + VARSTR_HEADER_SIZE + CHAR_BYTES);

source/libs/parser/test/parSelectTest.cpp

@@ -125,8 +125,6 @@ TEST_F(ParserSelectTest, nonstdFunc) {
   useDb("root", "test");
 
   run("SELECT DIFF(c1) FROM t1");
-
-  // run("SELECT DIFF(c1) FROM t1 INTERVAL(10s)");
 }
 
 TEST_F(ParserSelectTest, nonstdFuncSemanticCheck) {
@@ -139,12 +137,13 @@ TEST_F(ParserSelectTest, nonstdFuncSemanticCheck) {
   run("SELECT DIFF(c1), count(*) FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
 
   run("SELECT DIFF(c1), CSUM(c1) FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
+
+  // run("SELECT DIFF(c1) FROM t1 INTERVAL(10s)");
 }
 
-TEST_F(ParserSelectTest, clause) {
+TEST_F(ParserSelectTest, groupBy) {
   useDb("root", "test");
 
-  // GROUP BY clause
   run("SELECT COUNT(*) cnt FROM t1 WHERE c1 > 0");
 
   run("SELECT COUNT(*), c2 cnt FROM t1 WHERE c1 > 0 GROUP BY c2");
@@ -154,13 +153,19 @@ TEST_F(ParserSelectTest, clause) {
   run("SELECT COUNT(*), c1, c2 + 10, c1 + c2 cnt FROM t1 WHERE c1 > 0 GROUP BY c2, c1");
 
   run("SELECT COUNT(*), c1 + 10, c2 cnt FROM t1 WHERE c1 > 0 GROUP BY c1 + 10, c2");
+}
+
+TEST_F(ParserSelectTest, orderBy) {
+  useDb("root", "test");
 
-  // order by clause
   run("SELECT COUNT(*) cnt FROM t1 WHERE c1 > 0 GROUP BY c2 order by cnt");
 
   run("SELECT COUNT(*) cnt FROM t1 WHERE c1 > 0 GROUP BY c2 order by 1");
+}
+
+TEST_F(ParserSelectTest, distinct) {
+  useDb("root", "test");
 
-  // distinct clause
   // run("SELECT distinct c1, c2 FROM t1 WHERE c1 > 0 order by c1");
 
   // run("SELECT distinct c1 + 10, c2 FROM t1 WHERE c1 > 0 order by c1 + 10, c2");
@@ -194,6 +199,25 @@ TEST_F(ParserSelectTest, intervalSemanticCheck) {
       PARSER_STAGE_TRANSLATE);
 }
 
+TEST_F(ParserSelectTest, subquery) {
+  useDb("root", "test");
+
+  run("SELECT SUM(a) FROM (SELECT MAX(c1) a, ts FROM st1s1 INTERVAL(1m)) INTERVAL(1n)");
+
+  run("SELECT SUM(a) FROM (SELECT MAX(c1) a, _wstartts FROM st1s1 INTERVAL(1m)) INTERVAL(1n)");
+
+  run("SELECT SUM(a) FROM (SELECT MAX(c1) a, ts FROM st1s1 PARTITION BY TBNAME INTERVAL(1m)) INTERVAL(1n)");
+
+  run("SELECT SUM(a) FROM (SELECT MAX(c1) a, _wstartts FROM st1s1 PARTITION BY TBNAME INTERVAL(1m)) INTERVAL(1n)");
+}
+
+TEST_F(ParserSelectTest, subquerySemanticError) {
+  useDb("root", "test");
+
+  run("SELECT SUM(a) FROM (SELECT MAX(c1) a FROM st1s1 INTERVAL(1m)) INTERVAL(1n)", TSDB_CODE_PAR_NOT_ALLOWED_WIN_QUERY,
+      PARSER_STAGE_TRANSLATE);
+}
+
 TEST_F(ParserSelectTest, semanticError) {
   useDb("root", "test");
 

source/libs/scalar/src/scalar.c

@@ -899,7 +899,7 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
   }
 
   int32_t code = 0;
-  SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList, .param = pDst->param};
+  SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList, .param = pDst ? pDst->param : NULL};
 
   // TODO: OPT performance
   ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);

source/libs/scalar/src/sclvector.c

@@ -923,6 +923,7 @@ static void doReleaseVec(SColumnInfoData* pCol, int32_t type) {
 }
 
 char *getJsonValue(char *json, char *key){      //todo
+  json++;     // jump type
   int16_t cols = kvRowNCols(json);
   for (int i = 0; i < cols; ++i) {
     SColIdx *pColIdx = kvRowColIdxAt(json, i);

source/libs/scalar/test/scalar/scalarTests.cpp

@@ -1035,7 +1035,7 @@ void makeJsonArrow(SSDataBlock **src, SNode **opNode, void *json, char *key){
 
   SNode *pLeft = NULL, *pRight = NULL;
   scltMakeValueNode(&pRight, TSDB_DATA_TYPE_BINARY, keyVar);
-  scltMakeColumnNode(&pLeft, src, TSDB_DATA_TYPE_JSON, varDataLen(json), 1, json);
+  scltMakeColumnNode(&pLeft, src, TSDB_DATA_TYPE_JSON, kvRowLen(json), 1, json);
   scltMakeOpNode(opNode, OP_TYPE_JSON_GET_VALUE, TSDB_DATA_TYPE_JSON, pLeft, pRight);
 }
 
@@ -1088,18 +1088,17 @@ void makeCalculate(void *json, void *key, int32_t rightType, void *rightData, do
 
   }else if(opType == OP_TYPE_ADD || opType == OP_TYPE_SUB || opType == OP_TYPE_MULTI || opType == OP_TYPE_DIV ||
              opType == OP_TYPE_MOD || opType == OP_TYPE_MINUS){
-    double tmp = *((double *)colDataGetData(column, 0));
-    ASSERT_TRUE(tmp == exceptValue);
-    printf("result:%lf\n", tmp);
+    printf("1result:%f,except:%f\n", *((double *)colDataGetData(column, 0)), exceptValue);
+    ASSERT_TRUE(abs(*((double *)colDataGetData(column, 0)) - exceptValue) < 1e-15);
   }else if(opType == OP_TYPE_BIT_AND || opType == OP_TYPE_BIT_OR){
+    printf("2result:%ld,except:%f\n", *((int64_t *)colDataGetData(column, 0)), exceptValue);
     ASSERT_EQ(*((int64_t *)colDataGetData(column, 0)), exceptValue);
-    printf("result:%ld\n", *((int64_t *)colDataGetData(column, 0)));
   }else if(opType == OP_TYPE_GREATER_THAN || opType == OP_TYPE_GREATER_EQUAL || opType == OP_TYPE_LOWER_THAN ||
              opType == OP_TYPE_LOWER_EQUAL || opType == OP_TYPE_EQUAL || opType == OP_TYPE_NOT_EQUAL ||
              opType == OP_TYPE_IS_NULL || opType == OP_TYPE_IS_NOT_NULL || opType == OP_TYPE_IS_TRUE ||
              opType == OP_TYPE_LIKE || opType == OP_TYPE_NOT_LIKE || opType == OP_TYPE_MATCH || opType == OP_TYPE_NMATCH){
+    printf("3result:%d,except:%f\n", *((bool *)colDataGetData(column, 0)), exceptValue);
     ASSERT_EQ(*((bool *)colDataGetData(column, 0)), exceptValue);
-    printf("result:%d\n", *((bool *)colDataGetData(column, 0)));
   }
 
   taosArrayDestroyEx(blockList, scltFreeDataBlock);
@@ -1114,6 +1113,13 @@ TEST(columnTest, json_column_arith_op) {
   tdInitKVRowBuilder(&kvRowBuilder);
   parseJsontoTagData(rightv, &kvRowBuilder, NULL, 0);
   SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
+  char *tmp = (char *)taosMemoryRealloc(row, kvRowLen(row)+1);
+  if(tmp == NULL){
+    ASSERT_TRUE(0);
+  }
+  memmove(tmp+1, tmp, kvRowLen(tmp));
+  *tmp = TSDB_DATA_TYPE_JSON;
+  row = tmp;
 
   const int32_t len = 8;
   EOperatorType op[len] = {OP_TYPE_ADD, OP_TYPE_SUB, OP_TYPE_MULTI, OP_TYPE_DIV,
@@ -1166,6 +1172,9 @@ TEST(columnTest, json_column_arith_op) {
   for(int i = 0; i < len; i++){
     makeCalculate(row, key, TSDB_DATA_TYPE_INT, &input[i], eRes5[i], op[i]);
   }
+
+  tdDestroyKVRowBuilder(&kvRowBuilder);
+  taosMemoryFree(row);
 }
 
 void *prepareNchar(char* rightData){
@@ -1186,6 +1195,13 @@ TEST(columnTest, json_column_logic_op) {
   tdInitKVRowBuilder(&kvRowBuilder);
   parseJsontoTagData(rightv, &kvRowBuilder, NULL, 0);
   SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
+  char *tmp = (char *)taosMemoryRealloc(row, kvRowLen(row)+1);
+  if(tmp == NULL){
+    ASSERT_TRUE(0);
+  }
+  memmove(tmp+1, tmp, kvRowLen(tmp));
+  *tmp = TSDB_DATA_TYPE_JSON;
+  row = tmp;
 
   const int32_t len = 9;
   const int32_t len1 = 4;
@@ -1223,7 +1239,7 @@ TEST(columnTest, json_column_logic_op) {
   printf("--------------------json null---------------------\n");
 
   key = "k3";
-  double eRes2[len+len1] = {DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX, true, false, DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX};
+  bool eRes2[len+len1] = {false, false, false, false, false, false, true, false, false, false, false, false, false};
   for(int i = 0; i < len; i++){
     makeCalculate(row, key, TSDB_DATA_TYPE_INT, &input[i], eRes2[i], op[i]);
   }
@@ -1262,7 +1278,7 @@ TEST(columnTest, json_column_logic_op) {
   printf("--------------------json double---------------------\n");
 
   key = "k6";
-  bool eRes5[len+len1] = {true, false, false, false, false, true, false, true, true, false, false, false, true};
+  bool eRes5[len+len1] = {true, false, false, false, false, true, false, true, true, false, true, false, true};
   for(int i = 0; i < len; i++){
     makeCalculate(row, key, TSDB_DATA_TYPE_INT, &input[i], eRes5[i], op[i]);
   }
@@ -1275,7 +1291,7 @@ TEST(columnTest, json_column_logic_op) {
   printf("---------------------json not exist--------------------\n");
 
   key = "k10";
-  double eRes10[len+len1] = {DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX, true, false, DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX, DBL_MAX};
+  double eRes10[len+len1] = {false, false, false, false, false, false, true, false, false, false, false, false, false};
   for(int i = 0; i < len; i++){
     makeCalculate(row, key, TSDB_DATA_TYPE_INT, &input[i], eRes10[i], op[i]);
   }
@@ -1284,6 +1300,9 @@ TEST(columnTest, json_column_logic_op) {
     makeCalculate(row, key, TSDB_DATA_TYPE_NCHAR, rightData, eRes10[i], op[i]);
     taosMemoryFree(rightData);
   }
+
+  tdDestroyKVRowBuilder(&kvRowBuilder);
+  taosMemoryFree(row);
 }
 
 TEST(columnTest, smallint_value_add_int_column) {

source/libs/tdb/src/db/tdbPCache.c

@@ -14,6 +14,9 @@
  */
 #include "tdbInt.h"
 
+// #include <sys/types.h>
+// #include <unistd.h>
+
 struct SPCache {
   int         szPage;
   int         nPages;
@@ -32,7 +35,6 @@ static inline uint32_t tdbPCachePageHash(const SPgid *pPgid) {
   uint32_t *t = (uint32_t *)((pPgid)->fileid);
   return (uint32_t)(t[0] + t[1] + t[2] + t[3] + t[4] + t[5] + (pPgid)->pgno);
 }
-#define PAGE_IS_PINNED(pPage) ((pPage)->pLruNext == NULL)
 
 static int    tdbPCacheOpenImpl(SPCache *pCache);
 static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn);
@@ -80,16 +82,22 @@ int tdbPCacheClose(SPCache *pCache) {
 
 SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn) {
   SPage *pPage;
+  i32    nRef;
 
   tdbPCacheLock(pCache);
 
   pPage = tdbPCacheFetchImpl(pCache, pPgid, pTxn);
   if (pPage) {
-    tdbRefPage(pPage);
+    nRef = tdbRefPage(pPage);
   }
 
+  ASSERT(pPage);
+
   tdbPCacheUnlock(pCache);
 
+  // printf("thread %" PRId64 " fetch page %d pgno %d pPage %p nRef %d\n", taosGetSelfPthreadId(), pPage->id,
+  //        TDB_PAGE_PGNO(pPage), pPage, nRef);
+
   return pPage;
 }
 
@@ -98,16 +106,16 @@ void tdbPCacheRelease(SPCache *pCache, SPage *pPage, TXN *pTxn) {
 
   ASSERT(pTxn);
 
-  nRef = tdbUnrefPage(pPage);
-  ASSERT(nRef >= 0);
+  // nRef = tdbUnrefPage(pPage);
+  // ASSERT(nRef >= 0);
 
-  if (nRef == 0) {
   tdbPCacheLock(pCache);
-
+  nRef = tdbUnrefPage(pPage);
+  if (nRef == 0) {
     // test the nRef again to make sure
     // it is safe th handle the page
-    nRef = tdbGetPageRef(pPage);
-    if (nRef == 0) {
+    // nRef = tdbGetPageRef(pPage);
+    // if (nRef == 0) {
     if (pPage->isLocal) {
       tdbPCacheUnpinPage(pCache, pPage);
     } else {
@@ -118,10 +126,11 @@ void tdbPCacheRelease(SPCache *pCache, SPage *pPage, TXN *pTxn) {
 
       tdbPageDestroy(pPage, pTxn->xFree, pTxn->xArg);
     }
+    // }
   }
-
   tdbPCacheUnlock(pCache);
-  }
+  // printf("thread %" PRId64 " relas page %d pgno %d pPage %p nRef %d\n", taosGetSelfPthreadId(), pPage->id,
+  //        TDB_PAGE_PGNO(pPage), pPage, nRef);
 }
 
 int tdbPCacheGetPageSize(SPCache *pCache) { return pCache->szPage; }
@@ -223,6 +232,7 @@ static void tdbPCachePinPage(SPCache *pCache, SPage *pPage) {
 
     pCache->nRecyclable--;
 
+    // printf("pin page %d pgno %d pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
     tdbTrace("pin page %d", pPage->id);
   }
 }
@@ -243,6 +253,7 @@ static void tdbPCacheUnpinPage(SPCache *pCache, SPage *pPage) {
 
   pCache->nRecyclable++;
 
+  // printf("unpin page %d pgno %d pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
   tdbTrace("unpin page %d", pPage->id);
 }
 
@@ -253,10 +264,12 @@ static void tdbPCacheRemovePageFromHash(SPCache *pCache, SPage *pPage) {
   h = tdbPCachePageHash(&(pPage->pgid));
   for (ppPage = &(pCache->pgHash[h % pCache->nHash]); (*ppPage) && *ppPage != pPage; ppPage = &((*ppPage)->pHashNext))
     ;
-  ASSERT(*ppPage == pPage);
-  *ppPage = pPage->pHashNext;
 
+  if (*ppPage) {
+    *ppPage = pPage->pHashNext;
     pCache->nPage--;
+    // printf("rmv page %d to hash, pgno %d, pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
+  }
 
   tdbTrace("remove page %d to hash", pPage->id);
 }
@@ -271,6 +284,7 @@ static void tdbPCacheAddPageToHash(SPCache *pCache, SPage *pPage) {
 
   pCache->nPage++;
 
+  // printf("add page %d to hash, pgno %d, pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
   tdbTrace("add page %d to hash", pPage->id);
 }
 

source/libs/tdb/src/db/tdbPager.c

@@ -265,6 +265,7 @@ int tdbPagerFetchPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPa
   pgid.pgno = pgno;
   pPage = tdbPCacheFetch(pPager->pCache, &pgid, pTxn);
   if (pPage == NULL) {
+    ASSERT(0);
     return -1;
   }
 
@@ -272,10 +273,14 @@ int tdbPagerFetchPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPa
   if (!TDB_PAGE_INITIALIZED(pPage)) {
     ret = tdbPagerInitPage(pPager, pPage, initPage, arg, loadPage);
     if (ret < 0) {
+      ASSERT(0);
       return -1;
     }
   }
 
+  // printf("thread %" PRId64 " pager fetch page %d pgno %d ppage %p\n", taosGetSelfPthreadId(), pPage->id,
+  //        TDB_PAGE_PGNO(pPage), pPage);
+
   ASSERT(TDB_PAGE_INITIALIZED(pPage));
   ASSERT(pPage->pPager == pPager);
 
@@ -284,7 +289,11 @@ int tdbPagerFetchPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPa
   return 0;
 }
 
-void tdbPagerReturnPage(SPager *pPager, SPage *pPage, TXN *pTxn) { tdbPCacheRelease(pPager->pCache, pPage, pTxn); }
+void tdbPagerReturnPage(SPager *pPager, SPage *pPage, TXN *pTxn) {
+  tdbPCacheRelease(pPager->pCache, pPage, pTxn);
+  // printf("thread %" PRId64 " pager retun page %d pgno %d ppage %p\n", taosGetSelfPthreadId(), pPage->id,
+  //        TDB_PAGE_PGNO(pPage), pPage);
+}
 
 static int tdbPagerAllocFreePage(SPager *pPager, SPgno *ppgno) {
   // TODO: Allocate a page from the free list
@@ -352,6 +361,7 @@ static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage
 
     ret = (*initPage)(pPage, arg, init);
     if (ret < 0) {
+      ASSERT(0);
       TDB_UNLOCK_PAGE(pPage);
       return -1;
     }
@@ -370,6 +380,7 @@ static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage
       }
     }
   } else {
+    ASSERT(0);
     return -1;
   }
 

source/libs/tdb/src/inc/tdbInt.h

@@ -275,15 +275,15 @@ static inline i32 tdbUnrefPage(SPage *pPage) {
 #define P_LOCK_FAIL -1
 
 static inline int tdbTryLockPage(tdb_spinlock_t *pLock) {
-  int ret;
-  if (tdbSpinlockTrylock(pLock) == 0) {
-    ret = P_LOCK_SUCC;
-  } else if (errno == EBUSY) {
-    ret = P_LOCK_BUSY;
+  int ret = tdbSpinlockTrylock(pLock);
+  if (ret == 0) {
+    return P_LOCK_SUCC;
+  } else if (ret == EBUSY) {
+    return P_LOCK_BUSY;
   } else {
-    ret = P_LOCK_FAIL;
+    ASSERT(0);
+    return P_LOCK_FAIL;
   }
-  return ret;
 }
 
 #define TDB_INIT_PAGE_LOCK(pPage)    tdbSpinlockInit(&((pPage)->lock), 0)

source/libs/tdb/test/tdbTest.cpp

@@ -4,6 +4,7 @@
 #include "os.h"
 #include "tdb.h"
 
+#include <shared_mutex>
 #include <string>
 #include <thread>
 #include <vector>
@@ -118,7 +119,7 @@ static int tDefaultKeyCmpr(const void *pKey1, int keyLen1, const void *pKey2, in
   return cret;
 }
 
-TEST(tdb_test, simple_insert1) {
+TEST(tdb_test, DISABLED_simple_insert1) {
   int           ret;
   TDB          *pEnv;
   TTB          *pDb;
@@ -238,7 +239,7 @@ TEST(tdb_test, simple_insert1) {
   GTEST_ASSERT_EQ(ret, 0);
 }
 
-TEST(tdb_test, simple_insert2) {
+TEST(tdb_test, DISABLED_simple_insert2) {
   int           ret;
   TDB          *pEnv;
   TTB          *pDb;
@@ -325,7 +326,7 @@ TEST(tdb_test, simple_insert2) {
   GTEST_ASSERT_EQ(ret, 0);
 }
 
-TEST(tdb_test, simple_delete1) {
+TEST(tdb_test, DISABLED_simple_delete1) {
   int       ret;
   TTB      *pDb;
   char      key[128];
@@ -420,7 +421,7 @@ TEST(tdb_test, simple_delete1) {
   tdbClose(pEnv);
 }
 
-TEST(tdb_test, simple_upsert1) {
+TEST(tdb_test, DISABLED_simple_upsert1) {
   int       ret;
   TDB      *pEnv;
   TTB      *pDb;
@@ -490,13 +491,13 @@ TEST(tdb_test, multi_thread_query) {
   TDB          *pEnv;
   TTB          *pDb;
   tdb_cmpr_fn_t compFunc;
-  int           nData = 20000;
+  int           nData = 1000000;
   TXN           txn;
 
   taosRemoveDir("tdb");
 
   // Open Env
-  ret = tdbOpen("tdb", 512, 1, &pEnv);
+  ret = tdbOpen("tdb", 4096, 10, &pEnv);
   GTEST_ASSERT_EQ(ret, 0);
 
   // Create a database
@@ -506,7 +507,7 @@ TEST(tdb_test, multi_thread_query) {
 
   char      key[64];
   char      val[64];
-  int64_t   poolLimit = 4096;  // 1M pool limit
+  int64_t   poolLimit = 4096 * 20;  // 1M pool limit
   int64_t   txnid = 0;
   SPoolMem *pPool;
 
@@ -598,3 +599,131 @@ TEST(tdb_test, multi_thread_query) {
   ret = tdbClose(pEnv);
   GTEST_ASSERT_EQ(ret, 0);
 }
+
+TEST(tdb_test, DISABLED_multi_thread1) {
+#if 0
+  int           ret;
+  TDB          *pDb;
+  TTB          *pTb;
+  tdb_cmpr_fn_t compFunc;
+  int           nData = 10000000;
+  TXN           txn;
+
+  std::shared_timed_mutex mutex;
+
+  taosRemoveDir("tdb");
+
+  // Open Env
+  ret = tdbOpen("tdb", 512, 1, &pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  ret = tdbTbOpen("db.db", -1, -1, NULL, pDb, &pTb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  auto insert = [](TDB *pDb, TTB *pTb, int nData, int *stop, std::shared_timed_mutex *mu) {
+    TXN       txn = {0};
+    char      key[128];
+    char      val[128];
+    SPoolMem *pPool = openPool();
+
+    txn.flags = TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED;
+    txn.txnId = -1;
+    txn.xMalloc = poolMalloc;
+    txn.xFree = poolFree;
+    txn.xArg = pPool;
+    tdbBegin(pDb, &txn);
+    for (int iData = 1; iData <= nData; iData++) {
+      sprintf(key, "key%d", iData);
+      sprintf(val, "value%d", iData);
+      {
+        std::lock_guard<std::shared_timed_mutex> wmutex(*mu);
+
+        int ret = tdbTbInsert(pTb, key, strlen(key), val, strlen(val), &txn);
+
+        GTEST_ASSERT_EQ(ret, 0);
+      }
+
+      if (pPool->size > 1024 * 1024) {
+        tdbCommit(pDb, &txn);
+
+        clearPool(pPool);
+        tdbBegin(pDb, &txn);
+      }
+    }
+
+    tdbCommit(pDb, &txn);
+    closePool(pPool);
+
+    *stop = 1;
+  };
+
+  auto query = [](TTB *pTb, int *stop, std::shared_timed_mutex *mu) {
+    TBC  *pDBC;
+    void *pKey = NULL;
+    void *pVal = NULL;
+    int   vLen, kLen;
+    int   ret;
+    TXN   txn;
+
+    SPoolMem *pPool = openPool();
+    txn.flags = 0;
+    txn.txnId = 0;
+    txn.xMalloc = poolMalloc;
+    txn.xFree = poolFree;
+    txn.xArg = pPool;
+
+    for (;;) {
+      if (*stop) break;
+
+      clearPool(pPool);
+      int count = 0;
+      {
+        std::shared_lock<std::shared_timed_mutex> rMutex(*mu);
+
+        ret = tdbTbcOpen(pTb, &pDBC, &txn);
+        GTEST_ASSERT_EQ(ret, 0);
+
+        tdbTbcMoveToFirst(pDBC);
+
+        for (;;) {
+          ret = tdbTbcNext(pDBC, &pKey, &kLen, &pVal, &vLen);
+          if (ret < 0) break;
+          count++;
+        }
+
+        std::cout << count << std::endl;
+
+        tdbTbcClose(pDBC);
+      }
+
+      usleep(500000);
+    }
+
+    closePool(pPool);
+    tdbFree(pKey);
+    tdbFree(pVal);
+  };
+
+  std::vector<std::thread> threads;
+  int                      nThreads = 10;
+  int                      stop = 0;
+  for (int i = 0; i < nThreads; i++) {
+    if (i == 0) {
+      threads.push_back(std::thread(insert, pDb, pTb, nData, &stop, &mutex));
+    } else {
+      threads.push_back(std::thread(query, pTb, &stop, &mutex));
+    }
+  }
+
+  for (auto &th : threads) {
+    th.join();
+  }
+
+  // Close a database
+  tdbTbClose(pTb);
+
+  // Close Env
+  ret = tdbClose(pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+#endif
+}
\ No newline at end of file

source/util/src/tqueue.c

@@ -162,7 +162,7 @@ void *taosAllocateQitem(int32_t size, EQItype itype) {
     uTrace("item:%p, node:%p is allocated", pNode->item, pNode);
   }
 
-  return (void *)pNode->item;
+  return pNode->item;
 }
 
 void taosFreeQitem(void *pItem) {

tests/script/tsim/stable/add_column.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sql connect
+
+print ========== prepare stb and ctb
+sql create database db vgroups 1
+sql create table db.stb (ts timestamp, c1 int, c2 binary(4)) tags(t1 int, t2 float, t3 binary(16)) comment "abd"
+sql create table db.ctb using db.stb tags(1, 2, "3")
+sql insert into db.ctb values(now, 1, "2")
+
+sql show db.stables
+if $rows != 1 then
+  return -1
+endi
+if $data[0][0] != stb then
+  return -1
+endi
+if $data[0][1] != db then
+  return -1
+endi
+if $data[0][3] != 3 then
+  return -1
+endi
+if $data[0][4] != 3 then
+  return -1
+endi
+if $data[0][6] != abd then
+  return -1
+endi
+
+sql show db.tables
+if $rows != 1 then
+  return -1
+endi
+if $data[0][0] != ctb then
+  return -1
+endi
+if $data[0][1] != db then
+  return -1
+endi
+if $data[0][3] != 3 then
+  return -1
+endi
+if $data[0][4] != stb then
+  return -1
+endi
+if $data[0][6] != 2 then
+  return -1
+endi
+if $data[0][9] != CHILD_TABLE then
+  return -1
+endi
+
+sql select * from db.stb
+if $rows != 1 then
+  return -1
+endi
+if $data[0][1] != 1 then
+  return -1
+endi
+if $data[0][2] != 2 then
+  return -1
+endi
+if $data[0][3] != 1 then
+  return -1
+endi
+
+print ========== add column
+sql alter table db.stb add column c3 int
+sql alter table db.stb add column c4 bigint
+sql alter table db.stb add column c5 binary(12)
+
+sql show db.stables
+if $data[0][3] != 6 then
+  return -1
+endi
+
+sql show db.tables
+if $data[0][3] != 6 then
+  return -1
+endi
+
+sql select * from db.stb
+print $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] $data[0][5] $data[0][6]
+if $rows != 1 then
+  return -1
+endi
+if $data[0][1] != 1 then
+  return -1
+endi
+if $data[0][2] != 2 then
+  return -1
+endi
+if $data[0][3] != NULL then
+  return -1
+endi
+if $data[0][4] != NULL then
+  return -1
+endi
+if $data[0][5] != NULL then
+  return -1
+endi
+if $data[0][6] != 1 then
+  return -1
+endi
+

tests/script/tsim/testsuit.sim

+#run  user/pass_alter.sim
+#run  user/basic1.sim
+#run  user/privilege2.sim
+#run  user/user_len.sim
+#run  user/privilege1.sim
+#run  user/pass_len.sim
+#run  tstream/basic1.sim
+#run  tstream/basic0.sim
+#run  table/basic1.sim
+#run  trans/create_db.sim
+#run  stable/alter1.sim
+#run  stable/vnode3.sim
+#run  stable/metrics.sim
+#run  stable/show.sim
+#run  stable/values.sim
+#run  stable/dnode3.sim
+#run  stable/refcount.sim
+#run  stable/disk.sim
+#run  db/basic1.sim
+#run  db/basic3.sim
+#run  db/basic7.sim
+#run  db/basic6.sim
+#run  db/create_all_options.sim
+#run  db/basic2.sim
+#run  db/error1.sim
+#run  db/taosdlog.sim
+#run  db/alter_option.sim
+#run  mnode/basic1.sim
+#run  parser/fourArithmetic-basic.sim
+#run  parser/groupby-basic.sim
+#run  snode/basic1.sim
+#run  query/time_process.sim
+#run  query/stddev.sim
+#run  query/interval-offset.sim
+#run  query/charScalarFunction.sim
+#run  query/complex_select.sim
+#run  query/explain.sim
+#run  query/crash_sql.sim
+#run  query/diff.sim
+#run  query/complex_limit.sim
+#run  query/complex_having.sim
+#run  query/udf.sim
+#run  query/complex_group.sim
+#run  query/interval.sim
+#run  query/session.sim
+
+print ========> dead lock failed when 2 rows in outputCapacity
+run  query/scalarFunction.sim
+run  query/scalarNull.sim
+run  query/complex_where.sim
+run  tmq/basic1.sim
+run  tmq/basic4.sim
+run  tmq/basic1Of2Cons.sim
+run  tmq/prepareBasicEnv-1vgrp.sim
+run  tmq/topic.sim
+run  tmq/basic4Of2Cons.sim
+run  tmq/prepareBasicEnv-4vgrp.sim
+run  tmq/basic3.sim
+run  tmq/basic2Of2Cons.sim
+run  tmq/basic2.sim
+run  tmq/basic3Of2Cons.sim
+run  tmq/basic2Of2ConsOverlap.sim
+run  tmq/clearConsume.sim
+run  qnode/basic1.sim
+run  dnode/basic1.sim
+run  show/basic.sim
+run  insert/basic1.sim
+run  insert/basic0.sim
+run  insert/backquote.sim
+run  insert/null.sim
+run  sync/oneReplica1VgElectWithInsert.sim
+run  sync/threeReplica1VgElect.sim
+run  sync/oneReplica1VgElect.sim
+run  sync/insertDataByRunBack.sim
+run  sync/threeReplica1VgElectWihtInsert.sim
+run  sma/tsmaCreateInsertData.sim
+run  sma/rsmaCreateInsertQuery.sim
+run  valgrind/checkError.sim
+run  bnode/basic1.sim

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

@@ -1079,6 +1079,291 @@ class TDTestCase:
 
         tdLog.printNoPrefix("======== test case 10 end ...... ")
 
+    def tmqCase11(self, cfgPath, buildPath):
+        tdLog.printNoPrefix("======== test case 11: ")
+        
+        self.initConsumerTable()
+
+        # create and start thread
+        parameterDict = {'cfg':        '',       \
+                         'actionType': 0,        \
+                         'dbName':     'db11',    \
+                         'dropFlag':   1,        \
+                         'vgroups':    4,        \
+                         'replica':    1,        \
+                         'stbName':    'stb1',   \
+                         'ctbNum':     10,       \
+                         'rowsPerTbl': 10000,    \
+                         'batchNum':   100,      \
+                         'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
+        parameterDict['cfg'] = cfgPath
+        
+        self.create_database(tdSql, parameterDict["dbName"])
+        self.create_stable(tdSql, parameterDict["dbName"], parameterDict["stbName"])
+        self.create_ctables(tdSql, parameterDict["dbName"], parameterDict["stbName"], parameterDict["ctbNum"])
+        self.insert_data(tdSql,\
+                         parameterDict["dbName"],\
+                         parameterDict["stbName"],\
+                         parameterDict["ctbNum"],\
+                         parameterDict["rowsPerTbl"],\
+                         parameterDict["batchNum"])
+
+        tdLog.info("create topics from stb1")
+        topicFromStb1 = 'topic_stb1'
+        
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(topicFromStb1, parameterDict['dbName'], parameterDict['stbName']))
+        consumerId     = 0
+        expectrowcnt   = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
+        topicList      = topicFromStb1
+        ifcheckdata    = 0
+        ifManualCommit = 1
+        keyList        = 'group.id:cgrp1,\
+                        enable.auto.commit:false,\
+                        auto.commit.interval.ms:6000,\
+                        auto.offset.reset:none'
+        self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("start consume processor")
+        pollDelay = 5
+        showMsg   = 1
+        showRow   = 1
+        self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
+
+        tdLog.info("start to check consume result")
+        expectRows = 1
+        resultList = self.selectConsumeResult(expectRows)
+        totalConsumeRows = 0
+        for i in range(expectRows):
+            totalConsumeRows += resultList[i]
+        
+        if totalConsumeRows != 0:
+            tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, 0))
+            tdLog.exit("tmq consume rows error!")
+
+        self.initConsumerInfoTable()
+        consumerId = 1
+        keyList    = 'group.id:cgrp1,\
+                      enable.auto.commit:false,\
+                      auto.commit.interval.ms:6000,\
+                      auto.offset.reset:none'
+        self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("again start consume processor")
+        self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
+
+        tdLog.info("again check consume result")
+        expectRows = 2
+        resultList = self.selectConsumeResult(expectRows)
+        totalConsumeRows = 0
+        for i in range(expectRows):
+            totalConsumeRows += resultList[i]
+        
+        if totalConsumeRows != 0:
+            tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, 0))
+            tdLog.exit("tmq consume rows error!")
+
+        tdSql.query("drop topic %s"%topicFromStb1)
+
+        tdLog.printNoPrefix("======== test case 11 end ...... ")
+
+    def tmqCase12(self, cfgPath, buildPath):
+        tdLog.printNoPrefix("======== test case 12: ")
+        
+        self.initConsumerTable()
+
+        # create and start thread
+        parameterDict = {'cfg':        '',       \
+                         'actionType': 0,        \
+                         'dbName':     'db12',    \
+                         'dropFlag':   1,        \
+                         'vgroups':    4,        \
+                         'replica':    1,        \
+                         'stbName':    'stb1',   \
+                         'ctbNum':     10,       \
+                         'rowsPerTbl': 10000,    \
+                         'batchNum':   100,      \
+                         'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
+        parameterDict['cfg'] = cfgPath
+        
+        self.create_database(tdSql, parameterDict["dbName"])
+        self.create_stable(tdSql, parameterDict["dbName"], parameterDict["stbName"])
+        self.create_ctables(tdSql, parameterDict["dbName"], parameterDict["stbName"], parameterDict["ctbNum"])
+        self.insert_data(tdSql,\
+                         parameterDict["dbName"],\
+                         parameterDict["stbName"],\
+                         parameterDict["ctbNum"],\
+                         parameterDict["rowsPerTbl"],\
+                         parameterDict["batchNum"])
+
+        tdLog.info("create topics from stb1")
+        topicFromStb1 = 'topic_stb1'
+        
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(topicFromStb1, parameterDict['dbName'], parameterDict['stbName']))
+        consumerId     = 0
+        expectrowcnt   = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
+        topicList      = topicFromStb1
+        ifcheckdata    = 0
+        ifManualCommit = 0
+        keyList        = 'group.id:cgrp1,\
+                        enable.auto.commit:false,\
+                        auto.commit.interval.ms:6000,\
+                        auto.offset.reset:earliest'
+        self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("start consume processor")
+        pollDelay = 5
+        showMsg   = 1
+        showRow   = 1
+        self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
+
+        tdLog.info("start to check consume result")
+        expectRows = 1
+        resultList = self.selectConsumeResult(expectRows)
+        totalConsumeRows = 0
+        for i in range(expectRows):
+            totalConsumeRows += resultList[i]
+        
+        if totalConsumeRows != expectrowcnt/4:
+            tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4))
+            tdLog.exit("tmq consume rows error!")
+
+        self.initConsumerInfoTable()
+        consumerId = 1
+        keyList    = 'group.id:cgrp1,\
+                      enable.auto.commit:false,\
+                      auto.commit.interval.ms:6000,\
+                      auto.offset.reset:none'
+        self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("again start consume processor")
+        self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
+
+        tdLog.info("again check consume result")
+        expectRows = 2
+        resultList = self.selectConsumeResult(expectRows)
+        totalConsumeRows = 0
+        for i in range(expectRows):
+            totalConsumeRows += resultList[i]
+        
+        if totalConsumeRows != expectrowcnt/4:
+            tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4))
+            tdLog.exit("tmq consume rows error!")
+
+        tdSql.query("drop topic %s"%topicFromStb1)
+
+        tdLog.printNoPrefix("======== test case 12 end ...... ")
+
+    def tmqCase13(self, cfgPath, buildPath):
+        tdLog.printNoPrefix("======== test case 13: ")
+        
+        self.initConsumerTable()
+
+        # create and start thread
+        parameterDict = {'cfg':        '',       \
+                         'actionType': 0,        \
+                         'dbName':     'db13',    \
+                         'dropFlag':   1,        \
+                         'vgroups':    4,        \
+                         'replica':    1,        \
+                         'stbName':    'stb1',   \
+                         'ctbNum':     10,       \
+                         'rowsPerTbl': 10000,    \
+                         'batchNum':   100,      \
+                         'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
+        parameterDict['cfg'] = cfgPath
+        
+        self.create_database(tdSql, parameterDict["dbName"])
+        self.create_stable(tdSql, parameterDict["dbName"], parameterDict["stbName"])
+        self.create_ctables(tdSql, parameterDict["dbName"], parameterDict["stbName"], parameterDict["ctbNum"])
+        self.insert_data(tdSql,\
+                         parameterDict["dbName"],\
+                         parameterDict["stbName"],\
+                         parameterDict["ctbNum"],\
+                         parameterDict["rowsPerTbl"],\
+                         parameterDict["batchNum"])
+
+        tdLog.info("create topics from stb1")
+        topicFromStb1 = 'topic_stb1'
+        
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(topicFromStb1, parameterDict['dbName'], parameterDict['stbName']))
+        consumerId     = 0
+        expectrowcnt   = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
+        topicList      = topicFromStb1
+        ifcheckdata    = 0
+        ifManualCommit = 1
+        keyList        = 'group.id:cgrp1,\
+                        enable.auto.commit:false,\
+                        auto.commit.interval.ms:6000,\
+                        auto.offset.reset:earliest'
+        self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("start consume processor")
+        pollDelay = 5
+        showMsg   = 1
+        showRow   = 1
+        self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
+
+        tdLog.info("start to check consume result")
+        expectRows = 1
+        resultList = self.selectConsumeResult(expectRows)
+        totalConsumeRows = 0
+        for i in range(expectRows):
+            totalConsumeRows += resultList[i]
+        
+        if totalConsumeRows != expectrowcnt/4:
+            tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4))
+            tdLog.exit("tmq consume rows error!")
+
+        self.initConsumerInfoTable()
+        consumerId     = 1
+        ifManualCommit = 1
+        keyList        = 'group.id:cgrp1,\
+                          enable.auto.commit:false,\
+                          auto.commit.interval.ms:6000,\
+                          auto.offset.reset:none'
+        self.insertConsumerInfo(consumerId, expectrowcnt/2,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("again start consume processor")
+        self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
+
+        tdLog.info("again check consume result")
+        expectRows = 2
+        resultList = self.selectConsumeResult(expectRows)
+        totalConsumeRows = 0
+        for i in range(expectRows):
+            totalConsumeRows += resultList[i]
+        
+        if totalConsumeRows != expectrowcnt*(1/2+1/4):
+            tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt*(1/2+1/4)))
+            tdLog.exit("tmq consume rows error!")
+
+        self.initConsumerInfoTable()
+        consumerId     = 2
+        ifManualCommit = 1
+        keyList        = 'group.id:cgrp1,\
+                          enable.auto.commit:false,\
+                          auto.commit.interval.ms:6000,\
+                          auto.offset.reset:none'
+        self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("again start consume processor")
+        self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
+
+        tdLog.info("again check consume result")
+        expectRows = 3
+        resultList = self.selectConsumeResult(expectRows)
+        totalConsumeRows = 0
+        for i in range(expectRows):
+            totalConsumeRows += resultList[i]
+        
+        if totalConsumeRows != expectrowcnt:
+            tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt))
+            tdLog.exit("tmq consume rows error!")
+
+        tdSql.query("drop topic %s"%topicFromStb1)
+
+        tdLog.printNoPrefix("======== test case 13 end ...... ")
+
     def run(self):
         tdSql.prepare()
 
@@ -1100,7 +1385,9 @@ class TDTestCase:
         self.tmqCase8(cfgPath, buildPath)
         self.tmqCase9(cfgPath, buildPath)
         self.tmqCase10(cfgPath, buildPath)        
-
+        self.tmqCase11(cfgPath, buildPath)
+        self.tmqCase12(cfgPath, buildPath)
+        self.tmqCase13(cfgPath, buildPath)
 
     def stop(self):
         tdSql.close()

tools/shell/src/shellEngine.c

@@ -315,6 +315,7 @@ void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *field, i
       break;
     case TSDB_DATA_TYPE_BINARY:
     case TSDB_DATA_TYPE_NCHAR:
+    case TSDB_DATA_TYPE_JSON:
       memcpy(buf, val, length);
       buf[length] = 0;
       taosFprintfFile(pFile, "\'%s\'", buf);
@@ -384,19 +385,25 @@ void shellPrintNChar(const char *str, int32_t length, int32_t width) {
   while (pos < length) {
     TdWchar wc;
     int32_t bytes = taosMbToWchar(&wc, str + pos, MB_CUR_MAX);
-    if (bytes == 0) {
+    if (bytes <= 0) {
       break;
     }
-    pos += bytes;
-    if (pos > length) {
+
+    if (pos + bytes > length) {
       break;
     }
-
+    int w = 0;
 #ifdef WINDOWS
-    int32_t w = bytes;
+    w = bytes;
 #else
-    int32_t w = taosWcharWidth(wc);
+    if(*(str + pos) == '\t' || *(str + pos) == '\n' || *(str + pos) == '\r'){
+      w = bytes;
+    }else{
+      w = taosWcharWidth(wc);
+    }
 #endif
+    pos += bytes;
+
     if (w <= 0) {
       continue;
     }
@@ -496,6 +503,7 @@ void shellPrintField(const char *val, TAOS_FIELD *field, int32_t width, int32_t
       break;
     case TSDB_DATA_TYPE_BINARY:
     case TSDB_DATA_TYPE_NCHAR:
+    case TSDB_DATA_TYPE_JSON:
       shellPrintNChar(val, length, width);
       break;
     case TSDB_DATA_TYPE_TIMESTAMP:
@@ -604,7 +612,6 @@ int32_t shellCalcColWidth(TAOS_FIELD *field, int32_t precision) {
     case TSDB_DATA_TYPE_DOUBLE:
       return TMAX(25, width);
 
-    case TSDB_DATA_TYPE_JSON:
     case TSDB_DATA_TYPE_BINARY:
       if (field->bytes > shell.args.displayWidth) {
         return TMAX(shell.args.displayWidth, width);
@@ -612,7 +619,8 @@ int32_t shellCalcColWidth(TAOS_FIELD *field, int32_t precision) {
         return TMAX(field->bytes, width);
       }
 
-    case TSDB_DATA_TYPE_NCHAR: {
+    case TSDB_DATA_TYPE_NCHAR:
+    case TSDB_DATA_TYPE_JSON: {
       int16_t bytes = field->bytes * TSDB_NCHAR_SIZE;
       if (bytes > shell.args.displayWidth) {
         return TMAX(shell.args.displayWidth, width);