Merge branch '3.0' of https://github.com/taosdata/TDengine into refact/tsdb_optimize

cmake/taostools_CMakeLists.txt.in

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

docs/en/12-taos-sql/03-table.md

@@ -57,7 +57,7 @@ table_option: {
 3. MAX_DELAY: specifies the maximum latency for pushing computation results. The default value is 15 minutes or the value of the INTERVAL parameter, whichever is smaller. Enter a value between 0 and 15 minutes in milliseconds, seconds, or minutes. You can enter multiple values separated by commas (,). Note: Retain the default value if possible. Configuring a small MAX_DELAY may cause results to be frequently pushed, affecting storage and query performance. This parameter applies only to supertables and takes effect only when the RETENTIONS parameter has been specified for the database.
 4. ROLLUP: specifies aggregate functions to roll up. Rolling up a function provides downsampled results based on multiple axes. This parameter applies only to supertables and takes effect only when the RETENTIONS parameter has been specified for the database. You can specify only one function to roll up. The rollup takes effect on all columns except TS. Enter one of the following values: avg, sum, min, max, last, or first.
 5. SMA: specifies functions on which to enable small materialized aggregates (SMA). SMA is user-defined precomputation of aggregates based on data blocks. Enter one of the following values: max, min, or sum This parameter can be used with supertables and standard tables.
-6. TTL: specifies the time to live (TTL) for the table. If the period specified by the TTL parameter elapses without any data being written to the table, TDengine will automatically delete the table. Note: The system may not delete the table at the exact moment that the TTL expires. Enter a value in days. The default value is 0. Note: The TTL parameter has a higher priority than the KEEP parameter. If a table is marked for deletion because the TTL has expired, it will be deleted even if the time specified by the KEEP parameter has not elapsed. This parameter can be used with standard tables and subtables.
+6. TTL: specifies the time to live (TTL) for the table. If TTL is specified when creatinga table, after the time period for which the table has been existing is over TTL, TDengine will automatically delete the table. Please be noted that the system may not delete the table at the exact moment that the TTL expires but guarantee there is such a system and finally the table will be deleted. The unit of TTL is in days. The default value is 0, i.e. never expire. 
 
 ## Create Subtables
 

docs/en/13-operation/03-tolerance.md

@@ -27,4 +27,4 @@ The number of dnodes in a TDengine cluster must NOT be lower than the number of
 
 As long as the dnodes of a TDengine cluster are deployed on different physical machines and the replica number is higher than 1, high availability can be achieved without any other assistance. For disaster recovery, dnodes of a TDengine cluster should be deployed in geographically different data centers.
 
-Alternatively, you can use taosX to synchronize the data from one TDengine cluster to another cluster in a remote location. For more information, see [taosX](../../reference/taosX).
+Alternatively, you can use taosX to synchronize the data from one TDengine cluster to another cluster in a remote location. However, taosX is only available in TDengine enterprise version, for more information please contact tdengine.com. 

docs/zh/12-taos-sql/03-table.md

@@ -10,27 +10,27 @@ description: 对表的各种管理操作
 
 ```sql
 CREATE TABLE [IF NOT EXISTS] [db_name.]tb_name (create_definition [, create_definitionn] ...) [table_options]
- 
+
 CREATE TABLE create_subtable_clause
- 
+
 CREATE TABLE [IF NOT EXISTS] [db_name.]tb_name (create_definition [, create_definitionn] ...)
     [TAGS (create_definition [, create_definitionn] ...)]
     [table_options]
- 
+
 create_subtable_clause: {
     create_subtable_clause [create_subtable_clause] ...
   | [IF NOT EXISTS] [db_name.]tb_name USING [db_name.]stb_name [(tag_name [, tag_name] ...)] TAGS (tag_value [, tag_value] ...)
 }
- 
+
 create_definition:
     col_name column_definition
- 
+
 column_definition:
     type_name [comment 'string_value']
- 
+
 table_options:
     table_option ...
- 
+
 table_option: {
     COMMENT 'string_value'
   | WATERMARK duration[,duration]
@@ -54,12 +54,13 @@ table_option: {
    需要注意的是转义字符中的内容必须是可打印字符。
 
 **参数说明**
+
 1. COMMENT：表注释。可用于超级表、子表和普通表。
-2. WATERMARK：指定窗口的关闭时间，默认值为 5 秒，最小单位毫秒，范围为0到15分钟，多个以逗号分隔。只可用于超级表，且只有当数据库使用了RETENTIONS参数时，才可以使用此表参数。
-3. MAX_DELAY：用于控制推送计算结果的最大延迟，默认值为 interval 的值(但不能超过最大值)，最小单位毫秒，范围为1毫秒到15分钟，多个以逗号分隔。注：不建议 MAX_DELAY 设置太小，否则会过于频繁的推送结果，影响存储和查询性能，如无特殊需求，取默认值即可。只可用于超级表，且只有当数据库使用了RETENTIONS参数时，才可以使用此表参数。
-4. ROLLUP：Rollup 指定的聚合函数，提供基于多层级的降采样聚合结果。只可用于超级表。只有当数据库使用了RETENTIONS参数时，才可以使用此表参数。作用于超级表除TS列外的其它所有列，但是只能定义一个聚合函数。 聚合函数支持 avg, sum, min, max, last, first。
-5. SMA：Small Materialized Aggregates，提供基于数据块的自定义预计算功能。预计算类型包括MAX、MIN和SUM。可用于超级表/普通表。
-6. TTL：Time to Live，是用户用来指定表的生命周期的参数。如果在持续的TTL时间内，都没有数据写入该表，则TDengine系统会自动删除该表。这个TTL的时间只是一个大概时间，我们系统不保证到了时间一定会将其删除，而只保证存在这样一个机制。TTL单位是天，默认为0，表示不限制。用户需要注意，TTL优先级高于KEEP，即TTL时间满足删除机制时，即使当前数据的存在时间小于KEEP，此表也会被删除。只可用于子表和普通表。
+2. WATERMARK：指定窗口的关闭时间，默认值为 5 秒，最小单位毫秒，范围为 0 到 15 分钟，多个以逗号分隔。只可用于超级表，且只有当数据库使用了 RETENTIONS 参数时，才可以使用此表参数。
+3. MAX_DELAY：用于控制推送计算结果的最大延迟，默认值为 interval 的值(但不能超过最大值)，最小单位毫秒，范围为 1 毫秒到 15 分钟，多个以逗号分隔。注：不建议 MAX_DELAY 设置太小，否则会过于频繁的推送结果，影响存储和查询性能，如无特殊需求，取默认值即可。只可用于超级表，且只有当数据库使用了 RETENTIONS 参数时，才可以使用此表参数。
+4. ROLLUP：Rollup 指定的聚合函数，提供基于多层级的降采样聚合结果。只可用于超级表。只有当数据库使用了 RETENTIONS 参数时，才可以使用此表参数。作用于超级表除 TS 列外的其它所有列，但是只能定义一个聚合函数。 聚合函数支持 avg, sum, min, max, last, first。
+5. SMA：Small Materialized Aggregates，提供基于数据块的自定义预计算功能。预计算类型包括 MAX、MIN 和 SUM。可用于超级表/普通表。
+6. TTL：Time to Live，是用户用来指定表的生命周期的参数。如果创建表时指定了这个参数，当该表的存在时间超过 TTL 指定的时间后，TDengine 自动删除该表。这个 TTL 的时间只是一个大概时间，系统不保证到了时间一定会将其删除，而只保证存在这样一个机制且最终一定会删除。TTL 单位是天，默认为 0，表示不限制，到期时间为表创建时间加上 TTL 时间。
 
 ## 创建子表
 
@@ -89,7 +90,7 @@ CREATE TABLE [IF NOT EXISTS] tb_name1 USING stb_name TAGS (tag_value1, ...) [IF
 
 ```sql
 ALTER TABLE [db_name.]tb_name alter_table_clause
- 
+
 alter_table_clause: {
     alter_table_options
   | ADD COLUMN col_name column_type
@@ -97,10 +98,10 @@ alter_table_clause: {
   | MODIFY COLUMN col_name column_type
   | RENAME COLUMN old_col_name new_col_name
 }
- 
+
 alter_table_options:
     alter_table_option ...
- 
+
 alter_table_option: {
     TTL value
   | COMMENT 'string_value'
@@ -110,6 +111,7 @@ alter_table_option: {
 
 **使用说明**
 对普通表可以进行如下修改操作
+
 1. ADD COLUMN：添加列。
 2. DROP COLUMN：删除列。
 3. MODIFY COLUMN：修改列定义，如果数据列的类型是可变长类型，那么可以使用此指令修改其宽度，只能改大，不能改小。
@@ -143,15 +145,15 @@ ALTER TABLE tb_name RENAME COLUMN old_col_name new_col_name
 
 ```sql
 ALTER TABLE [db_name.]tb_name alter_table_clause
- 
+
 alter_table_clause: {
     alter_table_options
   | SET TAG tag_name = new_tag_value
 }
- 
+
 alter_table_options:
     alter_table_option ...
- 
+
 alter_table_option: {
     TTL value
   | COMMENT 'string_value'
@@ -159,6 +161,7 @@ alter_table_option: {
 ```
 
 **使用说明**
+
 1. 对子表的列和标签的修改，除了更改标签值以外，都要通过超级表才能进行。
 
 ### 修改子表标签值
@@ -169,7 +172,7 @@ ALTER TABLE tb_name SET TAG tag_name=new_tag_value;
 
 ## 删除表
 
-可以在一条SQL语句中删除一个或多个普通表或子表。
+可以在一条 SQL 语句中删除一个或多个普通表或子表。
 
 ```sql
 DROP TABLE [IF EXISTS] [db_name.]tb_name [, [IF EXISTS] [db_name.]tb_name] ...
@@ -179,7 +182,7 @@ DROP TABLE [IF EXISTS] [db_name.]tb_name [, [IF EXISTS] [db_name.]tb_name] ...
 
 ### 显示所有表
 
-如下SQL语句可以列出当前数据库中的所有表名。
+如下 SQL 语句可以列出当前数据库中的所有表名。
 
 ```sql
 SHOW TABLES [LIKE tb_name_wildchar];

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

@@ -476,7 +476,7 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) {
       if (code) goto _err;
 
       if (!state->aBlockL) {
-        state->aBlockL = taosArrayInit(0, sizeof(SBlockIdx));
+        state->aBlockL = taosArrayInit(0, sizeof(SBlockL));
       } else {
         taosArrayClear(state->aBlockL);
       }

source/libs/function/src/builtins.c

@@ -303,7 +303,7 @@ static int32_t translateInOutStr(SFunctionNode* pFunc, char* pErrBuf, int32_t le
   }
 
   SExprNode* pPara1 = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0);
-  if (!IS_VAR_DATA_TYPE(pPara1->resType.type)) {
+  if (!IS_STR_DATA_TYPE(pPara1->resType.type)) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
@@ -317,7 +317,7 @@ static int32_t translateTrimStr(SFunctionNode* pFunc, char* pErrBuf, int32_t len
   }
 
   SExprNode* pPara1 = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0);
-  if (!IS_VAR_DATA_TYPE(pPara1->resType.type)) {
+  if (!IS_STR_DATA_TYPE(pPara1->resType.type)) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
@@ -546,7 +546,7 @@ static int32_t translateApercentile(SFunctionNode* pFunc, char* pErrBuf, int32_t
   // param2
   if (3 == numOfParams) {
     uint8_t para3Type = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 2))->resType.type;
-    if (!IS_VAR_DATA_TYPE(para3Type)) {
+    if (!IS_STR_DATA_TYPE(para3Type)) {
       return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
     }
 
@@ -593,7 +593,7 @@ static int32_t translateApercentileImpl(SFunctionNode* pFunc, char* pErrBuf, int
     // param2
     if (3 == numOfParams) {
       uint8_t para3Type = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 2))->resType.type;
-      if (!IS_VAR_DATA_TYPE(para3Type)) {
+      if (!IS_STR_DATA_TYPE(para3Type)) {
         return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
       }
 
@@ -1388,7 +1388,7 @@ static int32_t translateSample(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
   }
 
   // set result type
-  if (IS_VAR_DATA_TYPE(colType)) {
+  if (IS_STR_DATA_TYPE(colType)) {
     pFunc->node.resType = (SDataType){.bytes = pCol->resType.bytes, .type = colType};
   } else {
     pFunc->node.resType = (SDataType){.bytes = tDataTypes[colType].bytes, .type = colType};
@@ -1431,7 +1431,7 @@ static int32_t translateTail(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   }
 
   // set result type
-  if (IS_VAR_DATA_TYPE(colType)) {
+  if (IS_STR_DATA_TYPE(colType)) {
     pFunc->node.resType = (SDataType){.bytes = pCol->resType.bytes, .type = colType};
   } else {
     pFunc->node.resType = (SDataType){.bytes = tDataTypes[colType].bytes, .type = colType};
@@ -1514,7 +1514,7 @@ static int32_t translateInterp(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
     for (int32_t i = 1; i < 3; ++i) {
       nodeType = nodeType(nodesListGetNode(pFunc->pParameterList, i));
       paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, i))->resType.type;
-      if (!IS_VAR_DATA_TYPE(paraType) || QUERY_NODE_VALUE != nodeType) {
+      if (!IS_STR_DATA_TYPE(paraType) || QUERY_NODE_VALUE != nodeType) {
         return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
       }
 
@@ -1682,7 +1682,7 @@ static int32_t translateLength(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
     return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  if (!IS_VAR_DATA_TYPE(((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type)) {
+  if (!IS_STR_DATA_TYPE(((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type)) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
@@ -1714,7 +1714,7 @@ static int32_t translateConcatImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t
   for (int32_t i = 0; i < numOfParams; ++i) {
     SNode*  pPara = nodesListGetNode(pFunc->pParameterList, i);
     uint8_t paraType = ((SExprNode*)pPara)->resType.type;
-    if (!IS_VAR_DATA_TYPE(paraType) && !IS_NULL_TYPE(paraType)) {
+    if (!IS_STR_DATA_TYPE(paraType) && !IS_NULL_TYPE(paraType)) {
       return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
     }
     if (TSDB_DATA_TYPE_NCHAR == paraType) {
@@ -1770,7 +1770,7 @@ static int32_t translateSubstr(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
 
   uint8_t para0Type = pPara0->resType.type;
   uint8_t para1Type = pPara1->resType.type;
-  if (!IS_VAR_DATA_TYPE(para0Type) || !IS_INTEGER_TYPE(para1Type)) {
+  if (!IS_STR_DATA_TYPE(para0Type) || !IS_INTEGER_TYPE(para1Type)) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
@@ -1802,7 +1802,7 @@ static int32_t translateCast(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   uint8_t para2Type = pFunc->node.resType.type;
 
   int32_t para2Bytes = pFunc->node.resType.bytes;
-  if (IS_VAR_DATA_TYPE(para2Type)) {
+  if (IS_STR_DATA_TYPE(para2Type)) {
     para2Bytes -= VARSTR_HEADER_SIZE;
   }
   if (para2Bytes <= 0 || para2Bytes > 4096) {  // cast dst var type length limits to 4096 bytes
@@ -1859,7 +1859,7 @@ static int32_t translateToUnixtimestamp(SFunctionNode* pFunc, char* pErrBuf, int
     return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  if (!IS_VAR_DATA_TYPE(((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type)) {
+  if (!IS_STR_DATA_TYPE(((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type)) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
@@ -1878,7 +1878,7 @@ static int32_t translateTimeTruncate(SFunctionNode* pFunc, char* pErrBuf, int32_
 
   uint8_t para1Type = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
   uint8_t para2Type = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 1))->resType.type;
-  if ((!IS_VAR_DATA_TYPE(para1Type) && !IS_INTEGER_TYPE(para1Type) && TSDB_DATA_TYPE_TIMESTAMP != para1Type) ||
+  if ((!IS_STR_DATA_TYPE(para1Type) && !IS_INTEGER_TYPE(para1Type) && TSDB_DATA_TYPE_TIMESTAMP != para1Type) ||
       !IS_INTEGER_TYPE(para2Type)) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
@@ -1911,7 +1911,7 @@ static int32_t translateTimeDiff(SFunctionNode* pFunc, char* pErrBuf, int32_t le
 
   for (int32_t i = 0; i < 2; ++i) {
     uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, i))->resType.type;
-    if (!IS_VAR_DATA_TYPE(paraType) && !IS_INTEGER_TYPE(paraType) && TSDB_DATA_TYPE_TIMESTAMP != paraType) {
+    if (!IS_STR_DATA_TYPE(paraType) && !IS_INTEGER_TYPE(paraType) && TSDB_DATA_TYPE_TIMESTAMP != paraType) {
       return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
     }
   }

source/libs/scalar/inc/sclInt.h

@@ -45,6 +45,8 @@ typedef struct SScalarCtx {
 #define SCL_IS_CONST_CALC(_ctx) (NULL == (_ctx)->pBlockList)
 //#define SCL_IS_NULL_VALUE_NODE(_node) ((QUERY_NODE_VALUE == nodeType(_node)) && (TSDB_DATA_TYPE_NULL == ((SValueNode *)_node)->node.resType.type) && (((SValueNode *)_node)->placeholderNo <= 0))
 #define SCL_IS_NULL_VALUE_NODE(_node) ((QUERY_NODE_VALUE == nodeType(_node)) && (TSDB_DATA_TYPE_NULL == ((SValueNode *)_node)->node.resType.type))
+#define SCL_IS_COMPARISON_OPERATOR(_opType) ((_opType) >= OP_TYPE_GREATER_THAN && (_opType) < OP_TYPE_IS_NOT_UNKNOWN)
+#define SCL_DOWNGRADE_DATETYPE(_type) ((_type) == TSDB_DATA_TYPE_BIGINT || TSDB_DATA_TYPE_DOUBLE == (_type) || (_type) == TSDB_DATA_TYPE_UBIGINT)
 
 #define sclFatal(...)  qFatal(__VA_ARGS__)
 #define sclError(...)  qError(__VA_ARGS__)

source/libs/scalar/src/scalar.c

@@ -9,6 +9,7 @@
 #include "scalar.h"
 #include "tudf.h"
 #include "ttime.h"
+#include "tcompare.h"
 
 int32_t scalarGetOperatorParamNum(EOperatorType type) {
   if (OP_TYPE_IS_NULL == type || OP_TYPE_IS_NOT_NULL == type || OP_TYPE_IS_TRUE == type || OP_TYPE_IS_NOT_TRUE == type
@@ -219,6 +220,82 @@ void sclFreeParamList(SScalarParam *param, int32_t paramNum) {
   taosMemoryFree(param);
 }
 
+void sclDowngradeValueType(SValueNode *valueNode) {
+  switch (valueNode->node.resType.type) {
+    case TSDB_DATA_TYPE_BIGINT: {
+      int8_t i8 = valueNode->datum.i;
+      if (i8 == valueNode->datum.i) {
+        valueNode->node.resType.type = TSDB_DATA_TYPE_TINYINT;
+        *(int8_t*)&valueNode->typeData = i8;
+        break;
+      }
+      int16_t i16 = valueNode->datum.i;
+      if (i16 == valueNode->datum.i) {
+        valueNode->node.resType.type = TSDB_DATA_TYPE_SMALLINT;
+        *(int16_t*)&valueNode->typeData = i16;
+        break;
+      }
+      int32_t i32 = valueNode->datum.i;
+      if (i32 == valueNode->datum.i) {
+        valueNode->node.resType.type = TSDB_DATA_TYPE_INT;
+        *(int32_t*)&valueNode->typeData = i32;
+        break;
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_UBIGINT:{
+      uint8_t u8 = valueNode->datum.i;
+      if (u8 == valueNode->datum.i) {
+        int8_t i8 = valueNode->datum.i;
+        if (i8 == valueNode->datum.i) {
+          valueNode->node.resType.type = TSDB_DATA_TYPE_TINYINT;
+          *(int8_t*)&valueNode->typeData = i8;
+        } else {
+          valueNode->node.resType.type = TSDB_DATA_TYPE_UTINYINT;
+          *(uint8_t*)&valueNode->typeData = u8;
+        }
+        break;
+      }
+      uint16_t u16 = valueNode->datum.i;
+      if (u16 == valueNode->datum.i) {
+        int16_t i16 = valueNode->datum.i;
+        if (i16 == valueNode->datum.i) {
+          valueNode->node.resType.type = TSDB_DATA_TYPE_SMALLINT;
+          *(int16_t*)&valueNode->typeData = i16;
+        } else {
+          valueNode->node.resType.type = TSDB_DATA_TYPE_USMALLINT;
+          *(uint16_t*)&valueNode->typeData = u16;
+        }
+        break;
+      }
+      uint32_t u32 = valueNode->datum.i;
+      if (u32 == valueNode->datum.i) {
+        int32_t i32 = valueNode->datum.i;
+        if (i32 == valueNode->datum.i) {
+          valueNode->node.resType.type = TSDB_DATA_TYPE_INT;
+          *(int32_t*)&valueNode->typeData = i32;
+        } else {
+          valueNode->node.resType.type = TSDB_DATA_TYPE_UINT;
+          *(uint32_t*)&valueNode->typeData = u32;
+        }
+        break;
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_DOUBLE: {
+      float f = valueNode->datum.d;
+      if (FLT_EQUAL(f, valueNode->datum.d)) {
+        valueNode->node.resType.type = TSDB_DATA_TYPE_FLOAT;
+        *(float*)&valueNode->typeData = f;
+        break;
+      }
+      break;
+    }
+    default:
+      break;
+  }
+}
+
 int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t *rowNum) {
   switch (nodeType(node)) {
     case QUERY_NODE_LEFT_VALUE: {
@@ -675,6 +752,10 @@ EDealRes sclRewriteNonConstOperator(SNode** pNode, SScalarCtx *ctx) {
         return DEAL_RES_ERROR;
       }
     }
+
+    if (SCL_IS_COMPARISON_OPERATOR(node->opType) && SCL_DOWNGRADE_DATETYPE(valueNode->node.resType.type)) {
+      sclDowngradeValueType(valueNode);
+    }    
   }
 
   if (node->pRight && (QUERY_NODE_VALUE == nodeType(node->pRight))) {
@@ -692,6 +773,10 @@ EDealRes sclRewriteNonConstOperator(SNode** pNode, SScalarCtx *ctx) {
         return DEAL_RES_ERROR;
       }
     }
+
+    if (SCL_IS_COMPARISON_OPERATOR(node->opType) && SCL_DOWNGRADE_DATETYPE(valueNode->node.resType.type)) {
+      sclDowngradeValueType(valueNode);
+    }    
   }
 
   if (node->pRight && (QUERY_NODE_NODE_LIST == nodeType(node->pRight))) {

source/libs/stream/src/streamMeta.c

@@ -265,6 +265,8 @@ int32_t streamLoadTasks(SStreamMeta* pMeta) {
     }
   }
 
+  tdbFree(pKey);
+  tdbFree(pVal);
   if (tdbTbcClose(pCur) < 0) {
     return -1;
   }