Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/data_format

example/src/tmq.c

@@ -22,7 +22,7 @@
 static int  running = 1;
 static void msg_process(TAOS_RES* msg) {
   char buf[1024];
-  memset(buf, 0, 1024);
+  /*memset(buf, 0, 1024);*/
   printf("topic: %s\n", tmq_get_topic_name(msg));
   printf("vg: %d\n", tmq_get_vgroup_id(msg));
   while (1) {

include/common/tcommon.h

@@ -48,6 +48,7 @@ enum {
 typedef enum EStreamType {
   STREAM_NORMAL = 1,
   STREAM_INVERT,
+  STREAM_REPROCESS,
   STREAM_INVALID,
 } EStreamType;
 

include/common/ttypes.h

@@ -50,6 +50,7 @@ typedef struct {
 #define varDataLenByData(v)    (*(VarDataLenT *)(((char *)(v)) - VARSTR_HEADER_SIZE))
 #define varDataSetLen(v, _len) (((VarDataLenT *)(v))[0] = (VarDataLenT)(_len))
 #define IS_VAR_DATA_TYPE(t)    (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR) || ((t) == TSDB_DATA_TYPE_JSON))
+#define IS_STR_DATA_TYPE(t)    (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR))
 
 #define varDataNetLen(v)  (htons(((VarDataLenT *)(v))[0]))
 #define varDataNetTLen(v) (sizeof(VarDataLenT) + varDataNetLen(v))

include/libs/function/function.h

@@ -126,7 +126,7 @@ enum {
 
 enum {
   MAIN_SCAN     = 0x0u,
-  REVERSE_SCAN  = 0x1u,
+  REVERSE_SCAN  = 0x1u,  // todo remove it
   REPEAT_SCAN   = 0x2u,  //repeat scan belongs to the master scan
   MERGE_STAGE   = 0x20u,
 };
@@ -222,13 +222,6 @@ enum {
 typedef struct tExprNode {
   int32_t nodeType;
   union {
-    struct {
-      int32_t           optr;   // binary operator
-      void             *info;   // support filter operation on this expression only available for leaf node
-      struct tExprNode *pLeft;  // left child pointer
-      struct tExprNode *pRight; // right child pointer
-    } _node;
-
     SSchema            *pSchema;// column node
     struct SVariant    *pVal;   // value node
 
@@ -237,12 +230,6 @@ typedef struct tExprNode {
       int32_t           functionId;
       int32_t           num;
       struct SFunctionNode    *pFunctNode;
-      // Note that the attribute of pChild is not the parameter of function, it is the columns that involved in the
-      // calculation instead.
-      // E.g., Cov(col1, col2), the column information, w.r.t. the col1 and col2, is kept in pChild nodes.
-      //  The concat function, concat(col1, col2), is a binary scalar
-      //  operator and is kept in the attribute of _node.
-      struct tExprNode **pChild;
     } _function;
 
     struct {
@@ -271,9 +258,10 @@ typedef struct SAggFunctionInfo {
 } SAggFunctionInfo;
 
 struct SScalarParam {
-  SColumnInfoData   *columnData;
-  SHashObj          *pHashFilter;
-  int32_t            numOfRows;
+  SColumnInfoData *columnData;
+  SHashObj        *pHashFilter;
+  void            *param;  // other parameter, such as meta handle from vnode, to extract table name/tag value
+  int32_t          numOfRows;
 };
 
 int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, SResultDataInfo* pInfo, int16_t extLength,
@@ -281,10 +269,6 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI
 
 bool qIsValidUdf(SArray* pUdfInfo, const char* name, int32_t len, int32_t* functionId);
 
-tExprNode* exprTreeFromBinary(const void* data, size_t size);
-
-tExprNode* exprdup(tExprNode* pTree);
-
 void resetResultRowEntryResult(SqlFunctionCtx* pCtx, int32_t num);
 void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell);
 int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num, SSDataBlock* pResBlock);

include/libs/nodes/plannodes.h

@@ -193,7 +193,6 @@ typedef struct SScanPhysiNode {
 } SScanPhysiNode;
 
 typedef SScanPhysiNode STagScanPhysiNode;
-typedef SScanPhysiNode SStreamScanPhysiNode;
 
 typedef struct SSystemTableScanPhysiNode {
   SScanPhysiNode scan;
@@ -217,6 +216,7 @@ typedef struct STableScanPhysiNode {
 } STableScanPhysiNode;
 
 typedef STableScanPhysiNode STableSeqScanPhysiNode;
+typedef STableScanPhysiNode SStreamScanPhysiNode;
 
 typedef struct SProjectPhysiNode {
   SPhysiNode node;

include/libs/nodes/querynodes.h

@@ -324,6 +324,7 @@ typedef struct SQuery {
   SArray*        pTableList;
   bool           showRewrite;
   int32_t        placeholderNum;
+  SArray*        pPlaceholderValues;
 } SQuery;
 
 void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);

include/libs/parser/parser.h

@@ -47,21 +47,23 @@ typedef struct SParseContext {
   bool             isSuperUser;
 } SParseContext;
 
-int32_t qParseQuerySql(SParseContext* pCxt, SQuery** pQuery);
+int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery);
 bool    isInsertSql(const char* pStr, size_t length);
 
 void qDestroyQuery(SQuery* pQueryNode);
 
 int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
 
-int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash);
-int32_t qResetStmtDataBlock(void* block, bool keepBuf);
-int32_t qCloneStmtDataBlock(void** pDst, void* pSrc);
-void    qFreeStmtDataBlock(void* pDataBlock);
-int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t vgId);
-void    qDestroyStmtDataBlock(void* pBlock);
-STableMeta *qGetTableMetaInDataBlock(void* pDataBlock);
+int32_t     qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash);
+int32_t     qResetStmtDataBlock(void* block, bool keepBuf);
+int32_t     qCloneStmtDataBlock(void** pDst, void* pSrc);
+void        qFreeStmtDataBlock(void* pDataBlock);
+int32_t     qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t vgId);
+void        qDestroyStmtDataBlock(void* pBlock);
+STableMeta* qGetTableMetaInDataBlock(void* pDataBlock);
 
+int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId);
+int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery);
 int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen);
 int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx,
                                 int32_t rowNum);
@@ -75,7 +77,7 @@ int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char*
 
 void*   smlInitHandle(SQuery* pQuery);
 void    smlDestroyHandle(void* pHandle);
-int32_t smlBindData(void* handle, SArray* tags, SArray* colsFormat, SArray* colsSchema, SArray* cols, bool format,
+int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format,
                     STableMeta* pTableMeta, char* tableName, char* msgBuf, int16_t msgBufLen);
 int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
 

source/client/inc/clientStmt.h

@@ -71,6 +71,7 @@ typedef struct SStmtBindInfo {
 
 typedef struct SStmtExecInfo {
   int32_t      affectedRows;
+  bool         emptyRes;
   SRequestObj* pRequest;
   SHashObj*    pVgHash;
   SHashObj*    pBlockHash;

source/client/src/clientImpl.c

@@ -180,7 +180,7 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
     return code;
   }
 
-  code = qParseQuerySql(&cxt, pQuery);
+  code = qParseSql(&cxt, pQuery);
   if (TSDB_CODE_SUCCESS == code) {
     if ((*pQuery)->haveResultSet) {
       setResSchemaInfo(&pRequest->body.resInfo, (*pQuery)->pResSchema, (*pQuery)->numOfResCols);
@@ -554,8 +554,8 @@ static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) {
   }
   taosMemoryFreeClear(db);
 
-  connectReq.connType  = pObj->connType;
-  connectReq.pid       = htonl(appInfo.pid);
+  connectReq.connType = pObj->connType;
+  connectReq.pid = htonl(appInfo.pid);
   connectReq.startTime = htobe64(appInfo.startTime);
 
   tstrncpy(connectReq.app, appInfo.appName, sizeof(connectReq.app));

source/client/src/clientMain.c

@@ -303,6 +303,7 @@ int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields)
         break;
     }
   }
+  str[len] = 0;
 
   return len;
 }
@@ -567,7 +568,7 @@ void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param
     // todo directly call fp
   }
 
-  taos_query_l(taos, sql, (int32_t) strlen(sql));
+  taos_query_l(taos, sql, (int32_t)strlen(sql));
 }
 
 void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {

source/client/src/clientStmt.c

@@ -279,6 +279,7 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) {
   }
 
   pStmt->exec.autoCreateTbl = false;
+  pStmt->exec.emptyRes = false;
   
   if (keepTable) {
     return TSDB_CODE_SUCCESS;
@@ -628,8 +629,7 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
       STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
     }
 
-    bool emptyResult = false;
-    STMT_RET(qStmtBindParam(pStmt->sql.pQueryPlan, bind, colIdx, pStmt->exec.pRequest->requestId, &emptyResult));
+    STMT_RET(qStmtBindParam(pStmt->sql.pQueryPlan, bind, colIdx, pStmt->exec.pRequest->requestId, &pStmt->exec.emptyRes));
   }
   
   STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
@@ -736,7 +736,11 @@ int stmtExec(TAOS_STMT *stmt) {
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
 
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
-    scheduleQuery(pStmt->exec.pRequest, pStmt->sql.pQueryPlan, pStmt->sql.nodeList, NULL);
+    if (pStmt->exec.emptyRes) {
+      pStmt->exec.pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
+    } else {
+      scheduleQuery(pStmt->exec.pRequest, pStmt->sql.pQueryPlan, pStmt->sql.nodeList, NULL);
+    }
   } else {
     STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash));
     launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true, (autoCreateTbl ? (void**)&pRsp : NULL));

source/client/test/smlTest.cpp

@@ -33,7 +33,7 @@ int main(int argc, char **argv) {
   return RUN_ALL_TESTS();
 }
 
-TEST(testCase, smlParseString_Test) {
+TEST(testCase, smlParseInfluxString_Test) {
   char       msg[256] = {0};
   SSmlMsgBuf msgBuf;
   msgBuf.buf = msg;
@@ -42,7 +42,7 @@ TEST(testCase, smlParseString_Test) {
 
   // case 1
   char *sql = "st,t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000    ,32,c=3";
-  int   ret = smlParseString(sql, &elements, &msgBuf);
+  int   ret = smlParseInfluxString(sql, &elements, &msgBuf);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(elements.measure, sql);
   ASSERT_EQ(elements.measureLen, strlen("st"));
@@ -60,13 +60,13 @@ TEST(testCase, smlParseString_Test) {
   // case 2  false
   sql = "st,t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2,c2=false,c4=4f64 1626006833639000000";
   memset(&elements, 0, sizeof(SSmlLineInfo));
-  ret = smlParseString(sql, &elements, &msgBuf);
+  ret = smlParseInfluxString(sql, &elements, &msgBuf);
   ASSERT_NE(ret, 0);
 
   // case 3  false
   sql = "st, t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2,c2=false,c4=4f64 1626006833639000000";
   memset(&elements, 0, sizeof(SSmlLineInfo));
-  ret = smlParseString(sql, &elements, &msgBuf);
+  ret = smlParseInfluxString(sql, &elements, &msgBuf);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 2);
   ASSERT_EQ(elements.colsLen, strlen("t1=3,t2=4,t3=t3"));
@@ -74,7 +74,7 @@ TEST(testCase, smlParseString_Test) {
   // case 4  tag is null
   sql = "st, c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000";
   memset(&elements, 0, sizeof(SSmlLineInfo));
-  ret = smlParseString(sql, &elements, &msgBuf);
+  ret = smlParseInfluxString(sql, &elements, &msgBuf);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(elements.measure, sql);
   ASSERT_EQ(elements.measureLen, strlen("st"));
@@ -92,7 +92,7 @@ TEST(testCase, smlParseString_Test) {
   // case 5 tag is null
   sql = " st   c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64  1626006833639000000 ";
   memset(&elements, 0, sizeof(SSmlLineInfo));
-  ret = smlParseString(sql, &elements, &msgBuf);
+  ret = smlParseInfluxString(sql, &elements, &msgBuf);
   sql++;
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(elements.measure, sql);
@@ -111,13 +111,13 @@ TEST(testCase, smlParseString_Test) {
   // case 6
   sql = " st   c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64   ";
   memset(&elements, 0, sizeof(SSmlLineInfo));
-  ret = smlParseString(sql, &elements, &msgBuf);
+  ret = smlParseInfluxString(sql, &elements, &msgBuf);
   ASSERT_EQ(ret, 0);
 
   // case 7
   sql = " st   ,   ";
   memset(&elements, 0, sizeof(SSmlLineInfo));
-  ret = smlParseString(sql, &elements, &msgBuf);
+  ret = smlParseInfluxString(sql, &elements, &msgBuf);
   sql++;
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 3);
@@ -126,7 +126,7 @@ TEST(testCase, smlParseString_Test) {
   // case 8 false
   sql = ", st   ,   ";
   memset(&elements, 0, sizeof(SSmlLineInfo));
-  ret = smlParseString(sql, &elements, &msgBuf);
+  ret = smlParseInfluxString(sql, &elements, &msgBuf);
   ASSERT_NE(ret, 0);
 }
 
@@ -140,15 +140,13 @@ TEST(testCase, smlParseCols_Error_Test) {
     "c=f64",              // double
     "c=8f64f",
     "c=8ef64",
-    "c=1.7976931348623158e+390f64",
     "c=f32",              // float
     "c=8f32f",
     "c=8wef32",
     "c=-3.402823466e+39f32",
-    "c=",                 // float
+    "c=",                 // double
     "c=8f",
     "c=8we",
-    "c=3.402823466e+39",
     "c=i8",              // tiny int
     "c=-8i8f",
     "c=8wei8",
@@ -218,7 +216,7 @@ TEST(testCase, smlParseCols_tag_Test) {
   SHashObj *dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
 
   const char *data =
-      "cbin=\"passit hello,c=2\",cnch=L\"iisdfsf\",cbool=false,cf64=4.31f64,cf32_=8.32,cf32=8.23f32,ci8=-34i8,cu8=89u8,ci16=233i16,cu16=898u16,ci32=98289i32,cu32=12323u32,ci64=-89238i64,ci=989i,cu64=8989323u64,cbooltrue=true,cboolt=t,cboolf=f,cnch_=l\"iuwq\"";
+      "cbin=\"passit helloc=2\",cnch=L\"iisdfsf\",cbool=false,cf64=4.31f64,cf64_=8.32,cf32=8.23f32,ci8=-34i8,cu8=89u8,ci16=233i16,cu16=898u16,ci32=98289i32,cu32=12323u32,ci64=-89238i64,ci=989i,cu64=8989323u64,cbooltrue=true,cboolt=t,cboolf=f,cnch_=l\"iuwq\"";
   int32_t len = strlen(data);
   int32_t ret = smlParseCols(data, len, cols, true, dumplicateKey, &msgBuf);
   ASSERT_EQ(ret, TSDB_CODE_SUCCESS);
@@ -230,7 +228,7 @@ TEST(testCase, smlParseCols_tag_Test) {
   ASSERT_EQ(strncasecmp(kv->key, "cbin", 4), 0);
   ASSERT_EQ(kv->keyLen, 4);
   ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
-  ASSERT_EQ(kv->valueLen, 18);
+  ASSERT_EQ(kv->valueLen, 17);
   ASSERT_EQ(strncasecmp(kv->value, "\"passit", 7), 0);
   taosMemoryFree(kv);
 
@@ -280,7 +278,7 @@ TEST(testCase, smlParseCols_Test) {
 
   SHashObj *dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
 
-  const char *data = "cbin=\"passit hello,c=2\",cnch=L\"iisdfsf\",cbool=false,cf64=4.31f64,cf32_=8.32,cf32=8.23f32,ci8=-34i8,cu8=89u8,ci16=233i16,cu16=898u16,ci32=98289i32,cu32=12323u32,ci64=-89238i64,ci=989i,cu64=8989323u64,cbooltrue=true,cboolt=t,cboolf=f,cnch_=l\"iuwq\"";
+  const char *data = "cbin=\"passit hello,c=2\",cnch=L\"iisdfsf\",cbool=false,cf64=4.31f64,cf64_=8.32,cf32=8.23f32,ci8=-34i8,cu8=89u8,ci16=233i16,cu16=898u16,ci32=98289i32,cu32=12323u32,ci64=-89238i64,ci=989i,cu64=8989323u64,cbooltrue=true,cboolt=t,cboolf=f,cnch_=l\"iuwq\"";
   int32_t len = strlen(data);
   int32_t ret = smlParseCols(data, len, cols, false, dumplicateKey, &msgBuf);
   ASSERT_EQ(ret, TSDB_CODE_SUCCESS);
@@ -321,17 +319,17 @@ TEST(testCase, smlParseCols_Test) {
   ASSERT_EQ(kv->type, TSDB_DATA_TYPE_DOUBLE);
   ASSERT_EQ(kv->length, 8);
   //ASSERT_EQ(kv->d, 4.31);
-  printf("4.31 = kv->f:%f\n", kv->d);
+  printf("4.31 = kv->d:%f\n", kv->d);
   taosMemoryFree(kv);
 
   // float
   kv = (SSmlKv *)taosArrayGetP(cols, 4);
-  ASSERT_EQ(strncasecmp(kv->key, "cf32_", 5), 0);
+  ASSERT_EQ(strncasecmp(kv->key, "cf64_", 5), 0);
   ASSERT_EQ(kv->keyLen, 5);
-  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_FLOAT);
-  ASSERT_EQ(kv->length, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_DOUBLE);
+  ASSERT_EQ(kv->length, 8);
   //ASSERT_EQ(kv->f, 8.32);
-  printf("8.32 = kv->f:%f\n", kv->f);
+  printf("8.32 = kv->d:%f\n", kv->d);
   taosMemoryFree(kv);
 
   // float
@@ -467,7 +465,7 @@ TEST(testCase, smlParseCols_Test) {
   taosHashCleanup(dumplicateKey);
 }
 
-TEST(testCase, smlParseLine_Test) {
+TEST(testCase, smlProcess_influx_Test) {
   TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
   ASSERT_NE(taos, nullptr);
 
@@ -483,7 +481,7 @@ TEST(testCase, smlParseLine_Test) {
   SSmlHandle *info = smlBuildSmlInfo(taos, request, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS, true);
   ASSERT_NE(info, nullptr);
 
-  const char *sql[9] = {
+  const char *sql[11] = {
     "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,velocity=0,heading=221,grade=0 1451606400000000000",
     "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,velocity=0,heading=221,grade=0,fuel_consumption=25 1451607400000000000",
     "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 1451608400000000000",
@@ -492,14 +490,24 @@ TEST(testCase, smlParseLine_Test) {
     "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 1451606400000000000",
     "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 1451606400000000000",
     "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 1451609400000000000",
-    "readings,fleet=South,name=truck_0,driver=Trish,model=H-2,device_version=v2.3 fuel_consumption=25,grade=0 1451629400000000000"
+    "readings,fleet=South,name=truck_0,driver=Trish,model=H-2,device_version=v2.3 fuel_consumption=25,grade=0 1451629400000000000",
+    "stable,t1=t1,t2=t2,t3=t3 c1=1,c2=2,c3=3,c4=4 1451629500000000000",
+    "stable,t2=t2,t1=t1,t3=t3 c1=1,c3=3,c4=4 1451629600000000000"
   };
-  smlInsertLines(info, (char**)sql, 9);
-//  for (int i = 0; i < 3; i++) {
-//    smlParseLine(info, sql[i]);
-//  }
+  smlProcess(info, (char**)sql, sizeof(sql)/sizeof(sql[0]));
+
+  TAOS_RES *res = taos_query(taos, "select * from t_6885c584b98481584ee13dac399e173d");
+  ASSERT_NE(res, nullptr);
+  int fieldNum = taos_field_count(res);
+  ASSERT_EQ(fieldNum, 11);
+  int rowNum = taos_affected_rows(res);
+  ASSERT_EQ(rowNum, 2);
+  for (int i = 0; i < rowNum; ++i) {
+    TAOS_ROW rows = taos_fetch_row(res);
+  }
 }
 
+// different types
 TEST(testCase, smlParseLine_error_Test) {
   TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
   ASSERT_NE(taos, nullptr);
@@ -520,24 +528,247 @@ TEST(testCase, smlParseLine_error_Test) {
       "measure,t1=3 c1=8",
       "measure,t2=3 c1=8u8"
   };
-  int ret = smlInsertLines(info, (char **)sql, 2);
+  int ret = smlProcess(info, (char **)sql, sizeof(sql)/sizeof(sql[0]));
   ASSERT_NE(ret, 0);
 }
 
-// TEST(testCase, smlParseTS_Test) {
-//   char       msg[256] = {0};
-//   SSmlMsgBuf msgBuf;
-//   msgBuf.buf = msg;
-//   msgBuf.len = 256;
-//   SSmlLineInfo elements = {0};
-//
-//   SSmlHandle* info = smlBuildSmlInfo(taos, request, protocol, precision, dataFormat);
-//   if(!info){
-//     return (TAOS_RES*)request;
-//   }
-//   ret = smlParseTS(info, elements.timestamp, elements.timestampLen, cols);
-//   if(ret != TSDB_CODE_SUCCESS){
-//     uError("SML:0x%"PRIx64" smlParseTS failed", info->id);
-//     return ret;
+ TEST(testCase, smlGetTimestampLen_Test) {
+   uint8_t len = smlGetTimestampLen(0);
+   ASSERT_EQ(len, 1);
+
+   len = smlGetTimestampLen(1);
+   ASSERT_EQ(len, 1);
+
+   len = smlGetTimestampLen(10);
+   ASSERT_EQ(len, 2);
+
+   len = smlGetTimestampLen(390);
+   ASSERT_EQ(len, 3);
+
+   len = smlGetTimestampLen(-1);
+   ASSERT_EQ(len, 1);
+
+   len = smlGetTimestampLen(-10);
+   ASSERT_EQ(len, 2);
+
+   len = smlGetTimestampLen(-390);
+   ASSERT_EQ(len, 3);
+ }
+
+ TEST(testCase, smlProcess_telnet_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, true);
+   ASSERT_NE(info, nullptr);
+
+   const char *sql[4] = {
+       "sys.if.bytes.out 1479496100 1.3E0 host=web01 interface=eth0",
+       "sys.if.bytes.out 1479496101 1.3E1 interface=eth0 host=web01 ",
+       "sys.if.bytes.out 1479496102 1.3E3 network=tcp",
+       "sys.procs.running 1479496100 42 host=web01"
+   };
+   int ret = smlProcess(info, (char**)sql, sizeof(sql)/sizeof(sql[0]));
+   ASSERT_EQ(ret, 0);
+
+   TAOS_RES *res = taos_query(taos, "select * from t_8c30283b3c4131a071d1e16cf6d7094a");
+   ASSERT_NE(res, nullptr);
+   int fieldNum = taos_field_count(res);
+   ASSERT_EQ(fieldNum, 2);
+   int rowNum = taos_affected_rows(res);
+   ASSERT_EQ(rowNum, 1);
+   for (int i = 0; i < rowNum; ++i) {
+     TAOS_ROW rows = taos_fetch_row(res);
+   }
+
+   res = taos_query(taos, "select * from t_6931529054e5637ca92c78a1ad441961");
+   ASSERT_NE(res, nullptr);
+   fieldNum = taos_field_count(res);
+   ASSERT_EQ(fieldNum, 2);
+   rowNum = taos_affected_rows(res);
+   ASSERT_EQ(rowNum, 2);
+   for (int i = 0; i < rowNum; ++i) {
+     TAOS_ROW rows = taos_fetch_row(res);
+   }
+ }
+
+ TEST(testCase, smlProcess_json_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_JSON_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS, true);
+   ASSERT_NE(info, nullptr);
+
+   const char *sql = "[\n"
+       "    {\n"
+       "        \"metric\": \"sys.cpu.nice\",\n"
+       "        \"timestamp\": 1346846400,\n"
+       "        \"value\": 18,\n"
+       "        \"tags\": {\n"
+       "           \"host\": \"web01\",\n"
+       "           \"dc\": \"lga\"\n"
+       "        }\n"
+       "    },\n"
+       "    {\n"
+       "        \"metric\": \"sys.cpu.nice\",\n"
+       "        \"timestamp\": 1346846400,\n"
+       "        \"value\": 9,\n"
+       "        \"tags\": {\n"
+       "           \"host\": \"web02\",\n"
+       "           \"dc\": \"lga\"\n"
+       "        }\n"
+       "    }\n"
+       "]";
+   int ret = smlProcess(info, (char**)(&sql), -1);
+   ASSERT_EQ(ret, 0);
+
+   TAOS_RES *res = taos_query(taos, "select * from t_cb27a7198d637b4f1c6464bd73f756a7");
+   ASSERT_NE(res, nullptr);
+   int fieldNum = taos_field_count(res);
+   ASSERT_EQ(fieldNum, 2);
+//   int rowNum = taos_affected_rows(res);
+//   ASSERT_EQ(rowNum, 1);
+//   for (int i = 0; i < rowNum; ++i) {
+//     TAOS_ROW rows = taos_fetch_row(res);
 //   }
-// }
+
+   sql = "{\n"
+       "    \"metric\": \"meter_current\",\n"
+       "    \"timestamp\": {\n"
+       "        \"value\"  : 1346846400,\n"
+       "        \"type\"   : \"s\"\n"
+       "    },\n"
+       "    \"value\": {\n"
+       "         \"value\" : 10.3,\n"
+       "         \"type\"  : \"i64\"\n"
+       "    },\n"
+       "    \"tags\": {\n"
+       "       \"groupid\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"bigint\"\n"
+       "       },\n"
+       "       \"location\": { \n"
+       "           \"value\" : \"北京\",\n"
+       "           \"type\"  : \"binary\"\n"
+       "       },\n"
+       "       \"id\": \"d1001\"\n"
+       "    }\n"
+       "}";
+   ret = smlProcess(info, (char**)(&sql), -1);
+   ASSERT_EQ(ret, 0);
+
+   sql = "{\n"
+       "    \"metric\": \"meter_current\",\n"
+       "    \"timestamp\": {\n"
+       "        \"value\"  : 1346846400,\n"
+       "        \"type\"   : \"s\"\n"
+       "    },\n"
+       "    \"value\": {\n"
+       "         \"value\" : 10.3,\n"
+       "         \"type\"  : \"i64\"\n"
+       "    },\n"
+       "    \"tags\": {\n"
+       "       \"t1\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"bigint\"\n"
+       "       },\n"
+       "       \"t2\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"int\"\n"
+       "       },\n"
+       "       \"t3\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"i16\"\n"
+       "       },\n"
+       "       \"t4\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"i8\"\n"
+       "       },\n"
+       "       \"t5\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"f32\"\n"
+       "       },\n"
+       "       \"t6\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"double\"\n"
+       "       },\n"
+       "       \"t7\": { \n"
+       "           \"value\" : \"8323\",\n"
+       "           \"type\"  : \"binary\"\n"
+       "       },\n"
+       "       \"t8\": { \n"
+       "           \"value\" : \"北京\",\n"
+       "           \"type\"  : \"binary\"\n"
+       "       },\n"
+       "       \"t9\": { \n"
+       "           \"value\" : true,\n"
+       "           \"type\"  : \"bool\"\n"
+       "       },\n"
+       "       \"id\": \"d1001\"\n"
+       "    }\n"
+       "}";
+   ret = smlProcess(info, (char**)(&sql), -1);
+   ASSERT_EQ(ret, 0);
+
+   sql = "{\n"
+       "    \"metric\": \"meter_current\",\n"
+       "    \"timestamp\": {\n"
+       "        \"value\"  : 1346846400000,\n"
+       "        \"type\"   : \"ms\"\n"
+       "    },\n"
+       "    \"value\": \"ni\",\n"
+       "    \"tags\": {\n"
+       "       \"t1\": { \n"
+       "           \"value\" : 20,\n"
+       "           \"type\"  : \"i64\"\n"
+       "       },\n"
+       "       \"t2\": { \n"
+       "           \"value\" : 25,\n"
+       "           \"type\"  : \"i32\"\n"
+       "       },\n"
+       "       \"t3\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"smallint\"\n"
+       "       },\n"
+       "       \"t4\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"tinyint\"\n"
+       "       },\n"
+       "       \"t5\": { \n"
+       "           \"value\" : 2,\n"
+       "           \"type\"  : \"float\"\n"
+       "       },\n"
+       "       \"t6\": { \n"
+       "           \"value\" : 0.2,\n"
+       "           \"type\"  : \"f64\"\n"
+       "       },\n"
+       "       \"t7\": \"nsj\",\n"
+       "       \"t8\": { \n"
+       "           \"value\" : \"北京\",\n"
+       "           \"type\"  : \"binary\"\n"
+       "       },\n"
+       "       \"t9\": false,\n"
+       "       \"id\": \"d1001\"\n"
+       "    }\n"
+       "}";
+   ret = smlProcess(info, (char**)(&sql), -1);
+   ASSERT_EQ(ret, 0);
+ }

source/common/src/tdatablock.c

@@ -1447,6 +1447,10 @@ void blockDebugShowData(const SArray* dataBlocks) {
       for (int32_t k = 0; k < colNum; k++) {
         SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
         void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
+        if (pColInfoData->hasNull) {
+          printf(" %15s |", "NULL");
+          continue;
+        }
         switch (pColInfoData->info.type) {
           case TSDB_DATA_TYPE_TIMESTAMP:
             formatTimestamp(pBuf, *(uint64_t*)var, TSDB_TIME_PRECISION_MILLI);
@@ -1464,6 +1468,9 @@ void blockDebugShowData(const SArray* dataBlocks) {
           case TSDB_DATA_TYPE_UBIGINT:
             printf(" %15lu |", *(uint64_t*)var);
             break;
+          case TSDB_DATA_TYPE_DOUBLE:
+            printf(" %15f |", *(double*)var);
+            break;
         }
       }
       printf("\n");

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

@@ -485,8 +485,10 @@ static int32_t mndProcessDropTopicReq(SNodeMsg *pReq) {
       return -1;
     }
   }
+  // TODO: check ref
 
   int32_t code = mndDropTopic(pMnode, pReq, pTopic);
+  // TODO: iterate and drop related subscriptions and offsets
   mndReleaseTopic(pMnode, pTopic);
 
   if (code != 0) {

source/dnode/vnode/inc/vnode.h

@@ -115,6 +115,7 @@ void    tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond *pCond)
 void    tsdbDestroyTableGroup(STableGroupInfo *pGroupList);
 int32_t tsdbGetOneTableGroup(void *pMeta, uint64_t uid, TSKEY startKey, STableGroupInfo *pGroupInfo);
 int32_t tsdbGetTableGroupFromIdList(SVnode *pVnode, SArray *pTableIdList, STableGroupInfo *pGroupInfo);
+void    tsdbCleanupReadHandle(tsdbReaderT queryHandle);
 
 // tq
 

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

@@ -99,7 +99,6 @@ int32_t tsdbInitSma(STsdb *pTsdb);
 int32_t tsdbDropTSma(STsdb *pTsdb, char *pMsg);
 int32_t tsdbDropTSmaData(STsdb *pTsdb, int64_t indexUid);
 int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg);
-void    tsdbCleanupReadHandle(tsdbReaderT queryHandle);
 typedef enum {
   TSDB_FILE_HEAD = 0,  // .head
   TSDB_FILE_DATA,      // .data

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

@@ -457,9 +457,9 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
     }
 
     if (pHeadWithCkSum->head.msgType != TDMT_VND_SUBMIT) {
-      walSkipFetchBody(pExec->pWalReader, pHeadWithCkSum);
+      ASSERT(walSkipFetchBody(pExec->pWalReader, pHeadWithCkSum) == 0);
     } else {
-      walFetchBody(pExec->pWalReader, &pHeadWithCkSum);
+      ASSERT(walFetchBody(pExec->pWalReader, &pHeadWithCkSum) == 0);
     }
 
     SWalReadHead* pHead = &pHeadWithCkSum->head;
@@ -950,6 +950,7 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int32_t parallel) {
              .reader = pStreamReader,
              .meta = pTq->pVnode->pMeta,
              .pMsgCb = &pTq->pVnode->msgCb,
+             .vnode = pTq->pVnode,
       };
       pTask->exec.runners[i].inputHandle = pStreamReader;
       pTask->exec.runners[i].executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);

source/libs/executor/inc/executorimpl.h

@@ -333,6 +333,8 @@ typedef struct SScanInfo {
 
 typedef struct STableScanInfo {
   void*           dataReader;
+  SReadHandle     readHandle;
+
   SFileBlockLoadRecorder readRecorder;
   int64_t         numOfRows;
   int64_t         elapsedTime;
@@ -348,6 +350,11 @@ typedef struct STableScanInfo {
   SArray*         pColMatchInfo;
   int32_t         numOfOutput;
 
+  SExprInfo*      pPseudoExpr;
+  int32_t         numOfPseudoExpr;
+  SqlFunctionCtx* pPseudoCtx;
+//  int32_t*        rowCellInfoOffset;
+
   SQueryTableDataCond cond;
   int32_t         scanFlag;     // table scan flag to denote if it is a repeat/reverse/main scan
   int32_t         dataBlockLoadFlag;
@@ -364,9 +371,18 @@ typedef struct STagScanInfo {
   STableGroupInfo *pTableGroups;
 } STagScanInfo;
 
+typedef enum EStreamScanMode {
+  STREAM_SCAN_FROM_READERHANDLE = 1,
+  STREAM_SCAN_FROM_RES,
+  STREAM_SCAN_FROM_UPDATERES,
+  STREAM_SCAN_FROM_DATAREADER,
+} EStreamScanMode;
+
 typedef struct SStreamBlockScanInfo {
   SArray*      pBlockLists;      // multiple SSDatablock.
   SSDataBlock* pRes;             // result SSDataBlock
+  SSDataBlock* pUpdateRes;       // update SSDataBlock
+  int32_t      updateResIndex;
   int32_t      blockType;        // current block type
   int32_t      validBlockIndex;  // Is current data has returned?
   SColumnInfo* pCols;            // the output column info
@@ -376,8 +392,12 @@ typedef struct SStreamBlockScanInfo {
   SArray*      pColMatchInfo;    //
   SNode*       pCondition;
   SArray*      tsArray;
-  SUpdateInfo*  pUpdateInfo;
+  SUpdateInfo* pUpdateInfo;
   int32_t      primaryTsIndex;    // primary time stamp slot id
+  void*        pDataReader;
+  EStreamScanMode scanMode;
+  SOperatorInfo* pOperatorDumy;
+  SInterval      interval;     // if the upstream is an interval operator, the interval info is also kept here.
 } SStreamBlockScanInfo;
 
 typedef struct SSysTableScanInfo {
@@ -628,7 +648,7 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI
                                   int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total,
                                   SArray* pColList);
 void    getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key, STimeWindow* win);
-int32_t getTableScanOrder(SOperatorInfo* pOperator);
+int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t *order, int32_t* scanFlag);
 
 void    doSetOperatorCompleted(SOperatorInfo* pOperator);
 void    doFilter(const SNode* pFilterNode, SSDataBlock* pBlock);
@@ -644,12 +664,17 @@ SSDataBlock* loadNextDataBlock(void* param);
 
 void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowCellInfoOffset);
 
+SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
+                            int32_t type);
+SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t* numOfExprs);
+SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode);
+int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode);
+
 SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo,
                                    char* pData, int16_t bytes, bool masterscan, uint64_t groupId,
                                    SExecTaskInfo* pTaskInfo, bool isIntervalQuery, SAggSupporter* pSup);
 
-SOperatorInfo* createTableScanOperatorInfo(void* pDataReader, SQueryTableDataCond* pCond, int32_t numOfOutput, int32_t dataLoadFlag, const uint8_t* scanInfo,
-                                           SArray* pColMatchInfo, SSDataBlock* pResBlock, SNode* pCondition, SInterval* pInterval, double sampleRatio, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, tsdbReaderT pDataReader, SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SExprInfo* pScalarExprInfo,
                                            int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
@@ -678,8 +703,9 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
                                        SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo,
                                        const STableGroupInfo* pTableGroupInfo);
 SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList,
-                                            SArray* pTableIdList, SExecTaskInfo* pTaskInfo, SNode* pConditions);
+SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataReader, SSDataBlock* pResBlock,
+                                             SArray* pColList, SArray* pTableIdList, SExecTaskInfo* pTaskInfo,
+                                             SNode* pConditions, SOperatorInfo* pOperatorDumy, SInterval* pInterval);
 
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
                                       SInterval* pInterval, STimeWindow* pWindow, SSDataBlock* pResBlock, int32_t fillType, SNodeListNode* fillVal,
@@ -704,7 +730,7 @@ SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntim
 int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
                            int32_t numOfOutput, SArray* pPseudoList);
 
-void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order, bool createDummyCol);
+void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol);
 
 void copyTsColoum(SSDataBlock* pRes, SqlFunctionCtx* pCtx, int32_t numOfOutput);
 
@@ -733,6 +759,15 @@ bool aggDecodeResultRow(SOperatorInfo* pOperator, SAggSupporter* pSup, SOptrBasi
                         int32_t length);
 void aggEncodeResultRow(SOperatorInfo* pOperator, SAggSupporter* pSup, SOptrBasicInfo* pInfo, char** result,
                         int32_t* length);
+STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts,
+                                       SInterval* pInterval, int32_t precision, STimeWindow* win);
+int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos,
+                                        TSKEY ekey, __block_search_fn_t searchFn, STableQueryInfo* item,
+                                        int32_t order);
+int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
+
+void doClearWindow(SIntervalAggOperatorInfo* pInfo, char* pData, int16_t bytes,
+    uint64_t groupId, int32_t numOfOutput);
 
 #ifdef __cplusplus
 }

source/libs/executor/src/executorimpl.c

@@ -344,6 +344,28 @@ SResultRow* getNewResultRow_rv(SDiskbasedBuf* pResultBuf, int64_t tableGroupId,
   return pResultRow;
 }
 
+void doClearWindow(SIntervalAggOperatorInfo* pInfo, char* pData, int16_t bytes,
+    uint64_t groupId, int32_t numOfOutput) {
+  SAggSupporter* pSup = &pInfo->aggSup;
+  SET_RES_WINDOW_KEY(pSup->keyBuf, pData, bytes, groupId);
+  SResultRowPosition* p1 =
+      (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable, pSup->keyBuf,
+          GET_RES_WINDOW_KEY_LEN(bytes));
+  SResultRow* pResult = getResultRowByPos(pSup->pResultBuf, p1);
+  SqlFunctionCtx* pCtx = pInfo->binfo.pCtx;
+  for (int32_t i = 0; i < numOfOutput; ++i) {
+    pCtx[i].resultInfo = getResultCell(pResult, i, pInfo->binfo.rowCellInfoOffset);
+    struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo;
+    if (fmIsWindowPseudoColumnFunc(pCtx[i].functionId)) {
+      continue;
+    }
+    pResInfo->initialized = false;
+    if (pCtx[i].functionId != -1) {
+      pCtx[i].fpSet.init(&pCtx[i], pResInfo);
+    }
+  }
+}
+
 /**
  * the struct of key in hash table
  * +----------+---------------+
@@ -654,7 +676,7 @@ static FORCE_INLINE TSKEY reviseWindowEkey(STaskAttr* pQueryAttr, STimeWindow* p
 }
 
 static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order,
-                                   bool createDummyCol);
+                                   int32_t scanFlag, bool createDummyCol);
 
 static void doSetInputDataBlockInfo(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock,
                                     int32_t order) {
@@ -665,12 +687,12 @@ static void doSetInputDataBlockInfo(SOperatorInfo* pOperator, SqlFunctionCtx* pC
   }
 }
 
-void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order,
+void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
                        bool createDummyCol) {
   if (pBlock->pBlockAgg != NULL) {
     doSetInputDataBlockInfo(pOperator, pCtx, pBlock, order);
   } else {
-    doSetInputDataBlock(pOperator, pCtx, pBlock, order, createDummyCol);
+    doSetInputDataBlock(pOperator, pCtx, pBlock, order, scanFlag, createDummyCol);
   }
 }
 
@@ -717,14 +739,14 @@ static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunc
 }
 
 static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order,
-                                   bool createDummyCol) {
+    int32_t scanFlag, bool createDummyCol) {
   int32_t code = TSDB_CODE_SUCCESS;
 
   for (int32_t i = 0; i < pOperator->numOfExprs; ++i) {
     pCtx[i].order = order;
     pCtx[i].size = pBlock->info.rows;
     pCtx[i].pSrcBlock = pBlock;
-    pCtx[i].currentStage = MAIN_SCAN;
+    pCtx[i].currentStage = scanFlag;
 
     SInputColumnInfoData* pInput = &pCtx[i].input;
     pInput->uid = pBlock->info.uid;
@@ -740,7 +762,7 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt
         pInput->numOfRows = pBlock->info.rows;
         pInput->startRowIndex = 0;
 
-        // the last parameter is the timestamp column
+        // NOTE: the last parameter is the primary timestamp column
         if (fmIsTimelineFunc(pCtx[i].functionId) && (j == pOneExpr->base.numOfParams - 1)) {
           pInput->pPTS = pInput->pData[j];
         }
@@ -884,7 +906,8 @@ int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBloc
     } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
       ASSERT(!fmIsAggFunc(pfCtx->functionId));
 
-      if (fmIsPseudoColumnFunc(pfCtx->functionId)) {
+      // _rowts/_c0, not tbname column
+      if (fmIsPseudoColumnFunc(pfCtx->functionId) && (!fmIsScanPseudoColumnFunc(pfCtx->functionId))) {
         // do nothing
       } else if (fmIsNonstandardSQLFunc(pfCtx->functionId)) {
         SResultRowEntryInfo* pResInfo = GET_RES_INFO(&pCtx[k]);
@@ -3506,7 +3529,7 @@ static SSDataBlock* doMerge(SOperatorInfo* pOperator) {
       break;
     }
 
-    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pDataBlock, TSDB_ORDER_ASC, true);
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pDataBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
     //  updateOutputBuf(&pInfo->binfo, &pAggInfo->bufCapacity, pBlock->info.rows * pAggInfo->resultRowFactor,
     //  pOperator->pRuntimeEnv, true);
     doMergeImpl(pOperator, pOperator->numOfExprs, pDataBlock);
@@ -3671,17 +3694,24 @@ _error:
   return NULL;
 }
 
-int32_t getTableScanOrder(SOperatorInfo* pOperator) {
-  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
+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) {
+    *order = TSDB_ORDER_ASC;
+    *scanFlag = MAIN_SCAN;
+    return TSDB_CODE_SUCCESS;
+  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
+    STableScanInfo* pTableScanInfo = pOperator->info;
+    *order = pTableScanInfo->cond.order;
+    *scanFlag = pTableScanInfo->scanFlag;
+    return TSDB_CODE_SUCCESS;
+  } else {
     if (pOperator->pDownstream == NULL || pOperator->pDownstream[0] == NULL) {
-      return TSDB_ORDER_ASC;
+      return TSDB_CODE_INVALID_PARA;
     } else {
-      return getTableScanOrder(pOperator->pDownstream[0]);
+      return getTableScanInfo(pOperator->pDownstream[0], order, scanFlag);
     }
   }
-
-  STableScanInfo* pTableScanInfo = pOperator->info;
-  return pTableScanInfo->cond.order;
 }
 
 // this is a blocking operator
@@ -3697,6 +3727,9 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
 
   SOperatorInfo* downstream = pOperator->pDownstream[0];
 
+  int32_t order = TSDB_ORDER_ASC;
+  int32_t scanFlag = MAIN_SCAN;
+
   while (1) {
     publishOperatorProfEvent(downstream, QUERY_PROF_BEFORE_OPERATOR_EXEC);
     SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
@@ -3709,11 +3742,14 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
     //      setTagValue(pOperator, pAggInfo->current->pTable, pInfo->pCtx, pOperator->numOfExprs);
     //    }
 
-    int32_t order = getTableScanOrder(pOperator);
+    int32_t code = getTableScanInfo(pOperator, &order, &scanFlag);
+    if (code != TSDB_CODE_SUCCESS) {
+      longjmp(pTaskInfo->env, code);
+    }
 
     // there is an scalar expression that needs to be calculated before apply the group aggregation.
     if (pAggInfo->pScalarExprInfo != NULL) {
-      int32_t code = projectApplyFunctions(pAggInfo->pScalarExprInfo, pBlock, pBlock, pAggInfo->pScalarCtx,
+      code = projectApplyFunctions(pAggInfo->pScalarExprInfo, pBlock, pBlock, pAggInfo->pScalarCtx,
                                            pAggInfo->numOfScalarExpr, NULL);
       if (code != TSDB_CODE_SUCCESS) {
         pTaskInfo->code = code;
@@ -3723,7 +3759,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
 
     // the pDataBlock are always the same one, no need to call this again
     setExecutionContext(pOperator->numOfExprs, pBlock->info.groupId, pTaskInfo, pAggInfo);
-    setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, true);
+    setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, scanFlag, true);
     doAggregateImpl(pOperator, 0, pInfo->pCtx);
 
 #if 0  // test for encode/decode result info
@@ -4004,6 +4040,9 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
   }
 #endif
 
+  int32_t order = 0;
+  int32_t scanFlag = 0;
+
   SOperatorInfo* downstream = pOperator->pDownstream[0];
 
   while (1) {
@@ -4035,15 +4074,14 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
     //    }
 
     // the pDataBlock are always the same one, no need to call this again
-    int32_t order = getTableScanOrder(pOperator->pDownstream[0]);
+    int32_t code = getTableScanInfo(pOperator->pDownstream[0], &order, &scanFlag);
 
-    setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, false);
+    setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, scanFlag, false);
     blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
 
-    pTaskInfo->code = projectApplyFunctions(pOperator->pExpr, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->numOfExprs,
-                          pProjectInfo->pPseudoColInfo);
-    if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
-      longjmp(pTaskInfo->env, pTaskInfo->code);
+    code = projectApplyFunctions(pOperator->pExpr, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->numOfExprs, pProjectInfo->pPseudoColInfo);
+    if (code != TSDB_CODE_SUCCESS) {
+      longjmp(pTaskInfo->env, code);
     }
 
     int32_t status = handleLimitOffset(pOperator, pBlock);
@@ -4642,8 +4680,22 @@ SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t*
 
       pExp->pExpr->_function.functionId = pFuncNode->funcId;
       pExp->pExpr->_function.pFunctNode = pFuncNode;
+
       strncpy(pExp->pExpr->_function.functionName, pFuncNode->functionName,
               tListLen(pExp->pExpr->_function.functionName));
+#if 1
+      // todo refactor: add the parameter for tbname function
+      if (strcmp(pExp->pExpr->_function.functionName, "tbname") == 0) {
+        pFuncNode->pParameterList = nodesMakeList();
+        ASSERT(LIST_LENGTH(pFuncNode->pParameterList) == 0);
+        SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
+        if (NULL == res) { // todo handle error
+        } else {
+          res->node.resType = (SDataType) {.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
+          nodesListAppend(pFuncNode->pParameterList, res);
+        }
+      }
+#endif
 
       int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);
 
@@ -4704,58 +4756,29 @@ static int32_t doCreateTableGroup(void* metaHandle, int32_t tableType, uint64_t
                                   uint64_t queryId, uint64_t taskId);
 static SArray* extractTableIdList(const STableGroupInfo* pTableGroupInfo);
 static SArray* extractColumnInfo(SNodeList* pNodeList);
-static SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
-                                   int32_t type);
 
 static SArray* createSortInfo(SNodeList* pNodeList);
 static SArray* extractPartitionColInfo(SNodeList* pNodeList);
-static int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode);
 static void    setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode);
 
-static SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {
-  SInterval interval = {
-      .interval = pTableScanNode->interval,
-      .sliding = pTableScanNode->sliding,
-      .intervalUnit = pTableScanNode->intervalUnit,
-      .slidingUnit = pTableScanNode->slidingUnit,
-      .offset = pTableScanNode->offset,
-  };
-
-  return interval;
-}
-
 SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle,
                                   uint64_t queryId, uint64_t taskId, STableGroupInfo* pTableGroupInfo) {
   int32_t type = nodeType(pPhyNode);
 
   if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
     if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
-      SScanPhysiNode*      pScanPhyNode = (SScanPhysiNode*)pPhyNode;
       STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
 
-      int32_t     numOfCols = 0;
       tsdbReaderT pDataReader = doCreateDataReader(pTableScanNode, pHandle, pTableGroupInfo, (uint64_t)queryId, taskId);
       if (pDataReader == NULL && terrno != 0) {
         return NULL;
       }
 
-      SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;
-
-      SArray* pColList = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
-      SSDataBlock* pResBlock = createResDataBlock(pDescNode);
-
-      SQueryTableDataCond cond = {0};
-      int32_t             code = initQueryTableDataCond(&cond, pTableScanNode);
-      if (code != TSDB_CODE_SUCCESS) {
-        return NULL;
-      }
+      SOperatorInfo* pOperator = createTableScanOperatorInfo(pTableScanNode, pDataReader, pHandle, pTaskInfo);
 
-      SInterval      interval = extractIntervalInfo(pTableScanNode);
-      SOperatorInfo* pOperator = createTableScanOperatorInfo(
-          pDataReader, &cond, numOfCols, pTableScanNode->dataRequired, pTableScanNode->scanSeq, pColList, pResBlock,
-          pScanPhyNode->node.pConditions, &interval, pTableScanNode->ratio, pTaskInfo);
       STableScanInfo* pScanInfo = pOperator->info;
       pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
+
       return pOperator;
     } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
       SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pPhyNode;
@@ -4763,18 +4786,48 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
       return createExchangeOperatorInfo(pHandle->pMsgCb->clientRpc, pExchange->pSrcEndPoints, pResBlock, pTaskInfo);
     } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
       SScanPhysiNode* pScanPhyNode = (SScanPhysiNode*)pPhyNode;  // simple child table.
+      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
 
-      int32_t code = doCreateTableGroup(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableGroupInfo,
-                                        queryId, taskId);
-      SArray* tableIdList = extractTableIdList(pTableGroupInfo);
+      int32_t     numOfCols = 0;
+
+      tsdbReaderT pDataReader = NULL;
+      if (pHandle->vnode) {
+        pDataReader = doCreateDataReader(pTableScanNode, pHandle, pTableGroupInfo, (uint64_t)queryId, taskId);
+      } else {
+        doCreateTableGroup(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableGroupInfo, 
+            queryId, taskId);
+      }
+      if (pDataReader == NULL && terrno != 0) {
+        qDebug("pDataReader is     NULL");
+        // return NULL;
+      } else {
+        qDebug("pDataReader is not NULL");
+      }
 
       SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;
+
+      SArray* pColList = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
+      SSDataBlock* pResBlockDumy = createResDataBlock(pDescNode);
+
+      SQueryTableDataCond cond = {0};
+      int32_t code = initQueryTableDataCond(&cond, pTableScanNode);
+      if (code != TSDB_CODE_SUCCESS) {
+        return NULL;
+      }
+
+      SInterval      interval = extractIntervalInfo(pTableScanNode);
+      SOperatorInfo* pOperatorDumy = createTableScanOperatorInfo(
+          pDataReader, &cond, numOfCols, pTableScanNode->dataRequired, pTableScanNode->scanSeq, pColList,
+          pResBlockDumy, pScanPhyNode->node.pConditions, &interval, pTableScanNode->ratio, pTaskInfo);
+
+      // int32_t code = doCreateTableGroup(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableGroupInfo,
+      //                                   queryId, taskId);
+      SArray* tableIdList = extractTableIdList(pTableGroupInfo);
       SSDataBlock*        pResBlock = createResDataBlock(pDescNode);
 
-      int32_t        numOfCols = 0;
       SArray*        pCols = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
-      SOperatorInfo* pOperator = createStreamScanOperatorInfo(pHandle->reader, pResBlock, pCols, tableIdList, pTaskInfo,
-                                                              pScanPhyNode->node.pConditions);
+      SOperatorInfo* pOperator = createStreamScanOperatorInfo(pHandle->reader, pDataReader, pResBlock, pCols, tableIdList, pTaskInfo,
+                                                              pScanPhyNode->node.pConditions, pOperatorDumy, &interval);
       taosArrayDestroy(tableIdList);
       return pOperator;
     } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
@@ -4945,7 +4998,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
   return pOptr;
 }
 
-static int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode) {
+int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode) {
   pCond->loadExternalRows = false;
 
   pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;

source/libs/executor/src/groupoperator.c

@@ -287,7 +287,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
     }
 
     // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, true);
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true);
 
     // there is an scalar expression that needs to be calculated right before apply the group aggregation.
     if (pInfo->pScalarExprInfo != NULL) {

source/libs/executor/src/scanoperator.c

@@ -13,7 +13,6 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include <libs/function/function.h>
 #include "filter.h"
 #include "function.h"
 #include "functionMgt.h"
@@ -284,6 +283,27 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
       continue;
     }
 
+    // currently only the tbname pseudo column
+    if (pTableScanInfo->numOfPseudoExpr > 0) {
+      int32_t dstSlotId = pTableScanInfo->pPseudoExpr->base.resSchema.slotId;
+      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotId);
+      colInfoDataEnsureCapacity(pColInfoData, 0, pBlock->info.rows);
+
+      struct SScalarFuncExecFuncs fpSet;
+      fmGetScalarFuncExecFuncs(pTableScanInfo->pPseudoExpr->pExpr->_function.functionId, &fpSet);
+
+      SColumnInfoData infoData = {0};
+      infoData.info.type = TSDB_DATA_TYPE_BIGINT;
+      infoData.info.bytes = sizeof(uint64_t);
+      colInfoDataEnsureCapacity(&infoData, 0, 1);
+
+      colDataAppendInt64(&infoData, 0, &pBlock->info.uid);
+      SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .param = pTableScanInfo->readHandle.meta, .columnData = &infoData};
+
+      SScalarParam param = {.columnData = pColInfoData};
+      fpSet.process(&srcParam, 1, &param);
+    }
+
     return pBlock;
   }
 
@@ -314,8 +334,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
 
       STimeWindow* pWin = &pTableScanInfo->cond.twindow;
       qDebug("%s start to repeat ascending order scan data blocks due to query func required, qrange:%" PRId64
-             "-%" PRId64,
-             GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
+             "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
 
       // do prepare for the next round table scan operation
       tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
@@ -359,10 +378,29 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
   return NULL;
 }
 
-SOperatorInfo* createTableScanOperatorInfo(void* pDataReader, SQueryTableDataCond* pCond, int32_t numOfOutput,
-                                           int32_t dataLoadFlag, const uint8_t* scanInfo, SArray* pColMatchInfo,
-                                           SSDataBlock* pResBlock, SNode* pCondition, SInterval* pInterval,
-                                           double sampleRatio, SExecTaskInfo* pTaskInfo) {
+SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {
+  SInterval interval = {
+      .interval = pTableScanNode->interval,
+      .sliding = pTableScanNode->sliding,
+      .intervalUnit = pTableScanNode->intervalUnit,
+      .slidingUnit = pTableScanNode->slidingUnit,
+      .offset = pTableScanNode->offset,
+  };
+
+  return interval;
+}
+
+static void destroyTableScanOperatorInfo(void* param, int32_t numOfOutput) {
+  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;
+  taosMemoryFree(pTableScanInfo->pResBlock);
+  tsdbCleanupReadHandle(pTableScanInfo->dataReader);
+
+  if (pTableScanInfo->pColMatchInfo != NULL) {
+    taosArrayDestroy(pTableScanInfo->pColMatchInfo);
+  }
+}
+
+SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, tsdbReaderT pDataReader, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo) {
   STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
   SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -373,27 +411,42 @@ SOperatorInfo* createTableScanOperatorInfo(void* pDataReader, SQueryTableDataCon
     return NULL;
   }
 
-  pInfo->cond = *pCond;
-  pInfo->scanInfo = (SScanInfo){.numOfAsc = scanInfo[0], .numOfDesc = scanInfo[1]};
+  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
 
-  pInfo->interval = *pInterval;
-  pInfo->sampleRatio = sampleRatio;
-  pInfo->dataBlockLoadFlag = dataLoadFlag;
-  pInfo->pResBlock = pResBlock;
-  pInfo->pFilterNode = pCondition;
-  pInfo->dataReader = pDataReader;
-  pInfo->scanFlag = MAIN_SCAN;
-  pInfo->pColMatchInfo = pColMatchInfo;
+  int32_t numOfCols = 0;
+  SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
+
+  int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
+  if (code != TSDB_CODE_SUCCESS) {
+    return NULL;
+  }
 
-  pOperator->name = "TableScanOperator";  // for dubug purpose
+  if (pTableScanNode->scan.pScanPseudoCols != NULL) {
+    pInfo->pPseudoExpr = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->numOfPseudoExpr);
+    pInfo->pPseudoCtx  = createSqlFunctionCtx(pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, &pInfo->rowCellInfoOffset);
+  }
+
+  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};
+
+  pInfo->readHandle        = *readHandle;
+  pInfo->interval          = extractIntervalInfo(pTableScanNode);
+  pInfo->sampleRatio       = pTableScanNode->ratio;
+  pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
+  pInfo->pResBlock         = createResDataBlock(pDescNode);
+  pInfo->pFilterNode       = pTableScanNode->scan.node.pConditions;
+  pInfo->dataReader        = pDataReader;
+  pInfo->scanFlag          = MAIN_SCAN;
+  pInfo->pColMatchInfo     = pColList;
+
+  pOperator->name         = "TableScanOperator";  // for debug purpose
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
-  pOperator->blocking = false;
-  pOperator->status = OP_NOT_OPENED;
-  pOperator->info = pInfo;
-  pOperator->numOfExprs = numOfOutput;
-  pOperator->pTaskInfo = pTaskInfo;
+  pOperator->blocking     = false;
+  pOperator->status       = OP_NOT_OPENED;
+  pOperator->info         = pInfo;
+  pOperator->numOfExprs   = numOfCols;
+  pOperator->pTaskInfo    = pTaskInfo;
 
-  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, NULL, NULL, NULL, NULL);
+  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo, NULL, NULL, NULL);
 
   static int32_t cost = 0;
 
@@ -515,7 +568,40 @@ static void doClearBufferedBlocks(SStreamBlockScanInfo* pInfo) {
   taosArrayClear(pInfo->pBlockLists);
 }
 
-static SSDataBlock* getUpdateDataBlock(SStreamBlockScanInfo* pInfo) {
+static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
+  SSDataBlock* pSDB = pInfo->pUpdateRes;
+  if (pInfo->updateResIndex < pSDB->info.rows) {
+    SColumnInfoData* pColDataInfo = taosArrayGet(pSDB->pDataBlock, 0);
+    TSKEY *tsCols = (TSKEY*)pColDataInfo->pData;
+    SResultRowInfo dumyInfo;
+    dumyInfo.cur.pageId = -1;
+    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[pInfo->updateResIndex], &pInfo->interval,
+                                        pInfo->interval.precision, NULL);
+    STableScanInfo* pTableScanInfo = pInfo->pOperatorDumy->info;
+    pTableScanInfo->cond.twindow = win;
+    tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
+    pInfo->updateResIndex += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, pInfo->updateResIndex,
+        win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
+    pTableScanInfo->scanTimes = 0;
+    return true;
+  } else {
+    return false;
+  }
+}
+
+static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) {
+  SSDataBlock* pResult = NULL;
+  pResult = doTableScan(pInfo->pOperatorDumy);
+  if (pResult == NULL) {
+    if (prepareDataScan(pInfo)) {
+      // scan next window data
+      pResult = doTableScan(pInfo->pOperatorDumy);
+    }
+  }
+  return pResult;
+}
+
+static SSDataBlock* getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool invertible) {
   SColumnInfoData* pColDataInfo = taosArrayGet(pInfo->pRes->pDataBlock, pInfo->primaryTsIndex);
   TSKEY*           ts = (TSKEY*)pColDataInfo->pData;
   for (int32_t i = 0; i < pInfo->pRes->info.rows; i++) {
@@ -523,13 +609,19 @@ static SSDataBlock* getUpdateDataBlock(SStreamBlockScanInfo* pInfo) {
       taosArrayPush(pInfo->tsArray, ts + i);
     }
   }
-  if (taosArrayGetSize(pInfo->tsArray) > 0) {
+  int32_t size = taosArrayGetSize(pInfo->tsArray);
+  if (size > 0 && invertible) {
     // TODO(liuyao) get from tsdb
     //  SSDataBlock* p = createOneDataBlock(pInfo->pRes, true);
     //  p->info.type = STREAM_INVERT;
     //  taosArrayClear(pInfo->tsArray);
     //  return p;
-    return NULL;
+    SSDataBlock* p = createOneDataBlock(pInfo->pRes, false);
+    taosArraySet(p->pDataBlock, 0, pInfo->tsArray);
+    p->info.rows = size;
+    p->info.type = STREAM_REPROCESS;
+    taosArrayClear(pInfo->tsArray);
+    return p;
   }
   return NULL;
 }
@@ -556,14 +648,23 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     int32_t current = pInfo->validBlockIndex++;
     return taosArrayGetP(pInfo->pBlockLists, current);
   } else {
-    if (total > 0) {
-      ASSERT(total == 2);
-      SSDataBlock* pRes = taosArrayGetP(pInfo->pBlockLists, 0);
-      SSDataBlock* pUpRes = taosArrayGetP(pInfo->pBlockLists, 1);
-      blockDataDestroy(pUpRes);
-      taosArrayClear(pInfo->pBlockLists);
-      return pRes;
+    if (pInfo->scanMode == STREAM_SCAN_FROM_RES) {
+      blockDataDestroy(pInfo->pUpdateRes);
+      pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
+      return pInfo->pRes;
+    } else if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) {
+      blockDataCleanup(pInfo->pRes);
+      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
+      return pInfo->pUpdateRes;
+    } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
+      SSDataBlock* pSDB = doDataScan(pInfo);
+      if (pSDB == NULL) {
+        pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
+      } else {
+        return pSDB;
+      }
     }
+
     SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
     blockDataCleanup(pInfo->pRes);
 
@@ -629,12 +730,18 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
 
     if (rows == 0) {
       pOperator->status = OP_EXEC_DONE;
-    } else {
-      SSDataBlock* upRes = getUpdateDataBlock(pInfo);
+    } else if (pInfo->interval.interval > 0) {
+      SSDataBlock* upRes = getUpdateDataBlock(pInfo, true); //TODO(liuyao) get invertible from plan
       if (upRes) {
-        taosArrayPush(pInfo->pBlockLists, &(pInfo->pRes));
-        taosArrayPush(pInfo->pBlockLists, &upRes);
-        return upRes;
+        pInfo->pUpdateRes = upRes;
+        if (upRes->info.type = STREAM_REPROCESS) {
+          pInfo->updateResIndex = 0;
+          prepareDataScan(pInfo);
+          pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;
+        } else if (upRes->info.type = STREAM_INVERT) {
+          pInfo->scanMode = STREAM_SCAN_FROM_RES;
+          return upRes;
+        }
       }
     }
 
@@ -642,8 +749,10 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
   }
 }
 
-SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList,
-                                            SArray* pTableIdList, SExecTaskInfo* pTaskInfo, SNode* pCondition) {
+SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataReader,
+    SSDataBlock* pResBlock, SArray* pColList, SArray* pTableIdList,
+    SExecTaskInfo* pTaskInfo, SNode* pCondition, SOperatorInfo* pOperatorDumy,
+    SInterval* pInterval) {
   SStreamBlockScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamBlockScanInfo));
   SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
@@ -683,7 +792,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock*
   }
 
   pInfo->primaryTsIndex = 0;                           // TODO(liuyao) get it from physical plan
-  pInfo->pUpdateInfo = updateInfoInit(60000, 0, 100);  // TODO(liuyao) get it from physical plan
+  pInfo->pUpdateInfo = updateInfoInitP(pInterval, 10000); // TODO(liuyao) get watermark from physical plan
   if (pInfo->pUpdateInfo == NULL) {
     taosMemoryFreeClear(pInfo);
     taosMemoryFreeClear(pOperator);
@@ -693,6 +802,10 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock*
   pInfo->readerHandle = streamReadHandle;
   pInfo->pRes = pResBlock;
   pInfo->pCondition = pCondition;
+  pInfo->pDataReader = pDataReader;
+  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
+  pInfo->pOperatorDumy = pOperatorDumy;
+  pInfo->interval = *pInterval;
 
   pOperator->name = "StreamBlockScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
@@ -1311,7 +1424,6 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
         metaReaderClear(&mr);
 
         colDataAppend(pDst, count, str, false);
-
         //        data = tsdbGetTableTagVal(item->pTable, pExprInfo[j].base.pColumns->info.colId, type, bytes);
         //        dst  = pColInfo->pData + count * pExprInfo[j].base.resSchema.bytes;
         //        doSetTagValueToResultBuf(dst, data, type, bytes);

source/libs/executor/src/timewindowoperator.c

@@ -82,7 +82,7 @@ static void getInitialStartTimeWindow(SInterval* pInterval, int32_t precision, T
 }
 
 // get the correct time window according to the handled timestamp
-static STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts,
+STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts,
                                        SInterval* pInterval, int32_t precision, STimeWindow* win) {
   STimeWindow w = {0};
 
@@ -186,7 +186,7 @@ static FORCE_INLINE int32_t getForwardStepsInBlock(int32_t numOfRows, __block_se
   return forwardStep;
 }
 
-static int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
+int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
   int32_t midPos = -1;
   int32_t numOfRows;
 
@@ -249,7 +249,7 @@ static int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
   return midPos;
 }
 
-static int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos,
+int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos,
                                         TSKEY ekey, __block_search_fn_t searchFn, STableQueryInfo* item,
                                         int32_t order) {
   assert(startPos >= 0 && startPos < pDataBlockInfo->rows);
@@ -775,7 +775,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
 
     //    setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->pCtx, pOperator->numOfExprs);
     // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, true);
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true);
     STableQueryInfo* pTableQueryInfo = pInfo->pCurrent;
 
     setIntervalQueryRange(pTableQueryInfo, pBlock->info.window.skey, &pTaskInfo->window);
@@ -910,7 +910,7 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
       break;
     }
 
-    setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, order, true);
+    setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, order, MAIN_SCAN, true);
     doStateWindowAggImpl(pOperator, pInfo, pBlock);
   }
 
@@ -988,6 +988,20 @@ static void setInverFunction(SqlFunctionCtx* pCtx, int32_t num, EStreamType type
     }
   }
 }
+static void doClearWindows(SIntervalAggOperatorInfo* pInfo, int32_t numOfOutput, SSDataBlock* pBlock) {
+  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
+  TSKEY *tsCols = (TSKEY*)pColDataInfo->pData;
+  int32_t step = 0;
+  for (int32_t i = 0; i < pBlock->info.rows; i += step) {
+    SResultRowInfo dumyInfo;
+    dumyInfo.cur.pageId = -1;
+    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[i], &pInfo->interval,
+                                        pInfo->interval.precision, NULL);
+    step = getNumOfRowsInTimeWindow(&pBlock->info, tsCols, i,
+        win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
+    doClearWindow(pInfo, (char*)&win.skey, sizeof(TKEY), pBlock->info.groupId, numOfOutput);
+  }
+}
 
 static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
   SIntervalAggOperatorInfo* pInfo = pOperator->info;
@@ -1024,10 +1038,14 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
 
     //    setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->pCtx, pOperator->numOfExprs);
     // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, true);
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true);
     if (pInfo->invertible) {
       setInverFunction(pInfo->binfo.pCtx, pOperator->numOfExprs, pBlock->info.type);
     }
+    if (pBlock->info.type == STREAM_REPROCESS) {
+      doClearWindows(pInfo, pOperator->numOfExprs, pBlock);
+      continue;
+    }
     pUpdated = hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0);
   }
 
@@ -1286,7 +1304,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
     }
 
     // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, order, true);
+    setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, order, MAIN_SCAN, true);
     doSessionWindowAggImpl(pOperator, pInfo, pBlock);
   }
 
@@ -1334,7 +1352,7 @@ static SSDataBlock* doAllIntervalAgg(SOperatorInfo* pOperator) {
 
     //    setTagValue(pOperator, pRuntimeEnv->current->pTable, pIntervalInfo->pCtx, pOperator->numOfExprs);
     // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pSliceInfo->binfo.pCtx, pBlock, order, true);
+    setInputDataBlock(pOperator, pSliceInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true);
     //    hashAllIntervalAgg(pOperator, &pSliceInfo->binfo.resultRowInfo, pBlock, 0);
   }
 

source/libs/function/inc/builtinsimpl.h

@@ -98,6 +98,10 @@ int32_t stateDurationFunction(SqlFunctionCtx* pCtx);
 bool getCsumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 int32_t csumFunction(SqlFunctionCtx* pCtx);
 
+bool getMavgFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+bool mavgFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
+int32_t mavgFunction(SqlFunctionCtx* pCtx);
+
 bool getSelectivityFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 
 #ifdef __cplusplus

source/libs/function/src/builtins.c

@@ -339,6 +339,27 @@ static int32_t translateCsum(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateMavg(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  if (2 != LIST_LENGTH(pFunc->pParameterList)) {
+    return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
+  if (QUERY_NODE_COLUMN != nodeType(pPara)) {
+    return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
+                           "The input parameter of MAVG function can only be column");
+  }
+
+  uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
+  uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 1))->resType.type;
+  if (!IS_NUMERIC_TYPE(colType) || !IS_INTEGER_TYPE(paraType)) {
+    return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes, .type = TSDB_DATA_TYPE_DOUBLE};
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t translateLastRow(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   // todo
   return TSDB_CODE_SUCCESS;
@@ -783,6 +804,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .processFunc  = csumFunction,
     .finalizeFunc = NULL
   },
+  {
+    .name = "mavg",
+    .type = FUNCTION_TYPE_MAVG,
+    .classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .translateFunc = translateMavg,
+    .getEnvFunc   = getMavgFuncEnv,
+    .initFunc     = mavgFunctionSetup,
+    .processFunc  = mavgFunction,
+    .finalizeFunc = NULL
+  },
   {
     .name = "abs",
     .type = FUNCTION_TYPE_ABS,

source/libs/function/src/builtinsimpl.c

@@ -21,7 +21,8 @@
 #include "tdatablock.h"
 #include "tpercentile.h"
 
-#define HISTOGRAM_MAX_BINS_NUM   100
+#define HISTOGRAM_MAX_BINS_NUM   1000
+#define MAVG_MAX_POINTS_NUM      1000
 
 typedef struct SSumRes {
   union {
@@ -141,6 +142,14 @@ typedef enum {
   STATE_OPER_EQ,
 } EStateOperType;
 
+typedef struct SMavgInfo {
+  int32_t pos;
+  double  sum;
+  int32_t numOfPoints;
+  bool    pointsMeet;
+  double  points[];
+} SMavgInfo;
+
 #define SET_VAL(_info, numOfElem, res) \
   do {                                 \
     if ((numOfElem) <= 0) {            \
@@ -1644,7 +1653,7 @@ int32_t percentileFunction(SqlFunctionCtx* pCtx) {
       pResInfo->complete = true;
       return 0;
     } else {
-      pInfo->pMemBucket = tMemBucketCreate(pCtx->inputBytes, pCtx->inputType, pInfo->minval, pInfo->maxval);
+      pInfo->pMemBucket = tMemBucketCreate(pCol->info.bytes, type, pInfo->minval, pInfo->maxval);
     }
   }
 
@@ -1695,30 +1704,28 @@ int32_t percentileFunction(SqlFunctionCtx* pCtx) {
         pInfo->numOfElems += 1;
       }
     }
+  } else {
+    // the second stage, calculate the true percentile value
+    int32_t start = pInput->startRowIndex;
+    for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
+      if (colDataIsNull_f(pCol->nullbitmap, i)) {
+        continue;
+      }
 
-    return 0;
-  }
-
-  // the second stage, calculate the true percentile value
-  int32_t start = pInput->startRowIndex;
-  for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
-    if (colDataIsNull_f(pCol->nullbitmap, i)) {
-      continue;
+      char* data = colDataGetData(pCol, i);
+      notNullElems += 1;
+      tMemBucketPut(pInfo->pMemBucket, data, 1);
     }
 
-    char* data = colDataGetData(pCol, i);
-
-    notNullElems += 1;
-    tMemBucketPut(pInfo->pMemBucket, data, 1);
+    SET_VAL(pResInfo, notNullElems, 1);
   }
 
-  SET_VAL(pResInfo, notNullElems, 1);
   return TSDB_CODE_SUCCESS;
 }
 
 int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SVariant* pVal = &pCtx->param[1].param;
-  double    v = pVal->nType == TSDB_DATA_TYPE_INT ? pVal->i : pVal->d;
+  double    v = (pVal->nType == TSDB_DATA_TYPE_BIGINT) ? pVal->i : pVal->d;
 
   SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
   SPercentileInfo*     ppInfo = (SPercentileInfo*)GET_ROWCELL_INTERBUF(pResInfo);
@@ -2949,3 +2956,81 @@ int32_t csumFunction(SqlFunctionCtx* pCtx) {
 
   return numOfElems;
 }
+
+bool getMavgFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
+  pEnv->calcMemSize = sizeof(SMavgInfo) + MAVG_MAX_POINTS_NUM * sizeof(double);
+  return true;
+}
+
+bool mavgFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo *pResultInfo) {
+  if (!functionSetup(pCtx, pResultInfo)) {
+    return false;
+  }
+
+  SMavgInfo *pInfo = GET_ROWCELL_INTERBUF(pResultInfo);
+  pInfo->pos = 0;
+  pInfo->sum = 0;
+  pInfo->numOfPoints = pCtx->param[1].param.i;
+  if (pInfo->numOfPoints < 1 || pInfo->numOfPoints > MAVG_MAX_POINTS_NUM) {
+    return false;
+  }
+  pInfo->pointsMeet = false;
+
+  return true;
+}
+
+int32_t mavgFunction(SqlFunctionCtx* pCtx) {
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  SMavgInfo*           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 numOfElems = 0;
+  int32_t type = pInputCol->info.type;
+  int32_t startOffset = pCtx->offset;
+  for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
+    int32_t pos = startOffset + numOfElems;
+    if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
+      //colDataAppendNULL(pOutput, i);
+      continue;
+    }
+
+    char* data = colDataGetData(pInputCol, i);
+    double v;
+    GET_TYPED_DATA(v, double, type, data);
+
+    if (!pInfo->pointsMeet && (pInfo->pos < pInfo->numOfPoints - 1)) {
+      pInfo->points[pInfo->pos] = v;
+      pInfo->sum += v;
+    } else {
+      if (!pInfo->pointsMeet && (pInfo->pos == pInfo->numOfPoints - 1)) {
+        pInfo->sum +=v;
+        pInfo->pointsMeet = true;
+      } else {
+        pInfo->sum = pInfo->sum + v - pInfo->points[pInfo->pos];
+      }
+
+      pInfo->points[pInfo->pos] = v;
+      double result = pInfo->sum / pInfo->numOfPoints;
+      colDataAppend(pOutput, pos, (char *)&result, false);
+
+      //TODO: remove this after pTsOutput is handled
+      if (pTsOutput != NULL) {
+        colDataAppendInt64(pTsOutput, pos, &tsList[i]);
+      }
+      numOfElems++;
+    }
+
+    pInfo->pos++;
+    if (pInfo->pos == pInfo->numOfPoints) {
+      pInfo->pos = 0;
+    }
+  }
+
+  return numOfElems;
+}

source/libs/function/src/texpr.c

@@ -27,19 +27,6 @@
 #include "tvariant.h"
 #include "tdef.h"
 
-//static uint8_t UNUSED_FUNC isQueryOnPrimaryKey(const char *primaryColumnName, const tExprNode *pLeft, const tExprNode *pRight) {
-//  if (pLeft->nodeType == TEXPR_COL_NODE) {
-//    // if left node is the primary column,return true
-//    return (strcmp(primaryColumnName, pLeft->pSchema->name) == 0) ? 1 : 0;
-//  } else {
-//    // if any children have query on primary key, their parents are also keep this value
-//    return ((pLeft->nodeType == TEXPR_BINARYEXPR_NODE && pLeft->_node.hasPK == 1) ||
-//            (pRight->nodeType == TEXPR_BINARYEXPR_NODE && pRight->_node.hasPK == 1)) == true
-//               ? 1
-//               : 0;
-//  }
-//}
-
 static void doExprTreeDestroy(tExprNode **pExpr, void (*fp)(void *));
 
 void tExprTreeDestroy(tExprNode *pNode, void (*fp)(void *)) {
@@ -64,21 +51,7 @@ static void doExprTreeDestroy(tExprNode **pExpr, void (*fp)(void *)) {
   }
 
   int32_t type = (*pExpr)->nodeType;
-  if (type == TEXPR_BINARYEXPR_NODE) {
-    doExprTreeDestroy(&(*pExpr)->_node.pLeft, fp);
-    doExprTreeDestroy(&(*pExpr)->_node.pRight, fp);
-  
-    if (fp != NULL) {
-      fp((*pExpr)->_node.info);
-    }
-  } else if (type == TEXPR_UNARYEXPR_NODE) {
-    doExprTreeDestroy(&(*pExpr)->_node.pLeft, fp);
-    if (fp != NULL) {
-      fp((*pExpr)->_node.info);
-    }
-
-    assert((*pExpr)->_node.pRight == NULL);
-  } else if (type == TEXPR_VALUE_NODE) {
+  if (type == TEXPR_VALUE_NODE) {
     taosVariantDestroy((*pExpr)->pVal);
     taosMemoryFree((*pExpr)->pVal);
   } else if (type == TEXPR_COL_NODE) {
@@ -90,9 +63,7 @@ static void doExprTreeDestroy(tExprNode **pExpr, void (*fp)(void *)) {
 }
 
 bool exprTreeApplyFilter(tExprNode *pExpr, const void *pItem, SExprTraverseSupp *param) {
-  tExprNode *pLeft  = pExpr->_node.pLeft;
-  tExprNode *pRight = pExpr->_node.pRight;
-
+#if 0
   //non-leaf nodes, recursively traverse the expression tree in the post-root order
   if (pLeft->nodeType == TEXPR_BINARYEXPR_NODE && pRight->nodeType == TEXPR_BINARYEXPR_NODE) {
     if (pExpr->_node.optr == LOGIC_COND_TYPE_OR) {  // or
@@ -114,6 +85,9 @@ bool exprTreeApplyFilter(tExprNode *pExpr, const void *pItem, SExprTraverseSupp
   // handle the leaf node
   param->setupInfoFn(pExpr, param->pExtInfo);
   return param->nodeFilterFn(pItem, pExpr->_node.info);
+#endif
+
+  return 0;
 }
 
 // TODO: these three functions should be made global
@@ -141,59 +115,6 @@ static UNUSED_FUNC char* exception_strdup(const char* str) {
   return p;
 }
 
-static tExprNode* exprTreeFromBinaryImpl(SBufferReader* br) {
-  int32_t anchor = CLEANUP_GET_ANCHOR();
-  if (CLEANUP_EXCEED_LIMIT()) {
-    THROW(TSDB_CODE_QRY_EXCEED_TAGS_LIMIT);
-    return NULL;
-  }
-
-  tExprNode* pExpr = exception_calloc(1, sizeof(tExprNode));
-  CLEANUP_PUSH_VOID_PTR_PTR(true, tExprTreeDestroy, pExpr, NULL);
-  pExpr->nodeType = tbufReadUint8(br);
-  
-  if (pExpr->nodeType == TEXPR_VALUE_NODE) {
-    SVariant* pVal = exception_calloc(1, sizeof(SVariant));
-    pExpr->pVal = pVal;
-  
-    pVal->nType = tbufReadUint32(br);
-    if (pVal->nType == TSDB_DATA_TYPE_BINARY) {
-      tbufReadToBuffer(br, &pVal->nLen, sizeof(pVal->nLen));
-      pVal->pz = taosMemoryCalloc(1, pVal->nLen + 1);
-      tbufReadToBuffer(br, pVal->pz, pVal->nLen);
-    } else {
-      pVal->i = tbufReadInt64(br);
-    }
-    
-  } else if (pExpr->nodeType == TEXPR_COL_NODE) {
-    SSchema* pSchema = exception_calloc(1, sizeof(SSchema));
-    pExpr->pSchema = pSchema;
-
-    pSchema->colId = tbufReadInt16(br);
-    pSchema->bytes = tbufReadInt16(br);
-    pSchema->type = tbufReadUint8(br);
-    tbufReadToString(br, pSchema->name, TSDB_COL_NAME_LEN);
-    
-  } else if (pExpr->nodeType == TEXPR_BINARYEXPR_NODE) {
-    pExpr->_node.optr = tbufReadUint8(br);
-    pExpr->_node.pLeft = exprTreeFromBinaryImpl(br);
-    pExpr->_node.pRight = exprTreeFromBinaryImpl(br);
-    assert(pExpr->_node.pLeft != NULL && pExpr->_node.pRight != NULL);
-  }
-  
-  CLEANUP_EXECUTE_TO(anchor, false);
-  return pExpr;
-}
-
-tExprNode* exprTreeFromBinary(const void* data, size_t size) {
-  if (size == 0) {
-    return NULL;
-  }
-
-  SBufferReader br = tbufInitReader(data, size, false);
-  return exprTreeFromBinaryImpl(&br);
-}
-
 void buildFilterSetFromBinary(void **q, const char *buf, int32_t len) {
   SBufferReader br = tbufInitReader(buf, len, false); 
   uint32_t type  = tbufReadUint32(&br);     
@@ -405,38 +326,3 @@ err_ret:
   taosHashCleanup(pObj);
   taosMemoryFreeClear(tmp);
 }
-
-tExprNode* exprdup(tExprNode* pNode) {
-  if (pNode == NULL) {
-    return NULL;
-  }
-
-  tExprNode* pCloned = taosMemoryCalloc(1, sizeof(tExprNode));
-  if (pNode->nodeType == TEXPR_BINARYEXPR_NODE) {
-    tExprNode* pLeft  = exprdup(pNode->_node.pLeft);
-    tExprNode* pRight = exprdup(pNode->_node.pRight);
-
-    pCloned->_node.pLeft  = pLeft;
-    pCloned->_node.pRight = pRight;
-    pCloned->_node.optr  = pNode->_node.optr;
-  } else if (pNode->nodeType == TEXPR_VALUE_NODE) {
-    pCloned->pVal = taosMemoryCalloc(1, sizeof(SVariant));
-    taosVariantAssign(pCloned->pVal, pNode->pVal);
-  } else if (pNode->nodeType == TEXPR_COL_NODE) {
-    pCloned->pSchema = taosMemoryCalloc(1, sizeof(SSchema));
-    *pCloned->pSchema = *pNode->pSchema;
-  } else if (pNode->nodeType == TEXPR_FUNCTION_NODE) {
-    strcpy(pCloned->_function.functionName, pNode->_function.functionName);
-
-    int32_t num = pNode->_function.num;
-    pCloned->_function.num = num;
-    pCloned->_function.pChild = taosMemoryCalloc(num, POINTER_BYTES);
-    for(int32_t i = 0; i < num; ++i) {
-      pCloned->_function.pChild[i] = exprdup(pNode->_function.pChild[i]);
-    }
-  }
-
-  pCloned->nodeType = pNode->nodeType;
-  return pCloned;
-}
-

source/libs/nodes/src/nodesCodeFuncs.c

@@ -1142,9 +1142,9 @@ static int32_t jsonToPhysiTableScanNode(const SJson* pJson, void* pObj) {
   return code;
 }
 
-static int32_t physiStreamScanNodeToJson(const void* pObj, SJson* pJson) { return physiScanNodeToJson(pObj, pJson); }
+static int32_t physiStreamScanNodeToJson(const void* pObj, SJson* pJson) { return physiTableScanNodeToJson(pObj, pJson); }
 
-static int32_t jsonToPhysiStreamScanNode(const SJson* pJson, void* pObj) { return jsonToPhysiScanNode(pJson, pObj); }
+static int32_t jsonToPhysiStreamScanNode(const SJson* pJson, void* pObj) { return jsonToPhysiTableScanNode(pJson, pObj); }
 
 static const char* jkSysTableScanPhysiPlanMnodeEpSet = "MnodeEpSet";
 static const char* jkSysTableScanPhysiPlanShowRewrite = "ShowRewrite";

source/libs/parser/inc/parAst.h

@@ -32,6 +32,7 @@ typedef struct SAstCreateContext {
   bool           notSupport;
   SNode*         pRootNode;
   int16_t        placeholderNo;
+  SArray*        pPlaceholderValues;
   int32_t        errCode;
 } SAstCreateContext;
 

source/libs/parser/src/parAstCreater.c

@@ -44,6 +44,7 @@ void initAstCreateContext(SParseContext* pParseCxt, SAstCreateContext* pCxt) {
   pCxt->notSupport = false;
   pCxt->pRootNode = NULL;
   pCxt->placeholderNo = 0;
+  pCxt->pPlaceholderValues = NULL;
   pCxt->errCode = TSDB_CODE_SUCCESS;
 }
 
@@ -299,6 +300,14 @@ SNode* createPlaceholderValueNode(SAstCreateContext* pCxt, const SToken* pLitera
   val->literal = strndup(pLiteral->z, pLiteral->n);
   CHECK_OUT_OF_MEM(val->literal);
   val->placeholderNo = ++pCxt->placeholderNo;
+  if (NULL == pCxt->pPlaceholderValues) {
+    pCxt->pPlaceholderValues = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
+    if (NULL == pCxt->pPlaceholderValues) {
+      nodesDestroyNode(val);
+      return NULL;
+    }
+  }
+  taosArrayPush(pCxt->pPlaceholderValues, &val);
   return (SNode*)val;
 }
 

source/libs/parser/src/parAstParser.c

@@ -81,6 +81,8 @@ abort_parse:
     }
     (*pQuery)->pRoot = cxt.pRootNode;
     (*pQuery)->placeholderNum = cxt.placeholderNo;
+    TSWAP((*pQuery)->pPlaceholderValues, cxt.pPlaceholderValues);
   }
+  taosArrayDestroy(cxt.pPlaceholderValues);
   return cxt.errCode;
 }

source/libs/parser/src/parInsert.c

@@ -1069,8 +1069,9 @@ static void destroyInsertParseContext(SInsertParseContext* pCxt) {
 //   [...];
 static int32_t parseInsertBody(SInsertParseContext* pCxt) {
   int32_t tbNum = 0;
-  char    tbFName[TSDB_TABLE_FNAME_LEN];
-  bool    autoCreateTbl = false;
+  char tbFName[TSDB_TABLE_FNAME_LEN];
+  bool autoCreateTbl = false;
+  STableMeta *pMeta = NULL;
 
   // for each table
   while (1) {
@@ -1127,10 +1128,12 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
     CHECK_CODE(getDataBlockFromList(pCxt->pTableBlockHashObj, tbFName, strlen(tbFName), TSDB_DEFAULT_PAYLOAD_SIZE,
                                     sizeof(SSubmitBlk), getTableInfo(pCxt->pTableMeta).rowSize, pCxt->pTableMeta,
                                     &dataBuf, NULL, &pCxt->createTblReq));
-
+    pMeta = pCxt->pTableMeta;
+    pCxt->pTableMeta = NULL;
+    
     if (TK_NK_LP == sToken.type) {
       // pSql -> field1_name, ...)
-      CHECK_CODE(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pCxt->pTableMeta)));
+      CHECK_CODE(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pMeta)));
       NEXT_TOKEN(pCxt->pSql, sToken);
     }
 
@@ -1166,8 +1169,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
       return TSDB_CODE_TSC_OUT_OF_MEMORY;
     }
     memcpy(tags, &pCxt->tags, sizeof(pCxt->tags));
-    (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pTableMeta, tags, tbFName, autoCreateTbl,
-                                pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj);
+    (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pMeta, tags, tbFName, autoCreateTbl, pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj);
 
     memset(&pCxt->tags, 0, sizeof(pCxt->tags));
     pCxt->pVgroupsHashObj = NULL;
@@ -1677,8 +1679,8 @@ static int32_t smlBuildTagRow(SArray* cols, SKVRowBuilder* tagsBuilder, SParsedD
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t smlBindData(void* handle, SArray* tags, SArray* colsFormat, SArray* colsSchema, SArray* cols, bool format,
-                    STableMeta* pTableMeta, char* tableName, char* msgBuf, int16_t msgBufLen) {
+int32_t smlBindData(void *handle, SArray *tags, SArray *colsSchema, SArray *cols, bool format,
+                    STableMeta *pTableMeta, char *tableName, char *msgBuf, int16_t msgBufLen) {
   SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};
 
   SSmlExecHandle* smlHandle = (SSmlExecHandle*)handle;
@@ -1720,8 +1722,8 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsFormat, SArray* cols
 
   initRowBuilder(&pDataBlock->rowBuilder, pDataBlock->pTableMeta->sversion, &pDataBlock->boundColumnInfo);
 
-  int32_t rowNum = format ? taosArrayGetSize(colsFormat) : taosArrayGetSize(cols);
-  if (rowNum <= 0) {
+  int32_t rowNum = taosArrayGetSize(cols);
+  if(rowNum <= 0) {
     return buildInvalidOperationMsg(&pBuf, "cols size <= 0");
   }
   ret = allocateMemForSize(pDataBlock, extendedRowSize * rowNum);
@@ -1732,13 +1734,10 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsFormat, SArray* cols
   for (int32_t r = 0; r < rowNum; ++r) {
     STSRow* row = (STSRow*)(pDataBlock->pData + pDataBlock->size);  // skip the SSubmitBlk header
     tdSRowResetBuf(pBuilder, row);
-    void*  rowData = NULL;
+    void *rowData = taosArrayGetP(cols, r);
     size_t rowDataSize = 0;
-    if (format) {
-      rowData = taosArrayGetP(colsFormat, r);
+    if(format){
       rowDataSize = taosArrayGetSize(rowData);
-    } else {
-      rowData = taosArrayGetP(cols, r);
     }
 
     // 1. set the parsed value from sql string

source/libs/parser/src/parser.c

@@ -39,16 +39,99 @@ static int32_t parseSqlIntoAst(SParseContext* pCxt, SQuery** pQuery) {
   if (TSDB_CODE_SUCCESS == code) {
     code = authenticate(pCxt, *pQuery);
   }
-  if (TSDB_CODE_SUCCESS == code) {
+  if (TSDB_CODE_SUCCESS == code && 0 == (*pQuery)->placeholderNum) {
     code = translate(pCxt, *pQuery);
   }
-  if (TSDB_CODE_SUCCESS == code) {
+  if (TSDB_CODE_SUCCESS == code && 0 == (*pQuery)->placeholderNum) {
     code = calculateConstant(pCxt, *pQuery);
   }
   return code;
 }
 
-int32_t qParseQuerySql(SParseContext* pCxt, SQuery** pQuery) {
+static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) {
+  if (pParam->is_null && 1 == *(pParam->is_null)) {
+    pVal->node.resType.type = TSDB_DATA_TYPE_NULL;
+    pVal->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_NULL].bytes;
+    return TSDB_CODE_SUCCESS;
+  }
+  int32_t inputSize = (NULL != pParam->length ? *(pParam->length) : tDataTypes[pParam->buffer_type].bytes);
+  pVal->node.resType.type = pParam->buffer_type;
+  pVal->node.resType.bytes = inputSize;
+  switch (pParam->buffer_type) {
+    case TSDB_DATA_TYPE_BOOL:
+      pVal->datum.b = *((bool*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_TINYINT:
+      pVal->datum.i = *((int8_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_SMALLINT:
+      pVal->datum.i = *((int16_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_INT:
+      pVal->datum.i = *((int32_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_BIGINT:
+      pVal->datum.i = *((int64_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_FLOAT:
+      pVal->datum.d = *((float*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_DOUBLE:
+      pVal->datum.d = *((double*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_VARCHAR:
+    case TSDB_DATA_TYPE_VARBINARY:
+      pVal->datum.p = taosMemoryCalloc(1, pVal->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
+      if (NULL == pVal->datum.p) {
+        return TSDB_CODE_OUT_OF_MEMORY;
+      }
+      varDataSetLen(pVal->datum.p, pVal->node.resType.bytes);
+      strncpy(varDataVal(pVal->datum.p), (const char*)pParam->buffer, pVal->node.resType.bytes);
+      break;
+    case TSDB_DATA_TYPE_NCHAR: {
+      pVal->node.resType.bytes *= TSDB_NCHAR_SIZE;
+      pVal->datum.p = taosMemoryCalloc(1, pVal->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
+      if (NULL == pVal->datum.p) {
+        return TSDB_CODE_OUT_OF_MEMORY;
+      }
+
+      int32_t output = 0;
+      if (!taosMbsToUcs4(pParam->buffer, inputSize, (TdUcs4*)varDataVal(pVal->datum.p), pVal->node.resType.bytes,
+                         &output)) {
+        return errno;
+      }
+      varDataSetLen(pVal->datum.p, output);
+      pVal->node.resType.bytes = output;
+      break;
+    }
+    case TSDB_DATA_TYPE_TIMESTAMP:
+      pVal->datum.i = *((int64_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_UTINYINT:
+      pVal->datum.u = *((uint8_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_USMALLINT:
+      pVal->datum.u = *((uint16_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_UINT:
+      pVal->datum.u = *((uint32_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_UBIGINT:
+      pVal->datum.u = *((uint64_t*)pParam->buffer);
+      break;
+    case TSDB_DATA_TYPE_JSON:
+    case TSDB_DATA_TYPE_DECIMAL:
+    case TSDB_DATA_TYPE_BLOB:
+    case TSDB_DATA_TYPE_MEDIUMBLOB:
+      // todo
+    default:
+      break;
+  }
+  pVal->translate = true;
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery) {
   int32_t code = TSDB_CODE_SUCCESS;
   if (isInsertSql(pCxt->pSql, pCxt->sqlLen)) {
     code = parseInsertSql(pCxt, pQuery);
@@ -77,3 +160,29 @@ void qDestroyQuery(SQuery* pQueryNode) {
 int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) {
   return extractResultSchema(pRoot, numOfCols, pSchema);
 }
+
+int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId) {
+  int32_t code = TSDB_CODE_SUCCESS;
+
+  if (colIdx < 0) {
+    int32_t size = taosArrayGetSize(pQuery->pPlaceholderValues);
+    for (int32_t i = 0; i < size; ++i) {
+      code = setValueByBindParam((SValueNode*)taosArrayGetP(pQuery->pPlaceholderValues, i), pParams + i);
+      if (TSDB_CODE_SUCCESS != code) {
+        return code;
+      }
+    }
+  } else {
+    code = setValueByBindParam((SValueNode*)taosArrayGetP(pQuery->pPlaceholderValues, colIdx), pParams);
+  }
+
+  return code;
+}
+
+int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery) {
+  int32_t code = translate(pCxt, pQuery);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = calculateConstant(pCxt, pQuery);
+  }
+  return code;
+}

source/libs/planner/src/planPhysiCreater.c

@@ -460,9 +460,13 @@ static int32_t createTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubp
   memcpy(pTableScan->scanSeq, pScanLogicNode->scanSeq, sizeof(pScanLogicNode->scanSeq));
   pTableScan->scanRange = pScanLogicNode->scanRange;
   pTableScan->ratio = pScanLogicNode->ratio;
-  vgroupInfoToNodeAddr(pScanLogicNode->pVgroupList->vgroups, &pSubplan->execNode);
-  taosArrayPush(pCxt->pExecNodeList, &pSubplan->execNode);
-  pSubplan->execNodeStat.tableNum = pScanLogicNode->pVgroupList->vgroups[0].numOfTable;
+  if (pScanLogicNode->pVgroupList) {
+    vgroupInfoToNodeAddr(pScanLogicNode->pVgroupList->vgroups, &pSubplan->execNode);
+    pSubplan->execNodeStat.tableNum = pScanLogicNode->pVgroupList->vgroups[0].numOfTable;
+  }
+  if (pCxt->pExecNodeList) {
+    taosArrayPush(pCxt->pExecNodeList, &pSubplan->execNode);
+  }
   tNameGetFullDbName(&pScanLogicNode->tableName, pSubplan->dbFName);
   pTableScan->dataRequired = pScanLogicNode->dataRequired;
   pTableScan->pDynamicScanFuncs = nodesCloneList(pScanLogicNode->pDynamicScanFuncs);
@@ -505,13 +509,12 @@ static int32_t createSystemTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan*
 
 static int32_t createStreamScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubplan, SScanLogicNode* pScanLogicNode,
                                          SPhysiNode** pPhyNode) {
-  SStreamScanPhysiNode* pScan =
-      (SStreamScanPhysiNode*)makePhysiNode(pCxt, pScanLogicNode->pMeta->tableInfo.precision,
-                                           (SLogicNode*)pScanLogicNode, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN);
-  if (NULL == pScan) {
-    return TSDB_CODE_OUT_OF_MEMORY;
+  int32_t res = createTableScanPhysiNode(pCxt, pSubplan, pScanLogicNode, pPhyNode);
+  if (res == TSDB_CODE_SUCCESS) {
+    ENodeType type = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
+    setNodeType(*pPhyNode, type);
   }
-  return createScanPhysiNodeFinalize(pCxt, pScanLogicNode, (SScanPhysiNode*)pScan, pPhyNode);
+  return res;
 }
 
 static int32_t createScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubplan, SScanLogicNode* pScanLogicNode,
@@ -786,7 +789,7 @@ static int32_t doCreateExchangePhysiNode(SPhysiPlanContext* pCxt, SExchangeLogic
 }
 static int32_t createStreamScanPhysiNodeByExchange(SPhysiPlanContext* pCxt, SExchangeLogicNode* pExchangeLogicNode,
                                                    SPhysiNode** pPhyNode) {
-  SStreamScanPhysiNode* pScan = (SStreamScanPhysiNode*)makePhysiNode(
+  SScanPhysiNode* pScan = (SScanPhysiNode*)makePhysiNode(
       pCxt, pExchangeLogicNode->precision, (SLogicNode*)pExchangeLogicNode, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN);
   if (NULL == pScan) {
     return TSDB_CODE_OUT_OF_MEMORY;

source/libs/planner/src/planner.c

@@ -224,7 +224,7 @@ static int32_t calcConstList(SNodeList* pList) {
 }
 
 static bool isEmptyResultCond(SNode** pCond) {
-  if (QUERY_NODE_VALUE != nodeType(*pCond)) {
+  if (NULL == *pCond || QUERY_NODE_VALUE != nodeType(*pCond)) {
     return false;
   }
   if (((SValueNode*)*pCond)->datum.b) {

source/libs/planner/test/planStmtTest.cpp

@@ -50,5 +50,5 @@ class PlanStmtTest : public PlannerTestBase {
 TEST_F(PlanStmtTest, stmt) {
   useDb("root", "test");
 
-  run("select * from t1 where c1 = ?");
+  // run("select * from t1 where c1 = ?");
 }

source/libs/planner/test/planTestUtil.cpp

@@ -198,7 +198,7 @@ class PlannerTestBaseImpl {
     cxt.pMsg = stmtEnv_.msgBuf_.data();
     cxt.msgLen = stmtEnv_.msgBuf_.max_size();
 
-    DO_WITH_THROW(qParseQuerySql, &cxt, pQuery);
+    DO_WITH_THROW(qParseSql, &cxt, pQuery);
     res_.ast_ = toString((*pQuery)->pRoot);
   }
 

source/libs/scalar/CMakeLists.txt

@@ -8,7 +8,7 @@ target_include_directories(
 )
 
 target_link_libraries(scalar
-        PRIVATE os util common nodes function qcom
+        PRIVATE os util common nodes function qcom vnode
         )
 
 if(${BUILD_TEST})

source/libs/scalar/inc/sclInt.h

@@ -26,6 +26,7 @@ typedef struct SScalarCtx {
   int32_t      code;
   SArray      *pBlockList;  /* element is SSDataBlock* */
   SHashObj    *pRes;  /* element is SScalarParam */
+  void        *param;      // additional parameter (meta actually) for acquire value such as tbname/tags values
 } SScalarCtx;
 
 
@@ -49,6 +50,7 @@ typedef struct SScalarCtx {
 
 int32_t doConvertDataType(SValueNode* pValueNode, SScalarParam* out);
 SColumnInfoData* createColumnInfoData(SDataType* pType, int32_t numOfRows);
+void sclConvertToTsValueNode(int8_t precision, SValueNode* valueNode);
 
 #define GET_PARAM_TYPE(_c)      ((_c)->columnData->info.type)
 #define GET_PARAM_BYTES(_c)     ((_c)->columnData->info.bytes)

source/libs/scalar/src/filter.c

@@ -3505,19 +3505,6 @@ int32_t fltAddValueNodeToConverList(SFltTreeStat *stat, SValueNode* pNode) {
   return TSDB_CODE_SUCCESS;
 }
 
-void fltConvertToTsValueNode(SFltTreeStat *stat, SValueNode* valueNode) {
-  char *timeStr = valueNode->datum.p;
-  if (convertStringToTimestamp(valueNode->node.resType.type, valueNode->datum.p, stat->precision, &valueNode->datum.i) !=
-      TSDB_CODE_SUCCESS) {
-    valueNode->datum.i = 0;
-  }
-  taosMemoryFree(timeStr);
-
-  valueNode->typeData = valueNode->datum.i;
-  valueNode->node.resType.type = TSDB_DATA_TYPE_TIMESTAMP;
-  valueNode->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes;
-}
-
 EDealRes fltReviseRewriter(SNode** pNode, void* pContext) {
   SFltTreeStat *stat = (SFltTreeStat *)pContext;
 
@@ -3566,7 +3553,7 @@ EDealRes fltReviseRewriter(SNode** pNode, void* pContext) {
       return DEAL_RES_CONTINUE;
     }
 
-    fltConvertToTsValueNode(stat, valueNode);
+    sclConvertToTsValueNode(stat->precision, valueNode);
 
     return DEAL_RES_CONTINUE;
   }
@@ -3614,6 +3601,11 @@ EDealRes fltReviseRewriter(SNode** pNode, void* pContext) {
       return DEAL_RES_CONTINUE;
     }
 
+    if (FILTER_GET_FLAG(stat->info->options, FLT_OPTION_TIMESTAMP) && node->opType >= OP_TYPE_NOT_EQUAL) {
+      stat->scalarMode = true;
+      return DEAL_RES_CONTINUE;
+    }
+
     if (NULL == node->pRight) {
       if (scalarGetOperatorParamNum(node->opType) > 1) {
         fltError("invalid operator, pRight:%p, nodeType:%d, opType:%d", node->pRight, nodeType(node), node->opType);
@@ -3695,7 +3687,7 @@ int32_t fltReviseNodes(SFilterInfo *pInfo, SNode** pNode, SFltTreeStat *pStat) {
   for (int32_t i = 0; i < nodeNum; ++i) {
     SValueNode *valueNode = *(SValueNode **)taosArrayGet(pStat->nodeList, i);
     
-    fltConvertToTsValueNode(pStat, valueNode);
+    sclConvertToTsValueNode(pStat->precision, valueNode);
   }
 
 _return:

source/libs/scalar/src/scalar.c

@@ -8,6 +8,7 @@
 #include "tdatablock.h"
 #include "scalar.h"
 #include "tudf.h"
+#include "ttime.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 
@@ -19,6 +20,19 @@ int32_t scalarGetOperatorParamNum(EOperatorType type) {
   return 2;
 }
 
+void sclConvertToTsValueNode(int8_t precision, SValueNode* valueNode) {
+  char *timeStr = valueNode->datum.p;
+  if (convertStringToTimestamp(valueNode->node.resType.type, valueNode->datum.p, precision, &valueNode->datum.i) !=
+      TSDB_CODE_SUCCESS) {
+    valueNode->datum.i = 0;
+  }
+  taosMemoryFree(timeStr);
+  
+  valueNode->node.resType.type = TSDB_DATA_TYPE_TIMESTAMP;
+  valueNode->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes;
+}
+
+
 SColumnInfoData* createColumnInfoData(SDataType* pType, int32_t numOfRows) {
   SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));
   if (pColumnData == NULL) {
@@ -251,6 +265,7 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
     *rowNum = param->numOfRows;
   }
 
+  param->param = ctx->param;
   return TSDB_CODE_SUCCESS;
 }
 
@@ -535,7 +550,7 @@ EDealRes sclRewriteBasedOnOptr(SNode** pNode, SScalarCtx *ctx, EOperatorType opT
 }
 
 
-EDealRes sclRewriteOperatorForNullValue(SNode** pNode, SScalarCtx *ctx) {
+EDealRes sclRewriteNonConstOperator(SNode** pNode, SScalarCtx *ctx) {
   SOperatorNode *node = (SOperatorNode *)*pNode;
 
   if (node->pLeft && (QUERY_NODE_VALUE == nodeType(node->pLeft))) {
@@ -543,6 +558,11 @@ EDealRes sclRewriteOperatorForNullValue(SNode** pNode, SScalarCtx *ctx) {
     if (SCL_IS_NULL_VALUE_NODE(valueNode) && (node->opType != OP_TYPE_IS_NULL && node->opType != OP_TYPE_IS_NOT_NULL)) {
       return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
     }
+
+    if (IS_STR_DATA_TYPE(valueNode->node.resType.type) && node->pRight && nodesIsExprNode(node->pRight) 
+      && ((SExprNode*)node->pRight)->resType.type == TSDB_DATA_TYPE_TIMESTAMP) {
+      sclConvertToTsValueNode(((SExprNode*)node->pRight)->resType.precision, valueNode);
+    }
   }
 
   if (node->pRight && (QUERY_NODE_VALUE == nodeType(node->pRight))) {
@@ -550,6 +570,11 @@ EDealRes sclRewriteOperatorForNullValue(SNode** pNode, SScalarCtx *ctx) {
     if (SCL_IS_NULL_VALUE_NODE(valueNode) && (node->opType != OP_TYPE_IS_NULL && node->opType != OP_TYPE_IS_NOT_NULL)) {
       return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
     }
+
+    if (IS_STR_DATA_TYPE(valueNode->node.resType.type) && node->pLeft && nodesIsExprNode(node->pLeft) 
+      && ((SExprNode*)node->pLeft)->resType.type == TSDB_DATA_TYPE_TIMESTAMP) {
+      sclConvertToTsValueNode(((SExprNode*)node->pLeft)->resType.precision, valueNode);
+    }
   }
 
   if (node->pRight && (QUERY_NODE_NODE_LIST == nodeType(node->pRight))) {
@@ -672,7 +697,7 @@ EDealRes sclRewriteOperator(SNode** pNode, SScalarCtx *ctx) {
   SOperatorNode *node = (SOperatorNode *)*pNode;
 
   if ((!SCL_IS_CONST_NODE(node->pLeft)) || (!SCL_IS_CONST_NODE(node->pRight))) {
-    return sclRewriteOperatorForNullValue(pNode, ctx);
+    return sclRewriteNonConstOperator(pNode, ctx);
   }
 
   SScalarParam output = {.columnData = taosMemoryCalloc(1, sizeof(SColumnInfoData))};
@@ -885,7 +910,7 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
   }
 
   int32_t code = 0;
-  SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList};
+  SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList, .param = pDst->param};
 
   // TODO: OPT performance
   ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);

source/libs/scalar/src/sclfunc.c

 #include "function.h"
 #include "scalar.h"
-#include "tdatablock.h"
-#include "ttime.h"
 #include "sclInt.h"
 #include "sclvector.h"
+#include "tdatablock.h"
 #include "tjson.h"
+#include "ttime.h"
+#include "vnode.h"
 
 typedef float (*_float_fn)(float);
 typedef double (*_double_fn)(double);
@@ -1512,6 +1513,21 @@ int32_t winEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
 
 int32_t qTbnameFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
   ASSERT(inputNum == 1);
-  colDataAppend(pOutput->columnData, pOutput->numOfRows, colDataGetData(pInput->columnData, 0), false);
+
+  SMetaReader mr = {0};
+  metaReaderInit(&mr, pInput->param, 0);
+
+  uint64_t uid = *(uint64_t *)colDataGetData(pInput->columnData, 0);
+  metaGetTableEntryByUid(&mr, uid);
+
+  char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
+  STR_TO_VARSTR(str, mr.me.name);
+  metaReaderClear(&mr);
+
+  for(int32_t i = 0; i < pInput->numOfRows; ++i) {
+    colDataAppend(pOutput->columnData, pOutput->numOfRows + i, str, false);
+  }
+
+  pOutput->numOfRows += pInput->numOfRows;
   return TSDB_CODE_SUCCESS;
 }

source/libs/stream/src/tstream.c

@@ -154,7 +154,7 @@ int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, in
 
   // sink
   if (pTask->sinkType == TASK_SINK__TABLE) {
-    /*blockDebugShowData(pRes);*/
+    blockDebugShowData(pRes);
     pTask->tbSink.tbSinkFunc(pTask, pTask->tbSink.vnode, 0, pRes);
   } else if (pTask->sinkType == TASK_SINK__SMA) {
     pTask->smaSink.smaSink(pTask->ahandle, pTask->smaSink.smaId, pRes);

source/libs/wal/src/walRead.c

@@ -54,8 +54,8 @@ int32_t walRegisterRead(SWalReadHandle *pRead, int64_t ver) {
   return 0;
 }
 
-static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, int64_t ver) {
-  int ret = 0;
+static int64_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, int64_t ver) {
+  int64_t ret = 0;
 
   TdFilePtr pIdxTFile = pRead->pReadIdxTFile;
   TdFilePtr pLogTFile = pRead->pReadLogTFile;
@@ -68,14 +68,14 @@ static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, i
     wError("failed to seek idx file, ver %ld, pos: %ld, since %s", ver, offset, terrstr());
     return -1;
   }
-  SWalIdxEntry entry;
+  SWalIdxEntry entry = {0};
   if ((ret = taosReadFile(pIdxTFile, &entry, sizeof(SWalIdxEntry))) != sizeof(SWalIdxEntry)) {
     if (ret < 0) {
       terrno = TAOS_SYSTEM_ERROR(errno);
       wError("failed to read idx file, since %s", terrstr());
     } else {
       terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
-      wError("read idx file incompletely, read bytes %d, bytes should be %lu", ret, sizeof(SWalIdxEntry));
+      wError("read idx file incompletely, read bytes %ld, bytes should be %lu", ret, sizeof(SWalIdxEntry));
     }
     return -1;
   }
@@ -156,7 +156,7 @@ static int32_t walReadSeekVer(SWalReadHandle *pRead, int64_t ver) {
 void walSetReaderCapacity(SWalReadHandle *pRead, int32_t capacity) { pRead->capacity = capacity; }
 
 int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead) {
-  int32_t code;
+  int64_t code;
 
   // TODO: valid ver
   if (ver > pRead->pWal->vers.commitVer) {
@@ -187,7 +187,7 @@ int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead) {
 }
 
 int32_t walSkipFetchBody(SWalReadHandle *pRead, const SWalHead *pHead) {
-  int32_t code;
+  int64_t code;
 
   ASSERT(pRead->curVersion == pHead->head.version);
 
@@ -214,23 +214,24 @@ int32_t walFetchBody(SWalReadHandle *pRead, SWalHead **ppHead) {
       return -1;
     }
     *ppHead = ptr;
+    pReadHead = &((*ppHead)->head);
     pRead->capacity = pReadHead->bodyLen;
   }
 
   if (pReadHead->bodyLen != taosReadFile(pRead->pReadLogTFile, pReadHead->body, pReadHead->bodyLen)) {
+    ASSERT(0);
     return -1;
   }
 
   if (pReadHead->version != ver) {
-    wError("unexpected wal log version: %" PRId64 ", read request version:%" PRId64 "", pRead->pHead->head.version,
-           ver);
+    wError("wal fetch body error: %" PRId64 ", read request version:%" PRId64 "", pRead->pHead->head.version, ver);
     pRead->curVersion = -1;
     terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
     return -1;
   }
 
   if (walValidBodyCksum(*ppHead) != 0) {
-    wError("unexpected wal log version: % " PRId64 ", since body checksum not passed", ver);
+    wError("wal fetch body error: % " PRId64 ", since body checksum not passed", ver);
     pRead->curVersion = -1;
     terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
     return -1;
@@ -257,7 +258,7 @@ int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **p
 }
 
 int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) {
-  int code;
+  int64_t code;
   // TODO: check wal life
   if (pRead->curVersion != ver) {
     if (walReadSeekVer(pRead, ver) < 0) {

source/libs/wal/src/walSeek.c

@@ -19,8 +19,8 @@
 #include "tref.h"
 #include "walInt.h"
 
-static int walSeekWritePos(SWal* pWal, int64_t ver) {
-  int code = 0;
+static int64_t walSeekWritePos(SWal* pWal, int64_t ver) {
+  int64_t code = 0;
 
   TdFilePtr pIdxTFile = pWal->pWriteIdxTFile;
   TdFilePtr pLogTFile = pWal->pWriteLogTFile;
@@ -45,7 +45,7 @@ static int walSeekWritePos(SWal* pWal, int64_t ver) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     return -1;
   }
-  return code;
+  return 0;
 }
 
 int walSetWrite(SWal* pWal) {
@@ -124,7 +124,7 @@ int walChangeWrite(SWal* pWal, int64_t ver) {
 }
 
 int walSeekWriteVer(SWal* pWal, int64_t ver) {
-  int code;
+  int64_t code;
   if (ver == pWal->vers.lastVer) {
     return 0;
   }

source/libs/wal/src/walWrite.c

@@ -30,8 +30,8 @@ int32_t walCommit(SWal *pWal, int64_t ver) {
 }
 
 int32_t walRollback(SWal *pWal, int64_t ver) {
-  int  code;
-  char fnameStr[WAL_FILE_LEN];
+  int64_t code;
+  char    fnameStr[WAL_FILE_LEN];
   if (ver > pWal->vers.lastVer || ver < pWal->vers.commitVer) {
     terrno = TSDB_CODE_WAL_INVALID_VER;
     return -1;

tests/test/c/tmqSim.c

@@ -333,8 +333,8 @@ void loop_consume(SThreadInfo* pInfo) {
 
       totalMsgs++;
 
-      if (totalMsgs >= pInfo->expectMsgCnt) {
-	  	taosFprintfFile(g_fp, "==== totalMsgs >= pInfo->expectMsgCnt, so break\n");
+      if (totalRows >= pInfo->expectMsgCnt) {
+	  	taosFprintfFile(g_fp, "==== totalRows >= pInfo->expectMsgCnt, so break\n");
         break;
       }
     } else {