Merge branch 'develop' into feature/TD-3188

Jenkinsfile

@@ -46,6 +46,7 @@ def pre_test(){
     git fetch origin +refs/pull/${CHANGE_ID}/merge
     git checkout -qf FETCH_HEAD
     git --no-pager diff --name-only FETCH_HEAD $(git merge-base FETCH_HEAD develop)|grep -v -E '.*md|//src//connector|Jenkinsfile' || exit 0
+    find ${WKC}/tests/pytest -name \'*\'.sql -exec rm -rf {} \\;
     cd ${WK}
     git reset --hard HEAD~10
     git checkout develop
@@ -115,7 +116,6 @@ pipeline {
               sh '''
               date
               cd ${WKC}/tests
-              find pytest -name '*'sql|xargs rm -rf
               ./test-all.sh p1
               date'''
             }
@@ -131,7 +131,6 @@ pipeline {
                 sh '''
                 date
                 cd ${WKC}/tests
-                find pytest -name '*'sql|xargs rm -rf
                 ./test-all.sh p2
                 date'''
             }

src/client/inc/tscUtil.h

@@ -123,6 +123,7 @@ int32_t tscGetDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, i
 bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo);
 bool tscIsTWAQuery(SQueryInfo* pQueryInfo);
 bool tscIsSecondStageQuery(SQueryInfo* pQueryInfo);
+bool tscGroupbyColumn(SQueryInfo* pQueryInfo);
 
 bool tscNonOrderedProjectionQueryOnSTable(SQueryInfo *pQueryInfo, int32_t tableIndex);
 bool tscOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex);
@@ -133,6 +134,7 @@ bool tscIsProjectionQuery(SQueryInfo* pQueryInfo);
 bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex);
 bool tscQueryTags(SQueryInfo* pQueryInfo);
 bool tscMultiRoundQuery(SQueryInfo* pQueryInfo, int32_t tableIndex);
+bool tscQueryBlockInfo(SQueryInfo* pQueryInfo);
 
 SSqlExpr* tscAddFuncInSelectClause(SQueryInfo* pQueryInfo, int32_t outputColIndex, int16_t functionId,
                                        SColumnIndex* pIndex, SSchema* pColSchema, int16_t colType);
@@ -152,7 +154,6 @@ SInternalField* tscFieldInfoInsert(SFieldInfo* pFieldInfo, int32_t index, TAOS_F
 SInternalField* tscFieldInfoGetInternalField(SFieldInfo* pFieldInfo, int32_t index);
 TAOS_FIELD* tscFieldInfoGetField(SFieldInfo* pFieldInfo, int32_t index);
 
-void tscFieldInfoUpdateOffset(SQueryInfo* pQueryInfo);
 void tscFieldInfoUpdateOffset(SQueryInfo* pQueryInfo);
 
 int16_t tscFieldInfoGetOffset(SQueryInfo* pQueryInfo, int32_t index);

src/client/inc/tsclient.h

@@ -198,9 +198,10 @@ typedef struct STableDataBlocks {
 typedef struct SQueryInfo {
   int16_t          command;       // the command may be different for each subclause, so keep it seperately.
   uint32_t         type;          // query/insert type
+  STimeWindow      window;        // the whole query time window
 
-  STimeWindow      window;        // query time window
-  SInterval        interval;
+  SInterval        interval;      // tumble time window
+  SSessionWindow   sessionWindow; // session time window
 
   SSqlGroupbyExpr  groupbyExpr;   // group by tags info
   SArray *         colList;       // SArray<SColumn*>
@@ -232,6 +233,7 @@ typedef struct SQueryInfo {
 typedef struct {
   int     command;
   uint8_t msgType;
+  char    reserve1[3];        // fix bus error on arm32
   bool    autoCreated;        // create table if it is not existed during retrieve table meta in mnode
 
   union {
@@ -244,8 +246,10 @@ typedef struct {
 
   char *       curSql;       // current sql, resume position of sql after parsing paused
   int8_t       parseFinished;
+  char    reserve2[3];        // fix bus error on arm32
 
   int16_t      numOfCols;
+  char    reserve3[2];        // fix bus error on arm32
   uint32_t     allocSize;
   char *       payload;
   int32_t      payloadLen;
@@ -255,7 +259,9 @@ typedef struct {
   int32_t      numOfParams;
 
   int8_t       dataSourceType;     // load data from file or not
+  char    reserve4[3];        // fix bus error on arm32
   int8_t       submitSchema;   // submit block is built with table schema
+  char    reserve5[3];        // fix bus error on arm32
   STagData     tagData;        // NOTE: pTagData->data is used as a variant length array
 
   SName      **pTableNameList; // all involved tableMeta list of current insert sql statement.
@@ -397,7 +403,6 @@ typedef struct SSqlStream {
 
 void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable);
 
-
 int  tscAcquireRpc(const char *key, const char *user, const char *secret,void **pRpcObj);
 void tscReleaseRpc(void *param);
 void tscInitMsgsFp();

src/client/src/tscLocalMerge.c

@@ -100,6 +100,10 @@ static void tscInitSqlContext(SSqlCmd *pCmd, SLocalMerger *pReducer, tOrderDescr
     } else if (functionId == TSDB_FUNC_APERCT) {
       pCtx->param[0].i64 = pExpr->param[0].i64;
       pCtx->param[0].nType  = pExpr->param[0].nType;
+    } else if (functionId == TSDB_FUNC_BLKINFO) {
+      pCtx->param[0].i64 = pExpr->param[0].i64;
+      pCtx->param[0].nType = pExpr->param[0].nType;
+      pCtx->numOfParams = 1;
     }
 
     pCtx->interBufBytes = pExpr->interBytes;
@@ -951,10 +955,10 @@ static void doFillResult(SSqlObj *pSql, SLocalMerger *pLocalMerge, bool doneOutp
   // todo extract function
   int64_t actualETime = (pQueryInfo->order.order == TSDB_ORDER_ASC)? pQueryInfo->window.ekey: pQueryInfo->window.skey;
 
-  tFilePage **pResPages = malloc(POINTER_BYTES * pQueryInfo->fieldsInfo.numOfOutput);
+  void** pResPages = malloc(POINTER_BYTES * pQueryInfo->fieldsInfo.numOfOutput);
   for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
     TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
-    pResPages[i] = calloc(1, sizeof(tFilePage) + pField->bytes * pLocalMerge->resColModel->capacity);
+    pResPages[i] = calloc(1, pField->bytes * pLocalMerge->resColModel->capacity);
   }
 
   while (1) {
@@ -966,7 +970,7 @@ static void doFillResult(SSqlObj *pSql, SLocalMerger *pLocalMerge, bool doneOutp
       if (pQueryInfo->limit.offset > 0) {
         for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
           TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
-          memmove(pResPages[i]->data, pResPages[i]->data + pField->bytes * pQueryInfo->limit.offset,
+          memmove(pResPages[i], ((char*)pResPages[i]) + pField->bytes * pQueryInfo->limit.offset,
                   (size_t)(newRows * pField->bytes));
         }
       }
@@ -1010,7 +1014,7 @@ static void doFillResult(SSqlObj *pSql, SLocalMerger *pLocalMerge, bool doneOutp
     int32_t offset = 0;
     for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
       TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
-      memcpy(pRes->data + offset * pRes->numOfRows, pResPages[i]->data, (size_t)(pField->bytes * pRes->numOfRows));
+      memcpy(pRes->data + offset * pRes->numOfRows, pResPages[i], (size_t)(pField->bytes * pRes->numOfRows));
       offset += pField->bytes;
     }
 

src/client/src/tscParseInsert.c

@@ -307,7 +307,8 @@ int32_t tsParseOneColumnData(SSchema *pSchema, SStrToken *pToken, char *payload,
           return tscInvalidSQLErrMsg(msg, "illegal float data", pToken->z);
         }
 
-        *((float *)payload) = (float)dv;
+//        *((float *)payload) = (float)dv;
+        SET_FLOAT_VAL(payload, dv);
       }
       break;
 

src/client/src/tscServer.c

@@ -497,8 +497,6 @@ int tscProcessSql(SSqlObj *pSql) {
       return pSql->res.code;
     }
   } else if (pCmd->command >= TSDB_SQL_LOCAL) {
-    //pSql->epSet = tscMgmtEpSet;
-//  } else {  // local handler
     return (*tscProcessMsgRsp[pCmd->command])(pSql);
   }
   
@@ -645,7 +643,6 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
     }
 
     pSql->epSet.inUse = rand()%pSql->epSet.numOfEps;
-
     pQueryMsg->head.vgId = htonl(vgId);
 
     STableIdInfo *pTableIdInfo = (STableIdInfo *)pMsg;
@@ -660,8 +657,6 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
     int32_t numOfVgroups = (int32_t)taosArrayGetSize(pTableMetaInfo->pVgroupTables);
     assert(index >= 0 && index < numOfVgroups);
 
-    tscDebug("%p query on stable, vgIndex:%d, numOfVgroups:%d", pSql, index, numOfVgroups);
-
     SVgroupTableInfo* pTableIdList = taosArrayGet(pTableMetaInfo->pVgroupTables, index);
 
     // set the vgroup info 
@@ -670,7 +665,10 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
     
     int32_t numOfTables = (int32_t)taosArrayGetSize(pTableIdList->itemList);
     pQueryMsg->numOfTables = htonl(numOfTables);  // set the number of tables
-  
+
+    tscDebug("%p query on stable, vgId:%d, numOfTables:%d, vgIndex:%d, numOfVgroups:%d", pSql,
+             pTableIdList->vgInfo.vgId, numOfTables, index, numOfVgroups);
+
     // serialize each table id info
     for(int32_t i = 0; i < numOfTables; ++i) {
       STableIdInfo* pItem = taosArrayGet(pTableIdList->itemList, i);
@@ -705,7 +703,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
   STableMeta * pTableMeta = pTableMetaInfo->pTableMeta;
 
   size_t numOfSrcCols = taosArrayGetSize(pQueryInfo->colList);
-  if (numOfSrcCols <= 0 && !tscQueryTags(pQueryInfo)) {
+  if (numOfSrcCols <= 0 && !tscQueryTags(pQueryInfo) && !tscQueryBlockInfo(pQueryInfo)) {
     tscError("%p illegal value of numOfCols in query msg: %" PRIu64 ", table cols:%d", pSql, (uint64_t)numOfSrcCols,
         tscGetNumOfColumns(pTableMeta));
 
@@ -756,6 +754,8 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
   pQueryMsg->vgroupLimit    = htobe64(pQueryInfo->vgroupLimit);
   pQueryMsg->sqlstrLen      = htonl(sqlLen);
   pQueryMsg->prevResultLen  = htonl(pQueryInfo->bufLen);
+  pQueryMsg->sw.gap         = htobe64(pQueryInfo->sessionWindow.gap);
+  pQueryMsg->sw.primaryColId = htonl(PRIMARYKEY_TIMESTAMP_COL_INDEX);
 
   size_t numOfOutput = tscSqlExprNumOfExprs(pQueryInfo);
   pQueryMsg->numOfOutput = htons((int16_t)numOfOutput);  // this is the stage one output column number
@@ -835,13 +835,31 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
     pSqlFuncExpr->colInfo.colIndex = htons(pExpr->colInfo.colIndex);
     pSqlFuncExpr->colInfo.flag     = htons(pExpr->colInfo.flag);
 
+    if (TSDB_COL_IS_UD_COL(pExpr->colInfo.flag)) {
+      pSqlFuncExpr->colType  = htons(pExpr->resType);
+      pSqlFuncExpr->colBytes = htons(pExpr->resBytes);
+    } else if (pExpr->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) {
+      SSchema *s = tGetTbnameColumnSchema();
+
+      pSqlFuncExpr->colType = htons(s->type);
+      pSqlFuncExpr->colBytes = htons(s->bytes);
+    } else if (pExpr->colInfo.colId == TSDB_BLOCK_DIST_COLUMN_INDEX) {
+      SSchema s = tGetBlockDistColumnSchema();
+
+      pSqlFuncExpr->colType = htons(s.type);
+      pSqlFuncExpr->colBytes = htons(s.bytes);
+    } else {
+      SSchema* s = tscGetColumnSchemaById(pTableMeta, pExpr->colInfo.colId);
+      pSqlFuncExpr->colType  = htons(s->type);
+      pSqlFuncExpr->colBytes = htons(s->bytes);
+    }
+
     pSqlFuncExpr->functionId  = htons(pExpr->functionId);
     pSqlFuncExpr->numOfParams = htons(pExpr->numOfParams);
     pSqlFuncExpr->resColId    = htons(pExpr->resColId);
     pMsg += sizeof(SSqlFuncMsg);
 
-    for (int32_t j = 0; j < pExpr->numOfParams; ++j) {
-      // todo add log
+    for (int32_t j = 0; j < pExpr->numOfParams; ++j) { // todo add log
       pSqlFuncExpr->arg[j].argType = htons((uint16_t)pExpr->param[j].nType);
       pSqlFuncExpr->arg[j].argBytes = htons(pExpr->param[j].nLen);
 
@@ -866,6 +884,8 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
     for (int32_t i = 0; i < output; ++i) {
       SInternalField* pField = tscFieldInfoGetInternalField(&pQueryInfo->fieldsInfo, i);
       SSqlExpr *pExpr = pField->pSqlExpr;
+
+      // this should be switched to projection query
       if (pExpr != NULL) {
         // the queried table has been removed and a new table with the same name has already been created already
         // return error msg
@@ -879,33 +899,31 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
           return TSDB_CODE_TSC_INVALID_SQL;
         }
 
-        pSqlFuncExpr1->colInfo.colId    = htons(pExpr->colInfo.colId);
-        pSqlFuncExpr1->colInfo.colIndex = htons(pExpr->colInfo.colIndex);
-        pSqlFuncExpr1->colInfo.flag     = htons(pExpr->colInfo.flag);
-
-        pSqlFuncExpr1->functionId  = htons(pExpr->functionId);
-        pSqlFuncExpr1->numOfParams = htons(pExpr->numOfParams);
-        pMsg += sizeof(SSqlFuncMsg);
-
-        for (int32_t j = 0; j < pExpr->numOfParams; ++j) {
-          // todo add log
-          pSqlFuncExpr1->arg[j].argType = htons((uint16_t)pExpr->param[j].nType);
-          pSqlFuncExpr1->arg[j].argBytes = htons(pExpr->param[j].nLen);
-
-          if (pExpr->param[j].nType == TSDB_DATA_TYPE_BINARY) {
-            memcpy(pMsg, pExpr->param[j].pz, pExpr->param[j].nLen);
-            pMsg += pExpr->param[j].nLen;
-          } else {
-            pSqlFuncExpr1->arg[j].argValue.i64 = htobe64(pExpr->param[j].i64);
+        pSqlFuncExpr1->numOfParams = 0;  // no params for projection query
+        pSqlFuncExpr1->functionId  = htons(TSDB_FUNC_PRJ);
+        pSqlFuncExpr1->colInfo.colId = htons(pExpr->resColId);
+        pSqlFuncExpr1->colInfo.flag = htons(TSDB_COL_NORMAL);
+
+        bool assign = false;
+        for (int32_t f = 0; f < tscSqlExprNumOfExprs(pQueryInfo); ++f) {
+          SSqlExpr *pe = tscSqlExprGet(pQueryInfo, f);
+          if (pe == pExpr) {
+            pSqlFuncExpr1->colInfo.colIndex = htons(f);
+            pSqlFuncExpr1->colType = htons(pe->resType);
+            pSqlFuncExpr1->colBytes = htons(pe->resBytes);
+            assign = true;
+            break;
           }
         }
 
+        assert(assign);
+        pMsg += sizeof(SSqlFuncMsg);
         pSqlFuncExpr1 = (SSqlFuncMsg *)pMsg;
       } else {
         assert(pField->pArithExprInfo != NULL);
         SExprInfo* pExprInfo = pField->pArithExprInfo;
 
-        pSqlFuncExpr1->colInfo.colId    = htons(pExprInfo->base.colInfo.colId);
+        pSqlFuncExpr1->colInfo.colId = htons(pExprInfo->base.colInfo.colId);
         pSqlFuncExpr1->functionId  = htons(pExprInfo->base.functionId);
         pSqlFuncExpr1->numOfParams = htons(pExprInfo->base.numOfParams);
         pMsg += sizeof(SSqlFuncMsg);

src/client/src/tscSub.c

@@ -503,9 +503,19 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
   SSqlCmd *pCmd = &pSql->cmd;
   STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
   SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
-  if (taosArrayGetSize(pSub->progress) > 0) { // fix crash in single tabel subscription
-    pQueryInfo->window.skey = ((SSubscriptionProgress*)taosArrayGet(pSub->progress, 0))->key;
-    tscDebug("subscribe:%s set subscribe skey:%"PRId64, pSub->topic, pQueryInfo->window.skey);
+  if (taosArrayGetSize(pSub->progress) > 0) { // fix crash in single table subscription
+
+    size_t size = taosArrayGetSize(pSub->progress);
+    TSKEY s = INT64_MAX;
+    for(int32_t i = 0; i < size; ++i) {
+      TSKEY k = ((SSubscriptionProgress*)taosArrayGet(pSub->progress, i))->key;
+      if (s > k) {
+        s = k;
+      }
+    }
+
+    pQueryInfo->window.skey = s;
+    tscDebug("subscribe:%s set next round subscribe skey:%"PRId64, pSub->topic, pQueryInfo->window.skey);
   }
 
   if (pSub->pTimer == NULL) {

src/client/src/tscSubquery.c

@@ -74,14 +74,14 @@ static bool allSubqueryDone(SSqlObj *pParentSql) {
   SSubqueryState *subState = &pParentSql->subState;
 
   //lock in caller
-  
+  tscDebug("%p total subqueries: %d", pParentSql, subState->numOfSub);
   for (int i = 0; i < subState->numOfSub; i++) {
     if (0 == subState->states[i]) {
-      tscDebug("%p subquery:%p,%d is NOT finished, total:%d", pParentSql, pParentSql->pSubs[i], i, subState->numOfSub);
+      tscDebug("%p subquery:%p, index: %d NOT finished, abort query completion check", pParentSql, pParentSql->pSubs[i], i);
       done = false;
       break;
     } else {
-      tscDebug("%p subquery:%p,%d is finished, total:%d", pParentSql, pParentSql->pSubs[i], i, subState->numOfSub);
+      tscDebug("%p subquery:%p, index: %d finished", pParentSql, pParentSql->pSubs[i], i);
     }
   }
 
@@ -453,7 +453,7 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) {
     pSubQueryInfo->tsBuf = NULL;
   
     // free result for async object will also free sqlObj
-    assert(tscSqlExprNumOfExprs(pSubQueryInfo) == 1); // ts_comp query only requires one resutl columns
+    assert(tscSqlExprNumOfExprs(pSubQueryInfo) == 1); // ts_comp query only requires one result columns
     taos_free_result(pPrevSub);
   
     SSqlObj *pNew = createSubqueryObj(pSql, (int16_t) i, tscJoinQueryCallback, pSupporter, TSDB_SQL_SELECT, NULL);
@@ -507,6 +507,7 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) {
     SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, 0);
     int16_t funcId = pExpr->functionId;
 
+    // add the invisible timestamp column
     if ((pExpr->colInfo.colId != PRIMARYKEY_TIMESTAMP_COL_INDEX) ||
         (funcId != TSDB_FUNC_TS && funcId != TSDB_FUNC_TS_DUMMY && funcId != TSDB_FUNC_PRJ)) {
 
@@ -847,6 +848,8 @@ static void tidTagRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
   SSqlRes* pRes = &pSql->res;
 
   SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
+
+  // todo, the type may not include TSDB_QUERY_TYPE_TAG_FILTER_QUERY
   assert(TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_TAG_FILTER_QUERY));
 
   if (pParentSql->res.code != TSDB_CODE_SUCCESS) {
@@ -1059,7 +1062,6 @@ static void tsCompRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
       tscError("%p invalid ts comp file from vnode, abort subquery, file size:%d", pSql, numOfRows);
 
       pParentSql->res.code = TAOS_SYSTEM_ERROR(errno);
-
       if (quitAllSubquery(pSql, pParentSql, pSupporter)){
         return;
       }
@@ -1880,6 +1882,13 @@ void doAppendData(SInterResult* pInterResult, TAOS_ROW row, int32_t numOfCols, S
       }
     }
 
+    if (p && taosArrayGetSize(p) > 0) {
+      SResPair *l = taosArrayGetLast(p);
+      if (l->key == key && key == INT64_MIN) {
+        continue;
+      }
+    }
+
     //append a new column
     if (p == NULL) {
       SStddevInterResult t = {.colId = id, .pResult = taosArrayInit(10, sizeof(SResPair)),};
@@ -1941,7 +1950,11 @@ void tscFirstRoundRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) {
 
           // tag or group by column
           if (TSDB_COL_IS_TAG(pExpr->colInfo.flag) || pExpr->functionId == TSDB_FUNC_PRJ) {
-            memcpy(p + offset, row[i], length[i]);
+            if (row[i] == NULL) {
+              setNull(p + offset, pExpr->resType, pExpr->resBytes);
+            } else {
+              memcpy(p + offset, row[i], length[i]);
+            }
             offset += pExpr->resBytes;
           }
         }
@@ -2639,12 +2652,17 @@ static SSqlObj *tscCreateSTableSubquery(SSqlObj *pSql, SRetrieveSupport *trsuppo
     SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pNew->cmd, 0);
 
     pQueryInfo->type |= TSDB_QUERY_TYPE_STABLE_SUBQUERY;
+
+    // clear the limit/offset info, since it should not be sent to vnode to be executed.
+    pQueryInfo->limit.limit = -1;
+    pQueryInfo->limit.offset = 0;
+
     assert(pQueryInfo->numOfTables == 1 && pNew->cmd.numOfClause == 1 && trsupport->subqueryIndex < pSql->subState.numOfSub);
     
     // launch subquery for each vnode, so the subquery index equals to the vgroupIndex.
     STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, table_index);
     pTableMetaInfo->vgroupIndex = trsupport->subqueryIndex;
-    
+
     pSql->pSubs[trsupport->subqueryIndex] = pNew;
   }
   
@@ -3098,30 +3116,6 @@ void tscBuildResFromSubqueries(SSqlObj *pSql) {
   }
 }
 
-static UNUSED_FUNC void transferNcharData(SSqlObj *pSql, int32_t columnIndex, TAOS_FIELD *pField) {
-  SSqlRes *pRes = &pSql->res;
-  
-  if (pRes->tsrow[columnIndex] != NULL && pField->type == TSDB_DATA_TYPE_NCHAR) {
-    // convert unicode to native code in a temporary buffer extra one byte for terminated symbol
-    if (pRes->buffer[columnIndex] == NULL) {
-      pRes->buffer[columnIndex] = malloc(pField->bytes + TSDB_NCHAR_SIZE);
-    }
-    
-    /* string terminated char for binary data*/
-    memset(pRes->buffer[columnIndex], 0, pField->bytes + TSDB_NCHAR_SIZE);
-    
-    int32_t length = taosUcs4ToMbs(pRes->tsrow[columnIndex], pRes->length[columnIndex], pRes->buffer[columnIndex]);
-    if ( length >= 0 ) {
-      pRes->tsrow[columnIndex] = (unsigned char*)pRes->buffer[columnIndex];
-      pRes->length[columnIndex] = length;
-    } else {
-      tscError("%p charset:%s to %s. val:%s convert failed.", pSql, DEFAULT_UNICODE_ENCODEC, tsCharset, (char*)pRes->tsrow[columnIndex]);
-      pRes->tsrow[columnIndex] = NULL;
-      pRes->length[columnIndex] = 0;
-    }
-  }
-}
-
 char *getArithmeticInputSrc(void *param, const char *name, int32_t colId) {
   SArithmeticSupport *pSupport = (SArithmeticSupport *) param;
 

src/client/src/tscUtil.c

@@ -97,6 +97,22 @@ bool tscQueryTags(SQueryInfo* pQueryInfo) {
   return true;
 }
 
+bool tscQueryBlockInfo(SQueryInfo* pQueryInfo) {
+  int32_t numOfCols = (int32_t) tscSqlExprNumOfExprs(pQueryInfo);
+
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i);
+    int32_t functId = pExpr->functionId;
+
+    // "select count(tbname)" query
+    if (functId == TSDB_FUNC_BLKINFO) {
+      return true;
+    }
+  }
+
+  return false;
+}
+
 bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) {
   if (pQueryInfo == NULL) {
     return false;
@@ -223,6 +239,21 @@ bool tscIsSecondStageQuery(SQueryInfo* pQueryInfo) {
   return false;
 }
 
+bool tscGroupbyColumn(SQueryInfo* pQueryInfo) {
+  STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
+  int32_t         numOfCols = tscGetNumOfColumns(pTableMetaInfo->pTableMeta);
+
+  SSqlGroupbyExpr* pGroupbyExpr = &pQueryInfo->groupbyExpr;
+  for (int32_t k = 0; k < pGroupbyExpr->numOfGroupCols; ++k) {
+    SColIndex* pIndex = taosArrayGet(pGroupbyExpr->columnInfo, k);
+    if (!TSDB_COL_IS_TAG(pIndex->flag) && pIndex->colIndex < numOfCols) {  // group by normal columns
+      return true;
+    }
+  }
+
+  return false;
+}
+
 bool tscIsTWAQuery(SQueryInfo* pQueryInfo) {
   size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
   for (int32_t i = 0; i < numOfExprs; ++i) {
@@ -1722,10 +1753,15 @@ void tscInitQueryInfo(SQueryInfo* pQueryInfo) {
   pQueryInfo->fieldsInfo.internalField = taosArrayInit(4, sizeof(SInternalField));
   
   assert(pQueryInfo->exprList == NULL);
-  pQueryInfo->exprList   = taosArrayInit(4, POINTER_BYTES);
-  pQueryInfo->colList    = taosArrayInit(4, POINTER_BYTES);
-  pQueryInfo->udColumnId = TSDB_UD_COLUMN_INDEX;
-  pQueryInfo->resColumnId= -1000;
+  pQueryInfo->exprList    = taosArrayInit(4, POINTER_BYTES);
+  pQueryInfo->colList     = taosArrayInit(4, POINTER_BYTES);
+  pQueryInfo->udColumnId  = TSDB_UD_COLUMN_INDEX;
+  pQueryInfo->resColumnId = -1000;
+  pQueryInfo->limit.limit = -1;
+  pQueryInfo->limit.offset = 0;
+
+  pQueryInfo->slimit.limit = -1;
+  pQueryInfo->slimit.offset = 0;
 }
 
 int32_t tscAddSubqueryInfo(SSqlCmd* pCmd) {

src/common/inc/tdataformat.h

@@ -283,12 +283,37 @@ typedef struct {
 #define keyCol(pCols) (&((pCols)->cols[0]))  // Key column
 #define dataColsTKeyAt(pCols, idx) ((TKEY *)(keyCol(pCols)->pData))[(idx)]
 #define dataColsKeyAt(pCols, idx) tdGetKey(dataColsTKeyAt(pCols, idx))
-#define dataColsTKeyFirst(pCols) (((pCols)->numOfRows == 0) ? TKEY_INVALID : dataColsTKeyAt(pCols, 0))
-#define dataColsKeyFirst(pCols) (((pCols)->numOfRows == 0) ? TSDB_DATA_TIMESTAMP_NULL : dataColsKeyAt(pCols, 0))
-#define dataColsTKeyLast(pCols) \
-  (((pCols)->numOfRows == 0) ? TKEY_INVALID : dataColsTKeyAt(pCols, (pCols)->numOfRows - 1))
-#define dataColsKeyLast(pCols) \
-  (((pCols)->numOfRows == 0) ? TSDB_DATA_TIMESTAMP_NULL : dataColsKeyAt(pCols, (pCols)->numOfRows - 1))
+static FORCE_INLINE TKEY dataColsTKeyFirst(SDataCols *pCols) {
+  if (pCols->numOfRows) {
+    return dataColsTKeyAt(pCols, 0);
+  } else {
+    return TKEY_INVALID;
+  }
+}
+
+static FORCE_INLINE TSKEY dataColsKeyFirst(SDataCols *pCols) {
+  if (pCols->numOfRows) {
+    return dataColsKeyAt(pCols, 0);
+  } else {
+    return TSDB_DATA_TIMESTAMP_NULL;
+  }
+}
+
+static FORCE_INLINE TKEY dataColsTKeyLast(SDataCols *pCols) {
+  if (pCols->numOfRows) {
+    return dataColsTKeyAt(pCols, pCols->numOfRows - 1);
+  } else {
+    return TKEY_INVALID;
+  }
+}
+
+static FORCE_INLINE TSKEY dataColsKeyLast(SDataCols *pCols) {
+  if (pCols->numOfRows) {
+    return dataColsKeyAt(pCols, pCols->numOfRows - 1);
+  } else {
+    return TSDB_DATA_TIMESTAMP_NULL;
+  }
+}
 
 SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows);
 void       tdResetDataCols(SDataCols *pCols);

src/common/inc/tname.h

@@ -33,7 +33,7 @@ typedef struct SDataStatis {
 
 typedef struct SColumnInfoData {
   SColumnInfo info;
-  void* pData;    // the corresponding block data in memory
+  char* pData;    // the corresponding block data in memory
 } SColumnInfoData;
 
 typedef struct SResPair {

src/connector/python/linux/python2/setup.py

@@ -5,7 +5,7 @@ with open("README.md", "r") as fh:
 
 setuptools.setup(
     name="taos",
-    version="2.0.6",
+    version="2.0.7",
     author="Taosdata Inc.",
     author_email="support@taosdata.com",
     description="TDengine python client package",

src/connector/python/linux/python2/taos/cinterface.py

@@ -22,10 +22,10 @@ def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False):
 
     if num_of_rows > 0:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
     else:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
 
 
 def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False):
@@ -145,10 +145,10 @@ def _crow_bigint_to_python(data, num_of_rows, nbytes=None, micro=False):
     """
     if num_of_rows > 0:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
     else:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
 
 
 def _crow_bigint_unsigned_to_python(
@@ -162,13 +162,13 @@ def _crow_bigint_unsigned_to_python(
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
     else:
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
 
 
@@ -600,7 +600,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_INT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BIGINT):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_FLOAT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_float))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_DOUBLE):
@@ -608,7 +608,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BINARY):
     #         return (ctypes.cast(data,  ctypes.POINTER(ctypes.c_char))[0:byte]).rstrip('\x00')
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TIMESTAMP):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_NCHAR):
     #         return (ctypes.cast(data,  ctypes.c_char_p).value).rstrip('\x00')
 

src/connector/python/linux/python3/setup.py

@@ -5,7 +5,7 @@ with open("README.md", "r") as fh:
 
 setuptools.setup(
     name="taos",
-    version="2.0.5",
+    version="2.0.7",
     author="Taosdata Inc.",
     author_email="support@taosdata.com",
     description="TDengine python client package",

src/connector/python/linux/python3/taos/cinterface.py

@@ -22,10 +22,10 @@ def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False):
 
     if num_of_rows > 0:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
     else:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
 
 
 def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False):
@@ -145,10 +145,10 @@ def _crow_bigint_to_python(data, num_of_rows, nbytes=None, micro=False):
     """
     if num_of_rows > 0:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
     else:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
 
 
 def _crow_bigint_unsigned_to_python(
@@ -162,13 +162,13 @@ def _crow_bigint_unsigned_to_python(
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
     else:
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
 
 
@@ -600,7 +600,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_INT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BIGINT):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_FLOAT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_float))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_DOUBLE):
@@ -608,7 +608,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BINARY):
     #         return (ctypes.cast(data,  ctypes.POINTER(ctypes.c_char))[0:byte]).rstrip('\x00')
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TIMESTAMP):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_NCHAR):
     #         return (ctypes.cast(data,  ctypes.c_char_p).value).rstrip('\x00')
 

src/connector/python/linux/python3/taos/cursor.py

 from .cinterface import CTaosInterface
 from .error import *
 from .constants import FieldType
-import threading
 
 # querySeqNum = 0
 
@@ -38,7 +37,6 @@ class TDengineCursor(object):
         self._block_iter = 0
         self._affected_rows = 0
         self._logfile = ""
-        self._threadId = threading.get_ident()
 
         if connection is not None:
             self._connection = connection
@@ -105,12 +103,6 @@ class TDengineCursor(object):
     def execute(self, operation, params=None):
         """Prepare and execute a database operation (query or command).
         """
-        # if threading.get_ident() != self._threadId:
-        #     info ="Cursor execute:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident())
-        #     raise OperationalError(info)
-        # print(info)
-        # return None
-
         if not operation:
             return None
 
@@ -280,12 +272,6 @@ class TDengineCursor(object):
     def _handle_result(self):
         """Handle the return result from query.
         """
-        # if threading.get_ident() != self._threadId:
-        #     info = "Cursor handleresult:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident())
-        #     raise OperationalError(info)
-        # print(info)
-        # return None
-
         self._description = []
         for ele in self._fields:
             self._description.append(

src/connector/python/osx/python3/setup.py

@@ -5,7 +5,7 @@ with open("README.md", "r") as fh:
 
 setuptools.setup(
     name="taos",
-    version="2.0.5",
+    version="2.0.7",
     author="Taosdata Inc.",
     author_email="support@taosdata.com",
     description="TDengine python client package",

src/connector/python/osx/python3/taos/cinterface.py

@@ -22,10 +22,10 @@ def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False):
 
     if num_of_rows > 0:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
     else:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
 
 
 def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False):
@@ -145,10 +145,10 @@ def _crow_bigint_to_python(data, num_of_rows, nbytes=None, micro=False):
     """
     if num_of_rows > 0:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
     else:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
 
 
 def _crow_bigint_unsigned_to_python(
@@ -162,13 +162,13 @@ def _crow_bigint_unsigned_to_python(
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
     else:
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
 
 
@@ -600,7 +600,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_INT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BIGINT):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_FLOAT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_float))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_DOUBLE):
@@ -608,7 +608,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BINARY):
     #         return (ctypes.cast(data,  ctypes.POINTER(ctypes.c_char))[0:byte]).rstrip('\x00')
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TIMESTAMP):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_NCHAR):
     #         return (ctypes.cast(data,  ctypes.c_char_p).value).rstrip('\x00')
 

src/connector/python/osx/python3/taos/cursor.py

 from .cinterface import CTaosInterface
 from .error import *
 from .constants import FieldType
-import threading
 
 # querySeqNum = 0
 
@@ -38,7 +37,6 @@ class TDengineCursor(object):
         self._block_iter = 0
         self._affected_rows = 0
         self._logfile = ""
-        self._threadId = threading.get_ident()
 
         if connection is not None:
             self._connection = connection
@@ -105,12 +103,6 @@ class TDengineCursor(object):
     def execute(self, operation, params=None):
         """Prepare and execute a database operation (query or command).
         """
-        # if threading.get_ident() != self._threadId:
-        #     info ="Cursor execute:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident())
-        #     raise OperationalError(info)
-        # print(info)
-        # return None
-
         if not operation:
             return None
 
@@ -280,12 +272,6 @@ class TDengineCursor(object):
     def _handle_result(self):
         """Handle the return result from query.
         """
-        # if threading.get_ident() != self._threadId:
-        #     info = "Cursor handleresult:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident())
-        #     raise OperationalError(info)
-        # print(info)
-        # return None
-
         self._description = []
         for ele in self._fields:
             self._description.append(

src/connector/python/windows/python2/setup.py

@@ -5,7 +5,7 @@ with open("README.md", "r") as fh:
 
 setuptools.setup(
     name="taos",
-    version="2.0.4",
+    version="2.0.7",
     author="Taosdata Inc.",
     author_email="support@taosdata.com",
     description="TDengine python client package",

src/connector/python/windows/python2/taos/cinterface.py

@@ -22,10 +22,10 @@ def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False):
 
     if num_of_rows > 0:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
     else:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
 
 
 def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False):
@@ -145,10 +145,10 @@ def _crow_bigint_to_python(data, num_of_rows, nbytes=None, micro=False):
     """
     if num_of_rows > 0:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
     else:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
 
 
 def _crow_bigint_unsigned_to_python(
@@ -162,13 +162,13 @@ def _crow_bigint_unsigned_to_python(
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
     else:
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
 
 
@@ -600,7 +600,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_INT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BIGINT):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_FLOAT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_float))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_DOUBLE):
@@ -608,7 +608,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BINARY):
     #         return (ctypes.cast(data,  ctypes.POINTER(ctypes.c_char))[0:byte]).rstrip('\x00')
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TIMESTAMP):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_NCHAR):
     #         return (ctypes.cast(data,  ctypes.c_char_p).value).rstrip('\x00')
 

src/connector/python/windows/python2/taos/cursor.py

 from .cinterface import CTaosInterface
 from .error import *
 from .constants import FieldType
-import threading
 
 # querySeqNum = 0
 
@@ -38,7 +37,6 @@ class TDengineCursor(object):
         self._block_iter = 0
         self._affected_rows = 0
         self._logfile = ""
-        self._threadId = threading.get_ident()
 
         if connection is not None:
             self._connection = connection

src/connector/python/windows/python3/setup.py

@@ -5,7 +5,7 @@ with open("README.md", "r") as fh:
 
 setuptools.setup(
     name="taos",
-    version="2.0.4",
+    version="2.0.7",
     author="Taosdata Inc.",
     author_email="support@taosdata.com",
     description="TDengine python client package",

src/connector/python/windows/python3/taos/cinterface.py

@@ -22,10 +22,10 @@ def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False):
 
     if num_of_rows > 0:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
     else:
         return list(map(_timestamp_converter, ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]))
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]))
 
 
 def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False):
@@ -145,10 +145,10 @@ def _crow_bigint_to_python(data, num_of_rows, nbytes=None, micro=False):
     """
     if num_of_rows > 0:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
     else:
         return [None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(
-            data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)]]
+            data, ctypes.POINTER(ctypes.c_int64))[:abs(num_of_rows)]]
 
 
 def _crow_bigint_unsigned_to_python(
@@ -162,13 +162,13 @@ def _crow_bigint_unsigned_to_python(
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
     else:
         return [
             None if ele == FieldType.C_BIGINT_UNSIGNED_NULL else ele for ele in ctypes.cast(
                 data, ctypes.POINTER(
-                    ctypes.c_ulong))[
+                    ctypes.c_uint64))[
                 :abs(num_of_rows)]]
 
 
@@ -600,7 +600,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_INT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BIGINT):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_FLOAT):
     #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_float))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_DOUBLE):
@@ -608,7 +608,7 @@ class CTaosInterface(object):
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BINARY):
     #         return (ctypes.cast(data,  ctypes.POINTER(ctypes.c_char))[0:byte]).rstrip('\x00')
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TIMESTAMP):
-    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_long))[0]
+    #         return ctypes.cast(data,  ctypes.POINTER(ctypes.c_int64))[0]
     #     elif (dtype == CTaosInterface.TSDB_DATA_TYPE_NCHAR):
     #         return (ctypes.cast(data,  ctypes.c_char_p).value).rstrip('\x00')
 

src/connector/python/windows/python3/taos/cursor.py

 from .cinterface import CTaosInterface
 from .error import *
 from .constants import FieldType
-import threading
 
 # querySeqNum = 0
 
@@ -38,7 +37,6 @@ class TDengineCursor(object):
         self._block_iter = 0
         self._affected_rows = 0
         self._logfile = ""
-        self._threadId = threading.get_ident()
 
         if connection is not None:
             self._connection = connection

src/inc/taoserror.h

@@ -244,6 +244,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TDB_IVD_CREATE_TABLE_INFO     TAOS_DEF_ERROR_CODE(0, 0x0612)  //"Invalid information to create table")
 #define TSDB_CODE_TDB_NO_AVAIL_DISK             TAOS_DEF_ERROR_CODE(0, 0x0613)  //"No available disk")
 #define TSDB_CODE_TDB_MESSED_MSG                TAOS_DEF_ERROR_CODE(0, 0x0614)  //"TSDB messed message")
+#define TSDB_CODE_TDB_IVLD_TAG_VAL              TAOS_DEF_ERROR_CODE(0, 0x0615)  //"TSDB invalid tag value")
 
 // query
 #define TSDB_CODE_QRY_INVALID_QHANDLE           TAOS_DEF_ERROR_CODE(0, 0x0700)  //"Invalid handle")
@@ -258,7 +259,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_QRY_IN_EXEC                   TAOS_DEF_ERROR_CODE(0, 0x0709)  //"Multiple retrieval of this query")
 #define TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW       TAOS_DEF_ERROR_CODE(0, 0x070A)  //"Too many time window in query")
 #define TSDB_CODE_QRY_NOT_ENOUGH_BUFFER         TAOS_DEF_ERROR_CODE(0, 0x070B)  //"Query buffer limit has reached")
-#define TSDB_CODE_QRY_INCONSISTAN               TAOS_DEF_ERROR_CODE(0, 0x070C)  //"File inconsistance in replica")
+#define TSDB_CODE_QRY_INCONSISTAN               TAOS_DEF_ERROR_CODE(0, 0x070C)  //"File inconsistency in replica")
 
 
 // grant

src/inc/taosmsg.h

@@ -397,7 +397,7 @@ typedef struct SColIndex {
   int16_t  colId;      // column id
   int16_t  colIndex;   // column index in colList if it is a normal column or index in tagColList if a tag
   uint16_t flag;       // denote if it is a tag or a normal column
-  char     name[TSDB_COL_NAME_LEN];
+  char     name[TSDB_COL_NAME_LEN];  // TODO remove it
 } SColIndex;
 
 /* sql function msg, to describe the message to vnode about sql function
@@ -405,7 +405,10 @@ typedef struct SColIndex {
 typedef struct SSqlFuncMsg {
   int16_t functionId;
   int16_t numOfParams;
+
   int16_t resColId;      // result column id, id of the current output column
+  int16_t colType;
+  int16_t colBytes;
 
   SColIndex colInfo;
   struct ArgElem {
@@ -485,12 +488,13 @@ typedef struct {
   int16_t     orderColId;
   int16_t     numOfCols;        // the number of columns will be load from vnode
   SInterval   interval;
+  SSessionWindow sw;            // session window
   uint16_t    tagCondLen;       // tag length in current query
   uint32_t    tbnameCondLen;    // table name filter condition string length
   int16_t     numOfGroupCols;   // num of group by columns
   int16_t     orderByIdx;
   int16_t     orderType;        // used in group by xx order by xxx
-  int64_t     vgroupLimit;       // limit the number of rows for each table, used in order by + limit in stable projection query.
+  int64_t     vgroupLimit;      // limit the number of rows for each table, used in order by + limit in stable projection query.
   int16_t     prjOrder;         // global order in super table projection query.
   int64_t     limit;
   int64_t     offset;
@@ -640,6 +644,7 @@ typedef struct {
   int32_t  maxtablesPerVnode;
   int32_t  maxVgroupsPerDb;
   char     arbitrator[TSDB_EP_LEN];   // tsArbitrator
+  char     reserve[2];                // to solve arm32 bus error
   char     timezone[64];              // tsTimezone
   int64_t  checkTime;                 // 1970-01-01 00:00:00.000
   char     locale[TSDB_LOCALE_LEN];   // tsLocale

src/inc/tsdb.h

@@ -158,13 +158,18 @@ int32_t tsdbInsertData(STsdbRepo *repo, SSubmitMsg *pMsg, SShellSubmitRspMsg *pR
 
 typedef void *TsdbQueryHandleT;  // Use void to hide implementation details
 
-// query condition to build vnode iterator
+#define BLOCK_LOAD_OFFSET_SEQ_ORDER   1
+#define BLOCK_LOAD_TABLE_SEQ_ORDER    2
+#define BLOCK_LOAD_TABLE_RR_ORDER     3
+
+// query condition to build multi-table data block iterator
 typedef struct STsdbQueryCond {
   STimeWindow  twindow;
   int32_t      order;             // desc|asc order to iterate the data block
   int32_t      numOfCols;
   SColumnInfo *colList;
   bool         loadExternalRows;  // load external rows or not
+  int32_t      type;              // data block load type:
 } STsdbQueryCond;
 
 typedef struct SMemRef {
@@ -181,17 +186,31 @@ typedef struct SDataBlockInfo {
   int32_t     tid;
 } SDataBlockInfo;
 
+typedef struct SFileBlockInfo {
+  int32_t numOfRows;
+} SFileBlockInfo;
+
 typedef struct {
   void *pTable;
   TSKEY lastKey;
 } STableKeyInfo;
 
 typedef struct {
-  size_t    numOfTables;
+  uint32_t  numOfTables;
   SArray *  pGroupList;
   SHashObj *map;  // speedup acquire the tableQueryInfo by table uid
 } STableGroupInfo;
 
+typedef struct {
+  uint16_t  rowSize;
+  uint16_t  numOfFiles;
+  uint32_t  numOfTables;
+  uint64_t  totalSize;
+  int32_t   firstSeekTimeUs;
+  uint32_t  numOfRowsInMemTable;
+  SArray   *dataBlockInfos;
+} STableBlockDist;
+
 /**
  * Get the data block iterator, starting from position according to the query condition
  *
@@ -252,16 +271,7 @@ int64_t tsdbGetNumOfRowsInMemTable(TsdbQueryHandleT* pHandle);
  * @param pQueryHandle
  * @return
  */
-bool tsdbNextDataBlock(TsdbQueryHandleT *pQueryHandle);
-/**
- * move to next block if exists but not merge data in memtable 
- *
- * @param pQueryHandle
- * @return
- */
-bool tsdbNextDataBlockWithoutMerge(TsdbQueryHandleT *pQueryHandle);
-
-SArray* tsdbGetExternalRow(TsdbQueryHandleT *pHandle, SMemRef* pMemRef, int16_t type);
+bool tsdbNextDataBlock(TsdbQueryHandleT pQueryHandle);
 
 /**
  * Get current data block information
@@ -306,7 +316,7 @@ int32_t tsdbQuerySTableByTagCond(STsdbRepo *tsdb, uint64_t uid, TSKEY key, const
                                  SColIndex *pColIndex, int32_t numOfCols);
 
 /**
- * destory the created table group list, which is generated by tag query
+ * destroy the created table group list, which is generated by tag query
  * @param pGroupList
  */
 void tsdbDestroyTableGroup(STableGroupInfo *pGroupList);
@@ -336,6 +346,12 @@ int32_t tsdbGetTableGroupFromIdList(STsdbRepo *tsdb, SArray *pTableIdList, STabl
  */
 void tsdbCleanupQueryHandle(TsdbQueryHandleT queryHandle);
 
+void tsdbResetQueryHandle(TsdbQueryHandleT queryHandle, STsdbQueryCond *pCond);
+
+void tsdbResetQueryHandleForNewTable(TsdbQueryHandleT queryHandle, STsdbQueryCond *pCond, STableGroupInfo* groupList);
+
+int32_t tsdbGetFileBlocksDistInfo(TsdbQueryHandleT* queryHandle, STableBlockDist* pTableBlockInfo);
+
 /**
  * get the statistics of repo usage
  * @param repo. point to the tsdbrepo

src/inc/ttokendef.h

@@ -138,100 +138,76 @@
 #define TK_FROM                           119
 #define TK_VARIABLE                       120
 #define TK_INTERVAL                       121
-#define TK_FILL                           122
-#define TK_SLIDING                        123
-#define TK_ORDER                          124
-#define TK_BY                             125
-#define TK_ASC                            126
-#define TK_DESC                           127
-#define TK_GROUP                          128
-#define TK_HAVING                         129
-#define TK_LIMIT                          130
-#define TK_OFFSET                         131
-#define TK_SLIMIT                         132
-#define TK_SOFFSET                        133
-#define TK_WHERE                          134
-#define TK_NOW                            135
-#define TK_RESET                          136
-#define TK_QUERY                          137
-#define TK_ADD                            138
-#define TK_COLUMN                         139
-#define TK_TAG                            140
-#define TK_CHANGE                         141
-#define TK_SET                            142
-#define TK_KILL                           143
-#define TK_CONNECTION                     144
-#define TK_STREAM                         145
-#define TK_COLON                          146
-#define TK_ABORT                          147
-#define TK_AFTER                          148
-#define TK_ATTACH                         149
-#define TK_BEFORE                         150
-#define TK_BEGIN                          151
-#define TK_CASCADE                        152
-#define TK_CLUSTER                        153
-#define TK_CONFLICT                       154
-#define TK_COPY                           155
-#define TK_DEFERRED                       156
-#define TK_DELIMITERS                     157
-#define TK_DETACH                         158
-#define TK_EACH                           159
-#define TK_END                            160
-#define TK_EXPLAIN                        161
-#define TK_FAIL                           162
-#define TK_FOR                            163
-#define TK_IGNORE                         164
-#define TK_IMMEDIATE                      165
-#define TK_INITIALLY                      166
-#define TK_INSTEAD                        167
-#define TK_MATCH                          168
-#define TK_KEY                            169
-#define TK_OF                             170
-#define TK_RAISE                          171
-#define TK_REPLACE                        172
-#define TK_RESTRICT                       173
-#define TK_ROW                            174
-#define TK_STATEMENT                      175
-#define TK_TRIGGER                        176
-#define TK_VIEW                           177
-#define TK_COUNT                          178
-#define TK_SUM                            179
-#define TK_AVG                            180
-#define TK_MIN                            181
-#define TK_MAX                            182
-#define TK_FIRST                          183
-#define TK_LAST                           184
-#define TK_TOP                            185
-#define TK_BOTTOM                         186
-#define TK_STDDEV                         187
-#define TK_PERCENTILE                     188
-#define TK_APERCENTILE                    189
-#define TK_LEASTSQUARES                   190
-#define TK_HISTOGRAM                      191
-#define TK_DIFF                           192
-#define TK_SPREAD                         193
-#define TK_TWA                            194
-#define TK_INTERP                         195
-#define TK_LAST_ROW                       196
-#define TK_RATE                           197
-#define TK_IRATE                          198
-#define TK_SUM_RATE                       199
-#define TK_SUM_IRATE                      200
-#define TK_AVG_RATE                       201
-#define TK_AVG_IRATE                      202
-#define TK_TBID                           203
-#define TK_SEMI                           204
-#define TK_NONE                           205
-#define TK_PREV                           206
-#define TK_LINEAR                         207
-#define TK_IMPORT                         208
-#define TK_METRIC                         209
-#define TK_TBNAME                         210
-#define TK_JOIN                           211
-#define TK_METRICS                        212
-#define TK_INSERT                         213
-#define TK_INTO                           214
-#define TK_VALUES                         215
+#define TK_SESSION                        122
+#define TK_FILL                           123
+#define TK_SLIDING                        124
+#define TK_ORDER                          125
+#define TK_BY                             126
+#define TK_ASC                            127
+#define TK_DESC                           128
+#define TK_GROUP                          129
+#define TK_HAVING                         130
+#define TK_LIMIT                          131
+#define TK_OFFSET                         132
+#define TK_SLIMIT                         133
+#define TK_SOFFSET                        134
+#define TK_WHERE                          135
+#define TK_NOW                            136
+#define TK_RESET                          137
+#define TK_QUERY                          138
+#define TK_ADD                            139
+#define TK_COLUMN                         140
+#define TK_TAG                            141
+#define TK_CHANGE                         142
+#define TK_SET                            143
+#define TK_KILL                           144
+#define TK_CONNECTION                     145
+#define TK_STREAM                         146
+#define TK_COLON                          147
+#define TK_ABORT                          148
+#define TK_AFTER                          149
+#define TK_ATTACH                         150
+#define TK_BEFORE                         151
+#define TK_BEGIN                          152
+#define TK_CASCADE                        153
+#define TK_CLUSTER                        154
+#define TK_CONFLICT                       155
+#define TK_COPY                           156
+#define TK_DEFERRED                       157
+#define TK_DELIMITERS                     158
+#define TK_DETACH                         159
+#define TK_EACH                           160
+#define TK_END                            161
+#define TK_EXPLAIN                        162
+#define TK_FAIL                           163
+#define TK_FOR                            164
+#define TK_IGNORE                         165
+#define TK_IMMEDIATE                      166
+#define TK_INITIALLY                      167
+#define TK_INSTEAD                        168
+#define TK_MATCH                          169
+#define TK_KEY                            170
+#define TK_OF                             171
+#define TK_RAISE                          172
+#define TK_REPLACE                        173
+#define TK_RESTRICT                       174
+#define TK_ROW                            175
+#define TK_STATEMENT                      176
+#define TK_TRIGGER                        177
+#define TK_VIEW                           178
+#define TK_SEMI                           179
+#define TK_NONE                           180
+#define TK_PREV                           181
+#define TK_LINEAR                         182
+#define TK_IMPORT                         183
+#define TK_METRIC                         184
+#define TK_TBNAME                         185
+#define TK_JOIN                           186
+#define TK_METRICS                        187
+#define TK_INSERT                         188
+#define TK_INTO                           189
+#define TK_VALUES                         190
+
 
 
 

src/inc/ttype.h

@@ -171,10 +171,10 @@ extern tDataTypeDescriptor tDataTypes[15];
 
 bool isValidDataType(int32_t type);
 
-void setVardataNull(char* val, int32_t type);
-void setNull(char *val, int32_t type, int32_t bytes);
-void setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems);
-void* getNullValue(int32_t type);
+void  setVardataNull(char* val, int32_t type);
+void  setNull(char *val, int32_t type, int32_t bytes);
+void  setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems);
+void *getNullValue(int32_t type);
 
 void assignVal(char *val, const char *src, int32_t len, int32_t type);
 void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf);

src/kit/taosdemo/insert.json

@@ -11,6 +11,7 @@
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,
     "num_of_records_per_req": 100,
+    "max_sql_len": 1024000,
 	"databases": [{
 		"dbinfo": {
 			"name": "db",
@@ -38,7 +39,9 @@
 			"auto_create_table": "no",
 			"data_source": "rand",
 			"insert_mode": "taosc",
-			"insert_rows": 100000,
+            "childtable_limit": 33,
+            "childtable_offset": 33,
+			"insert_rows": 1000,
 			"multi_thread_write_one_tbl": "no",
 			"number_of_tbl_in_one_sql": 0,
 			"rows_per_tbl": 100,

src/os/inc/osTime.h

@@ -72,6 +72,11 @@ typedef struct SInterval {
   int64_t offset;
 } SInterval;
 
+typedef struct SSessionWindow {
+  int64_t gap;             // gap between two session window(in microseconds)
+  int32_t primaryColId;    // primary timestamp column
+} SSessionWindow;
+
 int64_t taosTimeAdd(int64_t t, int64_t duration, char unit, int32_t precision);
 int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precision);
 int32_t taosTimeCountInterval(int64_t skey, int64_t ekey, int64_t interval, char unit, int32_t precision);

src/plugins/http/inc/httpJson.h

@@ -63,9 +63,11 @@ void httpJsonString(JsonBuf* buf, char* sVal, int32_t len);
 void httpJsonOriginString(JsonBuf* buf, char* sVal, int32_t len);
 void httpJsonStringForTransMean(JsonBuf* buf, char* SVal, int32_t maxLen);
 void httpJsonInt64(JsonBuf* buf, int64_t num);
+void httpJsonUInt64(JsonBuf* buf, uint64_t num);
 void httpJsonTimestamp(JsonBuf* buf, int64_t t, bool us);
 void httpJsonUtcTimestamp(JsonBuf* buf, int64_t t, bool us);
 void httpJsonInt(JsonBuf* buf, int32_t num);
+void httpJsonUInt(JsonBuf* buf, uint32_t num);
 void httpJsonFloat(JsonBuf* buf, float num);
 void httpJsonDouble(JsonBuf* buf, double num);
 void httpJsonNull(JsonBuf* buf);

src/plugins/http/src/httpJson.c

@@ -256,6 +256,12 @@ void httpJsonInt64(JsonBuf* buf, int64_t num) {
   buf->lst += snprintf(buf->lst, MAX_NUM_STR_SZ, "%" PRId64, num);
 }
 
+void httpJsonUInt64(JsonBuf* buf, uint64_t num) {
+  httpJsonItemToken(buf);
+  httpJsonTestBuf(buf, MAX_NUM_STR_SZ);
+  buf->lst += snprintf(buf->lst, MAX_NUM_STR_SZ, "%" PRIu64, num);
+}
+
 void httpJsonTimestamp(JsonBuf* buf, int64_t t, bool us) {
   char       ts[35] = {0};
   struct tm* ptm;
@@ -303,6 +309,12 @@ void httpJsonInt(JsonBuf* buf, int32_t num) {
   buf->lst += snprintf(buf->lst, MAX_NUM_STR_SZ, "%d", num);
 }
 
+void httpJsonUInt(JsonBuf* buf, uint32_t num) {
+  httpJsonItemToken(buf);
+  httpJsonTestBuf(buf, MAX_NUM_STR_SZ);
+  buf->lst += snprintf(buf->lst, MAX_NUM_STR_SZ, "%u", num);
+}
+
 void httpJsonFloat(JsonBuf* buf, float num) {
   httpJsonItemToken(buf);
   httpJsonTestBuf(buf, MAX_NUM_STR_SZ);

src/plugins/http/src/httpRestJson.c

@@ -162,6 +162,18 @@ bool restBuildSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result,
         case TSDB_DATA_TYPE_BIGINT:
           httpJsonInt64(jsonBuf, *((int64_t *)row[i]));
           break;
+        case TSDB_DATA_TYPE_UTINYINT:
+          httpJsonUInt(jsonBuf, *((uint8_t *)row[i]));
+          break;
+        case TSDB_DATA_TYPE_USMALLINT:
+          httpJsonUInt(jsonBuf, *((uint16_t *)row[i]));
+          break;
+        case TSDB_DATA_TYPE_UINT:
+          httpJsonUInt(jsonBuf, *((uint32_t *)row[i]));
+          break;
+        case TSDB_DATA_TYPE_UBIGINT:
+          httpJsonUInt64(jsonBuf, *((uint64_t *)row[i]));
+          break;
         case TSDB_DATA_TYPE_FLOAT:
           httpJsonFloat(jsonBuf, GET_FLOAT_VAL(row[i]));
           break;

src/query/inc/qAggMain.h

@@ -26,6 +26,7 @@ extern "C" {
 #include "taosdef.h"
 #include "trpc.h"
 #include "tvariant.h"
+#include "tsdb.h"
 
 #define TSDB_FUNC_INVALID_ID  -1
 #define TSDB_FUNC_COUNT        0
@@ -70,15 +71,17 @@ extern "C" {
 #define TSDB_FUNC_AVG_IRATE    34
 
 #define TSDB_FUNC_TID_TAG      35
-#define TSDB_FUNC_HISTOGRAM    36
-#define TSDB_FUNC_HLL          37
-#define TSDB_FUNC_MODE         38
-#define TSDB_FUNC_SAMPLE       39
-#define TSDB_FUNC_CEIL         40
-#define TSDB_FUNC_FLOOR        41
-#define TSDB_FUNC_ROUND        42
-#define TSDB_FUNC_MAVG         43
-#define TSDB_FUNC_CSUM         44
+#define TSDB_FUNC_BLKINFO      36
+
+#define TSDB_FUNC_HISTOGRAM    37
+#define TSDB_FUNC_HLL          38
+#define TSDB_FUNC_MODE         39
+#define TSDB_FUNC_SAMPLE       40
+#define TSDB_FUNC_CEIL         41
+#define TSDB_FUNC_FLOOR        42
+#define TSDB_FUNC_ROUND        43
+#define TSDB_FUNC_MAVG         44
+#define TSDB_FUNC_CSUM         45
 
 
 #define TSDB_FUNCSTATE_SO           0x1u    // single output
@@ -214,13 +217,14 @@ typedef struct SAggFunctionInfo {
   void (*xFinalize)(SQLFunctionCtx *pCtx);
   void (*mergeFunc)(SQLFunctionCtx *pCtx);
 
-  int32_t (*dataReqFunc)(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId);
+  int32_t (*dataReqFunc)(SQLFunctionCtx *pCtx, STimeWindow* w, int32_t colId);
 } SAggFunctionInfo;
 
 #define GET_RES_INFO(ctx) ((ctx)->resultInfo)
 
 int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, int16_t *type,
                           int16_t *len, int32_t *interBytes, int16_t extLength, bool isSuperTable);
+int32_t isValidFunction(const char* name, int32_t len);
 
 #define IS_STREAM_QUERY_VALID(x)  (((x)&TSDB_FUNCSTATE_STREAM) != 0)
 #define IS_MULTIOUTPUT(x)         (((x)&TSDB_FUNCSTATE_MO) != 0)
@@ -242,12 +246,16 @@ typedef struct STwaInfo {
   STimeWindow win;
 } STwaInfo;
 
+struct SBufferWriter;
+void blockDistInfoToBinary(STableBlockDist* pDist, struct SBufferWriter* bw);
+void blockDistInfoFromBinary(const char* data, int32_t len, STableBlockDist* pDist);
+
 /* global sql function array */
 extern struct SAggFunctionInfo aAggs[];
 
 extern int32_t functionCompatList[]; // compatible check array list
 
-bool topbot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, const char *minval, const char *maxval);
+bool topbot_datablock_filter(SQLFunctionCtx *pCtx, const char *minval, const char *maxval);
 
 /**
  * the numOfRes should be kept, since it may be used later
@@ -258,14 +266,14 @@ bool topbot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, const cha
     (_r)->initialized = false; \
   } while (0)
 
-static FORCE_INLINE void initResultInfo(SResultRowCellInfo *pResInfo, uint32_t bufLen) {
+static FORCE_INLINE void initResultInfo(SResultRowCellInfo *pResInfo, int32_t bufLen) {
   pResInfo->initialized = true;  // the this struct has been initialized flag
   
   pResInfo->complete = false;
   pResInfo->hasResult = false;
   pResInfo->numOfRes = 0;
   
-  memset(GET_ROWCELL_INTERBUF(pResInfo), 0, (size_t)bufLen);
+  memset(GET_ROWCELL_INTERBUF(pResInfo), 0, bufLen);
 }
 
 #ifdef __cplusplus

src/query/inc/qExecutor.h

@@ -12,8 +12,8 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
-#ifndef TDENGINE_QUERYEXECUTOR_H
-#define TDENGINE_QUERYEXECUTOR_H
+#ifndef TDENGINE_QEXECUTOR_H
+#define TDENGINE_QEXECUTOR_H
 
 #include "os.h"
 
@@ -37,30 +37,24 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int
 #define Q_STATUS_EQUAL(p, s)  (((p) & (s)) != 0u)
 #define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP)
 
-#define SET_STABLE_QUERY_OVER(_q) ((_q)->tableIndex = (int32_t)((_q)->tableqinfoGroupInfo.numOfTables))
-#define IS_STASBLE_QUERY_OVER(_q) ((_q)->tableIndex >= (int32_t)((_q)->tableqinfoGroupInfo.numOfTables))
-
 #define GET_TABLEGROUP(q, _index)   ((SArray*) taosArrayGetP((q)->tableqinfoGroupInfo.pGroupList, (_index)))
 
+#define GET_NUM_OF_RESULTS(_r) (((_r)->outputBuf) == NULL? 0:((_r)->outputBuf)->info.rows)
+
 enum {
   // when query starts to execute, this status will set
       QUERY_NOT_COMPLETED = 0x1u,
 
-  /* result output buffer is full, current query is paused.
-   * this status is only exist in group-by clause and diff/add/division/multiply/ query.
-   */
-      QUERY_RESBUF_FULL = 0x2u,
-
   /* query is over
    * 1. this status is used in one row result query process, e.g., count/sum/first/last/ avg...etc.
    * 2. when all data within queried time window, it is also denoted as query_completed
    */
-      QUERY_COMPLETED = 0x4u,
+      QUERY_COMPLETED = 0x2u,
 
   /* when the result is not completed return to client, this status will be
    * usually used in case of interval query with interpolation option
    */
-      QUERY_OVER = 0x8u,
+      QUERY_OVER = 0x4u,
 };
 
 typedef struct SResultRowPool {
@@ -86,13 +80,13 @@ typedef struct SSqlGroupbyExpr {
 
 typedef struct SResultRow {
   int32_t       pageId;      // pageId & rowId is the position of current result in disk-based output buffer
-  int32_t       rowId:29;    // row index in buffer page
+  int32_t       offset:29;    // row index in buffer page
   bool          startInterp; // the time window start timestamp has done the interpolation already.
   bool          endInterp;   // the time window end timestamp has done the interpolation already.
   bool          closed;      // this result status: closed or opened
   uint32_t      numOfRows;   // number of rows of current time window
   SResultRowCellInfo*  pCellInfo;  // For each result column, there is a resultInfo
-  union {STimeWindow win; char* key;};  // start key of current time window
+  union {STimeWindow win; char* key;};  // start key of current result row
 } SResultRow;
 
 typedef struct SGroupResInfo {
@@ -106,12 +100,11 @@ typedef struct SGroupResInfo {
  * If the number of generated results is greater than this value,
  * query query will be halt and return results to client immediate.
  */
-typedef struct SResultRec {
+typedef struct SRspResultInfo {
   int64_t total;      // total generated result size in rows
-  int64_t rows;       // current result set size in rows
-  int64_t capacity;   // capacity of current result output buffer
+  int32_t capacity;   // capacity of current result output buffer
   int32_t threshold;  // result size threshold in rows.
-} SResultRec;
+} SRspResultInfo;
 
 typedef struct SResultRowInfo {
   SResultRow** pResult;    // result list
@@ -138,7 +131,6 @@ typedef struct SSingleColumnFilterInfo {
 typedef struct STableQueryInfo {
   TSKEY       lastKey;
   int32_t     groupIndex;     // group id in table list
-  int16_t     queryRangeSet;  // denote if the query range is set, only available for interval query
   tVariant    tag;
   STimeWindow win;
   STSCursor   cur;
@@ -179,82 +171,136 @@ typedef struct {
   SArray* pResult;  // SArray<SStddevInterResult>
 } SInterResult;
 
+typedef struct SSDataBlock {
+  SDataStatis *pBlockStatis;
+  SArray      *pDataBlock;
+  SDataBlockInfo info;
+} SSDataBlock;
+
 typedef struct SQuery {
+  SLimitVal        limit;
+
+  bool             stableQuery;      // super table query or not
+  bool             topBotQuery;      // TODO used bitwise flag
+  bool             groupbyColumn;    // denote if this is a groupby normal column query
+  bool             hasTagResults;    // if there are tag values in final result or not
+  bool             timeWindowInterpo;// if the time window start/end required interpolation
+  bool             queryBlockDist;    // if query data block distribution
+  bool             stabledev;        // super table stddev query
+  int32_t          interBufSize;     // intermediate buffer sizse
+
+  SOrderVal        order;
   int16_t          numOfCols;
   int16_t          numOfTags;
-  SOrderVal        order;
+
   STimeWindow      window;
   SInterval        interval;
+  SSessionWindow   sw;
   int16_t          precision;
   int16_t          numOfOutput;
   int16_t          fillType;
   int16_t          checkResultBuf;   // check if the buffer is full during scan each block
-  SLimitVal        limit;
 
   int32_t          srcRowSize;       // todo extract struct
   int32_t          resultRowSize;
+  int32_t          intermediateResultRowSize; // intermediate result row size, in case of top-k query.
   int32_t          maxSrcColumnSize;
   int32_t          tagLen;           // tag value length of current query
-
   SSqlGroupbyExpr* pGroupbyExpr;
   SExprInfo*       pExpr1;
   SExprInfo*       pExpr2;
   int32_t          numOfExpr2;
-
   SColumnInfo*     colList;
   SColumnInfo*     tagColList;
   int32_t          numOfFilterCols;
   int64_t*         fillVal;
-  uint32_t         status;             // query status
-  SResultRec       rec;
-  int32_t          pos;
-  tFilePage**      sdata;
-  STableQueryInfo* current;
-  int32_t          numOfCheckedBlocks; // number of check data blocks
-
   SOrderedPrjQueryInfo prjInfo;        // limit value for each vgroup, only available in global order projection query.
   SSingleColumnFilterInfo* pFilterInfo;
+
+  STableQueryInfo* current;
+  void*            tsdb;
+  SMemRef          memRef;
+  STableGroupInfo  tableGroupInfo;       // table <tid, last_key> list  SArray<STableKeyInfo>
+  int32_t          vgId;
 } SQuery;
 
+typedef SSDataBlock* (*__operator_fn_t)(void* param);
+typedef void (*__optr_cleanup_fn_t)(void* param, int32_t num);
+
+struct SOperatorInfo;
+
 typedef struct SQueryRuntimeEnv {
-  jmp_buf              env;
-  SQuery*              pQuery;
-  SQLFunctionCtx*      pCtx;
-  int32_t              numOfRowsPerPage;
-  uint16_t*            offset;
-  uint16_t             scanFlag;         // denotes reversed scan of data or not
-  SFillInfo*           pFillInfo;
-  SResultRowInfo       resultRowInfo;
-
-  SQueryCostInfo       summary;
-  void*                pQueryHandle;
-  void*                pSecQueryHandle;  // another thread for
-  bool                 stableQuery;      // super table query or not
-  bool                 topBotQuery;      // TODO used bitwise flag
-  bool                 groupbyColumn;    // denote if this is a groupby normal column query
-  bool                 hasTagResults;    // if there are tag values in final result or not
-  bool                 timeWindowInterpo;// if the time window start/end required interpolation
-  bool                 queryWindowIdentical; // all query time windows are identical for all tables in one group
-  bool                 queryBlockDist;    // if query data block distribution
-  bool                 stabledev;        // super table stddev query
-  int32_t              interBufSize;     // intermediate buffer sizse
-  int32_t              prevGroupId;      // previous executed group id
-  SDiskbasedResultBuf* pResultBuf;       // query result buffer based on blocked-wised disk file
-  SHashObj*            pResultRowHashTable; // quick locate the window object for each result
-  char*                keyBuf;           // window key buffer
-  SResultRowPool*      pool;             // window result object pool
-
-  int32_t*             rowCellInfoOffset;// offset value for each row result cell info
-  char**               prevRow;
-
-  SArray*              prevResult;       // intermediate result, SArray<SInterResult>
-  STSBuf*              pTsBuf;           // timestamp filter list
-  STSCursor            cur;
-
-  char*                tagVal;           // tag value of current data block
-  SArithmeticSupport  *sasArray;
+  jmp_buf               env;
+  SQuery*               pQuery;
+  uint32_t              status;           // query status
+  void*                 qinfo;
+  uint8_t               scanFlag;         // denotes reversed scan of data or not
+  void*                 pQueryHandle;
+
+  int32_t               prevGroupId;      // previous executed group id
+  SDiskbasedResultBuf*  pResultBuf;       // query result buffer based on blocked-wised disk file
+  SHashObj*             pResultRowHashTable; // quick locate the window object for each result
+  char*                 keyBuf;           // window key buffer
+  SResultRowPool*       pool;             // window result object pool
+  char**                prevRow;
+
+  SArray*               prevResult;       // intermediate result, SArray<SInterResult>
+  STSBuf*               pTsBuf;           // timestamp filter list
+  STSCursor             cur;
+
+  char*                 tagVal;           // tag value of current data block
+  SArithmeticSupport   *sasArray;
+
+  SSDataBlock          *outputBuf;
+  STableGroupInfo       tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
+  struct SOperatorInfo *proot;
+  struct SOperatorInfo *pTableScanner;   // table scan operator
+  SGroupResInfo         groupResInfo;
+  int64_t               currentOffset;   // dynamic offset value
+
+  SRspResultInfo        resultInfo;
+  SHashObj             *pTableRetrieveTsMap;
 } SQueryRuntimeEnv;
 
+enum {
+  OP_IN_EXECUTING   = 1,
+  OP_RES_TO_RETURN  = 2,
+  OP_EXEC_DONE      = 3,
+};
+
+enum OPERATOR_TYPE_E {
+  OP_TableScan         = 1,
+  OP_DataBlocksOptScan = 2,
+  OP_TableSeqScan      = 3,
+  OP_TagScan           = 4,
+  OP_TableBlockInfoScan= 5,
+  OP_Aggregate         = 6,
+  OP_Arithmetic        = 7,
+  OP_Groupby           = 8,
+  OP_Limit             = 9,
+  OP_Offset            = 10,
+  OP_TimeWindow        = 11,
+  OP_SessionWindow     = 12,
+  OP_Fill              = 13,
+  OP_MultiTableAggregate     = 14,
+  OP_MultiTableTimeInterval  = 15,
+};
+
+typedef struct SOperatorInfo {
+  uint8_t               operatorType;
+  bool                  blockingOptr;  // block operator or not
+  uint8_t               status;        // denote if current operator is completed
+  int32_t               numOfOutput;   // number of columns of the current operator results
+  char                 *name;          // name, used to show the query execution plan
+  void                 *info;          // extension attribution
+  SExprInfo            *pExpr;
+  SQueryRuntimeEnv     *pRuntimeEnv;
+
+  struct SOperatorInfo *upstream;
+  __operator_fn_t       exec;
+  __optr_cleanup_fn_t   cleanup;
+} SOperatorInfo;
+
 enum {
   QUERY_RESULT_NOT_READY = 1,
   QUERY_RESULT_READY     = 2,
@@ -263,23 +309,11 @@ enum {
 typedef struct SQInfo {
   void*            signature;
   uint64_t         qId;
-  int32_t          code;   // error code to returned to client
-  int64_t          owner;  // if it is in execution
-  void*            tsdb;
-  SMemRef          memRef; 
-  int32_t          vgId;
-  STableGroupInfo  tableGroupInfo;       // table <tid, last_key> list  SArray<STableKeyInfo>
-  STableGroupInfo  tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
-  SQueryRuntimeEnv runtimeEnv;
-  SHashObj*        arrTableIdInfo;
-  int32_t          groupIndex;
+  int32_t          code;        // error code to returned to client
+  int64_t          owner;       // if it is in execution
 
-  /*
-   * the query is executed position on which meter of the whole list.
-   * when the index reaches the last one of the list, it means the query is completed.
-   */
-  int32_t          tableIndex;
-  SGroupResInfo    groupResInfo;
+  SQueryRuntimeEnv runtimeEnv;
+  SQuery           query;
   void*            pBuf;        // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;
 
   pthread_mutex_t  lock;        // used to synchronize the rsp/query threads
@@ -288,6 +322,7 @@ typedef struct SQInfo {
   void*            rspContext;  // response context
   int64_t          startExecTs; // start to exec timestamp
   char*            sql;         // query sql string
+  SQueryCostInfo   summary;
 } SQInfo;
 
 typedef struct SQueryParam {
@@ -306,10 +341,93 @@ typedef struct SQueryParam {
   SSqlGroupbyExpr *pGroupbyExpr;
 } SQueryParam;
 
+typedef struct STableScanInfo {
+  void           *pQueryHandle;
+  int32_t         numOfBlocks;
+  int32_t         numOfSkipped;
+  int32_t         numOfBlockStatis;
+  int64_t         numOfRows;
+                 
+  int32_t         order;        // scan order
+  int32_t         times;        // repeat counts
+  int32_t         current;
+  int32_t         reverseTimes; // 0 by default
+
+  SQLFunctionCtx *pCtx;         // next operator query context
+  SResultRowInfo *pResultRowInfo;
+  int32_t        *rowCellInfoOffset;
+  SExprInfo      *pExpr;
+  SSDataBlock     block;
+  bool            loadExternalRows; // load external rows (prev & next rows)
+  int32_t         numOfOutput;
+  int64_t         elapsedTime;
+
+  int32_t         tableIndex;
+} STableScanInfo;
+
+typedef struct STagScanInfo {
+  SColumnInfo* pCols;
+  SSDataBlock* pRes;
+  int32_t      totalTables;
+  int32_t      currentIndex;
+} STagScanInfo;
+
+typedef struct SOptrBasicInfo {
+  SResultRowInfo    resultRowInfo;
+  int32_t          *rowCellInfoOffset;  // offset value for each row result cell info
+  SQLFunctionCtx   *pCtx;
+  SSDataBlock      *pRes;
+} SOptrBasicInfo;
+
+typedef struct SOptrBasicInfo STableIntervalOperatorInfo;
+
+typedef struct SAggOperatorInfo {
+  SOptrBasicInfo binfo;
+  uint32_t       seed;
+} SAggOperatorInfo;
+
+typedef struct SArithOperatorInfo {
+  SOptrBasicInfo binfo;
+  int32_t        bufCapacity;
+  uint32_t       seed;
+} SArithOperatorInfo;
+
+typedef struct SLimitOperatorInfo {
+  int64_t limit;
+  int64_t total;
+} SLimitOperatorInfo;
+
+typedef struct SOffsetOperatorInfo {
+  int64_t offset;
+} SOffsetOperatorInfo;
+
+typedef struct SFillOperatorInfo {
+  SFillInfo   *pFillInfo;
+  SSDataBlock *pRes;
+  int64_t      totalInputRows;
+} SFillOperatorInfo;
+
+typedef struct SGroupbyOperatorInfo {
+  SOptrBasicInfo binfo;
+  int32_t        colIndex;
+  char          *prevData;   // previous group by value
+} SGroupbyOperatorInfo;
+
+typedef struct SSWindowOperatorInfo {
+  SOptrBasicInfo binfo;
+  STimeWindow    curWindow;  // current time window
+  TSKEY          prevTs;     // previous timestamp
+  int32_t        numOfRows;  // number of rows
+  int32_t        start;      // start row index
+} SSWindowOperatorInfo;
+
 void freeParam(SQueryParam *param);
 int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param);
 int32_t createQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo, SSqlFuncMsg **pExprMsg,
                                    SColumnInfo* pTagCols);
+int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo,
+                                           SSqlFuncMsg **pExprMsg, SExprInfo *prevExpr);
+
 SSqlGroupbyExpr *createGroupbyExprFromMsg(SQueryTableMsg *pQueryMsg, SColIndex *pColIndex, int32_t *code);
 SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
                         SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, bool stableQuery, char* sql, uint64_t *qId);
@@ -319,13 +437,9 @@ void freeColumnFilterInfo(SColumnFilterInfo* pFilter, int32_t numOfFilters);
 bool isQueryKilled(SQInfo *pQInfo);
 int32_t checkForQueryBuf(size_t numOfTables);
 bool doBuildResCheck(SQInfo* pQInfo);
-void setQueryStatus(SQuery *pQuery, int8_t status);
+void setQueryStatus(SQueryRuntimeEnv *pRuntimeEnv, int8_t status);
 
 bool onlyQueryTags(SQuery* pQuery);
-void buildTagQueryResult(SQInfo *pQInfo);
-void stableQueryImpl(SQInfo *pQInfo);
-void buildTableBlockDistResult(SQInfo *pQInfo);
-void tableQueryImpl(SQInfo *pQInfo);
 bool isValidQInfo(void *param);
 
 int32_t doDumpQueryResult(SQInfo *pQInfo, char *data);
@@ -337,4 +451,4 @@ void freeQInfo(SQInfo *pQInfo);
 
 int32_t getMaximumIdleDurationSec();
 
-#endif  // TDENGINE_QUERYEXECUTOR_H
+#endif  // TDENGINE_QEXECUTOR_H

src/query/inc/qFill.h

@@ -24,6 +24,8 @@ extern "C" {
 #include "qExtbuffer.h"
 #include "taosdef.h"
 
+struct SSDataBlock;
+
 typedef struct {
   STColumn col;             // column info
   int16_t  functionId;      // sql function id
@@ -78,7 +80,7 @@ void* taosDestroyFillInfo(SFillInfo *pFillInfo);
 
 void taosFillSetStartInfo(SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey);
 
-void taosFillSetDataBlockFromFilePage(SFillInfo* pFillInfo, const tFilePage** pInput);
+void taosFillSetInputDataBlock(SFillInfo* pFillInfo, const struct SSDataBlock* pInput);
 
 void taosFillCopyInputDataFromOneFilePage(SFillInfo* pFillInfo, const tFilePage* pInput);
 
@@ -88,7 +90,7 @@ int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, int64_t ekey, int32_t
 
 int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint* point1, SPoint* point2, int32_t inputType);
 
-int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, tFilePage** output, int32_t capacity);
+int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, void** output, int32_t capacity);
 
 #ifdef __cplusplus
 }

src/query/inc/qResultbuf.h

@@ -55,7 +55,6 @@ typedef struct SResultBufStatis {
 } SResultBufStatis;
 
 typedef struct SDiskbasedResultBuf {
-  int32_t   numOfRowsPerPage;
   int32_t   numOfPages;
   int64_t   totalBufSize;
   int64_t   fileSize;            // disk file size
@@ -77,7 +76,7 @@ typedef struct SDiskbasedResultBuf {
   SResultBufStatis statis;
 } SDiskbasedResultBuf;
 
-#define DEFAULT_INTERN_BUF_PAGE_SIZE  (256L)                          // in bytes
+#define DEFAULT_INTERN_BUF_PAGE_SIZE  (1024L)                          // in bytes
 #define PAGE_INFO_INITIALIZER         (SPageDiskInfo){-1, -1}
 
 /**
@@ -89,8 +88,7 @@ typedef struct SDiskbasedResultBuf {
  * @param handle
  * @return
  */
-int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t rowSize, int32_t pagesize,
-                                    int32_t inMemBufSize, const void* handle);
+int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pagesize, int32_t inMemBufSize, const void* handle);
 
 /**
  *
@@ -101,13 +99,6 @@ int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t ro
  */
 tFilePage* getNewDataBuf(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int32_t* pageId);
 
-/**
- *
- * @param pResultBuf
- * @return
- */
-size_t getNumOfRowsPerPage(const SDiskbasedResultBuf* pResultBuf);
-
 /**
  *
  * @param pResultBuf

src/query/inc/qSqlparser.h

@@ -45,7 +45,7 @@ typedef struct SLimitVal {
 
 typedef struct SOrderVal {
   uint32_t order;
-  int32_t orderColId;
+  int32_t  orderColId;
 } SOrderVal;
 
 typedef struct tVariantListItem {
@@ -58,14 +58,19 @@ typedef struct SIntervalVal {
   SStrToken offset;
 } SIntervalVal;
 
+typedef struct SSessionWindowVal {
+  SStrToken col;
+  SStrToken gap;
+} SSessionWindowVal;
+
 typedef struct SQuerySQL {
   struct tSQLExprList *pSelection;   // select clause
   SArray *             from;         // from clause  SArray<tVariantListItem>
   struct tSQLExpr *    pWhere;       // where clause [optional]
   SArray *             pGroupby;     // groupby clause, only for tags[optional], SArray<tVariantListItem>
   SArray *             pSortOrder;   // orderby [optional], SArray<tVariantListItem>
-  SStrToken            interval;     // interval [optional]
-  SStrToken            offset;       // offset window [optional]
+  SIntervalVal         interval;     // (interval, interval_offset) [optional]
+  SSessionWindowVal    sessionVal;   // session window [optional]
   SStrToken            sliding;      // sliding window [optional]
   SLimitVal            limit;        // limit offset [optional]
   SLimitVal            slimit;       // group limit offset [optional]
@@ -193,19 +198,32 @@ typedef struct SSqlInfo {
   };
 } SSqlInfo;
 
+#define NON_ARITHMEIC_EXPR 0
+#define NORMAL_ARITHMETIC  1
+#define AGG_ARIGHTMEIC     2
+
+enum SQL_NODE_TYPE {
+  SQL_NODE_TABLE_COLUMN= 1,
+  SQL_NODE_SQLFUNCTION = 2,
+  SQL_NODE_VALUE       = 3,
+  SQL_NODE_EXPR        = 4,
+};
+
 typedef struct tSQLExpr {
-  uint32_t         nSQLOptr;    // TK_FUNCTION: sql function, TK_LE: less than(binary expr)
-  
-  // the full sql string of function(col, param), which is actually the raw
-  // field name, since the function name is kept in nSQLOptr already
+  uint16_t         type;       // sql node type
+  uint32_t         tokenId;    // TK_FUNCTION: sql function, TK_LE: less than(binary expr)
+
+  // the whole string of the function(col, param), while the function name is kept in token
   SStrToken        operand;
-  SStrToken        colInfo;     // field id
-  tVariant         val;         // value only for string, float, int
+  uint32_t         functionId;  // function id
+
+  SStrToken        colInfo;     // table column info
+  tVariant         value;       // the use input value
   SStrToken        token;       // original sql expr string
 
   struct tSQLExpr *pLeft;       // left child
   struct tSQLExpr *pRight;      // right child
-  struct tSQLExprList *pParam;  // function parameters
+  struct tSQLExprList *pParam;  // function parameters list
 } tSQLExpr;
 
 // used in select clause. select <tSQLExprList> from xxx
@@ -251,8 +269,8 @@ tSQLExprList *tSqlExprListAppend(tSQLExprList *pList, tSQLExpr *pNode, SStrToken
 
 void tSqlExprListDestroy(tSQLExprList *pList);
 
-SQuerySQL *tSetQuerySqlElems(SStrToken *pSelectToken, tSQLExprList *pSelection, SArray *pFrom, tSQLExpr *pWhere,
-                             SArray *pGroupby, SArray *pSortOrder, SIntervalVal *pInterval,
+SQuerySQL *tSetQuerySqlNode(SStrToken *pSelectToken, tSQLExprList *pSelection, SArray *pFrom, tSQLExpr *pWhere,
+                             SArray *pGroupby, SArray *pSortOrder, SIntervalVal *pInterval, SSessionWindowVal *pSession,
                              SStrToken *pSliding, SArray *pFill, SLimitVal *pLimit, SLimitVal *pGLimit);
 
 SCreateTableSQL *tSetCreateSqlElems(SArray *pCols, SArray *pTags, SQuerySQL *pSelect, int32_t type);
@@ -302,16 +320,6 @@ void tSqlSetColumnType(TAOS_FIELD *pField, SStrToken *type);
 
 void *ParseAlloc(void *(*mallocProc)(size_t));
 
-enum {
-  TSQL_NODE_TYPE_EXPR  = 0x1,
-  TSQL_NODE_TYPE_ID    = 0x2,
-  TSQL_NODE_TYPE_VALUE = 0x4,
-};
-
-#define NON_ARITHMEIC_EXPR 0
-#define NORMAL_ARITHMETIC  1
-#define AGG_ARIGHTMEIC     2
-
 SSqlInfo qSQLParse(const char *str);
 
 #ifdef __cplusplus

src/query/inc/qTsbuf.h

@@ -112,13 +112,11 @@ STSBuf* tsBufClone(STSBuf* pTSBuf);
 
 STSGroupBlockInfo* tsBufGetGroupBlockInfo(STSBuf* pTSBuf, int32_t id);
 
-void tsBufFlush(STSBuf* pTSBuf);
-
+void    tsBufFlush(STSBuf* pTSBuf);
 void    tsBufResetPos(STSBuf* pTSBuf);
-STSElem tsBufGetElem(STSBuf* pTSBuf);
-
 bool    tsBufNextPos(STSBuf* pTSBuf);
 
+STSElem tsBufGetElem(STSBuf* pTSBuf);
 STSElem tsBufGetElemStartPos(STSBuf* pTSBuf, int32_t id, tVariant* tag);
 
 STSCursor tsBufGetCursor(STSBuf* pTSBuf);

src/query/inc/qUtil.h

@@ -27,7 +27,7 @@
 #define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t))
 
 #define curTimeWindowIndex(_winres)        ((_winres)->curIndex)
-#define GET_ROW_PARAM_FOR_MULTIOUTPUT(_q, tbq, sq) (((tbq) && (!sq))? (_q)->pExpr1[1].base.arg->argValue.i64:1)
+#define GET_ROW_PARAM_FOR_MULTIOUTPUT(_q, tbq, sq) (((tbq) && (!(sq)))? (_q)->pExpr1[1].base.arg->argValue.i64:1)
 
 int32_t getOutputInterResultBufSize(SQuery* pQuery);
 
@@ -44,22 +44,18 @@ void    closeResultRow(SResultRowInfo* pResultRowInfo, int32_t slot);
 bool    isResultRowClosed(SResultRowInfo *pResultRowInfo, int32_t slot);
 void    clearResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* pResultRow, int16_t type);
 
-SResultRowCellInfo* getResultCell(SQueryRuntimeEnv* pRuntimeEnv, const SResultRow* pRow, int32_t index);
+SResultRowCellInfo* getResultCell(const SResultRow* pRow, int32_t index, int32_t* offset);
 
 static FORCE_INLINE SResultRow *getResultRow(SResultRowInfo *pResultRowInfo, int32_t slot) {
   assert(pResultRowInfo != NULL && slot >= 0 && slot < pResultRowInfo->size);
   return pResultRowInfo->pResult[slot];
 }
 
-static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int32_t columnIndex, SResultRow *pResult,
-    tFilePage* page) {
-  assert(pResult != NULL && pRuntimeEnv != NULL);
+static FORCE_INLINE char *getPosInResultPage(SQuery *pQuery, tFilePage* page, int32_t rowOffset, int16_t offset) {
+  assert(rowOffset >= 0 && pQuery != NULL);
 
-  SQuery *pQuery = pRuntimeEnv->pQuery;
-
-  int32_t realRowId = (int32_t)(pResult->rowId * GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, pRuntimeEnv->topBotQuery, pRuntimeEnv->stableQuery));
-  return ((char *)page->data) + pRuntimeEnv->offset[columnIndex] * pRuntimeEnv->numOfRowsPerPage +
-      pQuery->pExpr1[columnIndex].bytes * realRowId;
+  int32_t numOfRows = (int32_t)GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, pQuery->topBotQuery, pQuery->stableQuery);
+  return ((char *)page->data) + rowOffset + offset * numOfRows;
 }
 
 bool isNullOperator(SColumnFilterElem *pFilter, const char* minval, const char* maxval, int16_t type);
@@ -74,8 +70,6 @@ void* destroyResultRowPool(SResultRowPool* p);
 int32_t getNumOfAllocatedResultRows(SResultRowPool* p);
 int32_t getNumOfUsedResultRows(SResultRowPool* p);
 
-bool isPointInterpoQuery(SQuery *pQuery);
-
 typedef struct {
   SArray* pResult;     // SArray<SResPair>
   int32_t colId;
@@ -85,12 +79,14 @@ void interResToBinary(SBufferWriter* bw, SArray* pRes, int32_t tagLen);
 SArray* interResFromBinary(const char* data, int32_t len);
 void freeInterResult(void* param);
 
-void    initGroupResInfo(SGroupResInfo* pGroupResInfo, SResultRowInfo* pResultInfo, int32_t offset);
+void    initGroupResInfo(SGroupResInfo* pGroupResInfo, SResultRowInfo* pResultInfo);
 void    cleanupGroupResInfo(SGroupResInfo* pGroupResInfo);
+bool    hasRemainDataInCurrentGroup(SGroupResInfo* pGroupResInfo);
 bool    hasRemainData(SGroupResInfo* pGroupResInfo);
+
 bool    incNextGroup(SGroupResInfo* pGroupResInfo);
 int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);
 
-int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, SQInfo *pQInfo);
+int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, SQueryRuntimeEnv *pRuntimeEnv, int32_t* offset);
 
 #endif  // TDENGINE_QUERYUTIL_H

src/query/inc/sql.y

@@ -329,8 +329,8 @@ signed(A) ::= PLUS INTEGER(X).    { A = strtol(X.z, NULL, 10); }
 signed(A) ::= MINUS INTEGER(X).   { A = -strtol(X.z, NULL, 10);}
 
 ////////////////////////////////// The CREATE TABLE statement ///////////////////////////////
-cmd ::= CREATE TABLE create_table_args. {}
-cmd ::= CREATE TABLE create_stable_args. {}
+cmd ::= CREATE TABLE  create_table_args.  {}
+cmd ::= CREATE TABLE  create_stable_args. {}
 cmd ::= CREATE STABLE create_stable_args. {}
 cmd ::= CREATE TABLE create_table_list(Z). { pInfo->type = TSDB_SQL_CREATE_TABLE; pInfo->pCreateTableInfo = Z;}
 
@@ -455,8 +455,8 @@ tagitem(A) ::= PLUS(X) FLOAT(Y).  {
 //////////////////////// The SELECT statement /////////////////////////////////
 %type select {SQuerySQL*}
 %destructor select {doDestroyQuerySql($$);}
-select(A) ::= SELECT(T) selcollist(W) from(X) where_opt(Y) interval_opt(K) fill_opt(F) sliding_opt(S) groupby_opt(P) orderby_opt(Z) having_opt(N) slimit_opt(G) limit_opt(L). {
-  A = tSetQuerySqlElems(&T, W, X, Y, P, Z, &K, &S, F, &L, &G);
+select(A) ::= SELECT(T) selcollist(W) from(X) where_opt(Y) interval_opt(K) session_option(H) fill_opt(F) sliding_opt(S) groupby_opt(P) orderby_opt(Z) having_opt(N) slimit_opt(G) limit_opt(L). {
+  A = tSetQuerySqlNode(&T, W, X, Y, P, Z, &K, &H, &S, F, &L, &G);
 }
 
 %type union {SSubclauseInfo*}
@@ -474,7 +474,7 @@ cmd ::= union(X). { setSqlInfo(pInfo, X, NULL, TSDB_SQL_SELECT); }
 // select server_version(), select client_version(),
 // select server_state();
 select(A) ::= SELECT(T) selcollist(W). {
-  A = tSetQuerySqlElems(&T, W, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL);
+  A = tSetQuerySqlNode(&T, W, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL);
 }
 
 // selcollist is a list of expressions that are to become the return
@@ -549,13 +549,21 @@ tablelist(A) ::= tablelist(Y) COMMA ids(X) cpxName(Z) ids(F). {
 tmvar(A) ::= VARIABLE(X).   {A = X;}
 
 %type interval_opt {SIntervalVal}
-interval_opt(N) ::= INTERVAL LP tmvar(E) RP.    {N.interval = E; N.offset.n = 0; N.offset.z = NULL; N.offset.type = 0;}
-interval_opt(N) ::= INTERVAL LP tmvar(E) COMMA tmvar(O) RP.    {N.interval = E; N.offset = O;}
+interval_opt(N) ::= INTERVAL LP tmvar(E) RP.    {N.interval = E; N.offset.n = 0;}
+interval_opt(N) ::= INTERVAL LP tmvar(E) COMMA tmvar(X) RP.    {N.interval = E; N.offset = X;}
 interval_opt(N) ::= .                           {memset(&N, 0, sizeof(N));}
 
+%type session_option {SSessionWindowVal}
+session_option(X) ::= .                                                  {X.col.n = 0; X.gap.n = 0;}
+session_option(X) ::= SESSION LP ids(V) cpxName(Z) COMMA tmvar(Y) RP.    {
+   V.n += Z.n;
+   X.col = V;
+   X.gap = Y;
+}
+
 %type fill_opt {SArray*}
 %destructor fill_opt {taosArrayDestroy($$);}
-fill_opt(N) ::= .                               {N = 0;     }
+fill_opt(N) ::= .                                           { N = 0;     }
 fill_opt(N) ::= FILL LP ID(Y) COMMA tagitemlist(X) RP.      {
     tVariant A = {0};
     toTSDBType(Y.type);
@@ -837,6 +845,5 @@ cmd ::= KILL QUERY INTEGER(X) COLON(Z) INTEGER(Y).        {X.n += (Z.n + Y.n); s
 %fallback ID ABORT AFTER ASC ATTACH BEFORE BEGIN CASCADE CLUSTER CONFLICT COPY DATABASE DEFERRED
   DELIMITERS DESC DETACH EACH END EXPLAIN FAIL FOR GLOB IGNORE IMMEDIATE INITIALLY INSTEAD
   LIKE MATCH KEY OF OFFSET RAISE REPLACE RESTRICT ROW STATEMENT TRIGGER VIEW ALL
-  COUNT SUM AVG MIN MAX FIRST LAST TOP BOTTOM STDDEV PERCENTILE APERCENTILE LEASTSQUARES HISTOGRAM DIFF
-  SPREAD TWA INTERP LAST_ROW RATE IRATE SUM_RATE SUM_IRATE AVG_RATE AVG_IRATE TBID NOW IPTOKEN SEMI NONE PREV LINEAR IMPORT
+  NOW IPTOKEN SEMI NONE PREV LINEAR IMPORT
   METRIC TBNAME JOIN METRICS STABLE NULL INSERT INTO VALUES.

src/query/src/qFill.c

@@ -23,18 +23,19 @@
 #include "qFill.h"
 #include "qExtbuffer.h"
 #include "queryLog.h"
+#include "qExecutor.h"
 
 #define FILL_IS_ASC_FILL(_f) ((_f)->order == TSDB_ORDER_ASC)
 #define DO_INTERPOLATION(_v1, _v2, _k1, _k2, _k) ((_v1) + ((_v2) - (_v1)) * (((double)(_k)) - ((double)(_k1))) / (((double)(_k2)) - ((double)(_k1))))
 
-static void setTagsValue(SFillInfo* pFillInfo, tFilePage** data, int32_t genRows) {
+static void setTagsValue(SFillInfo* pFillInfo, void** data, int32_t genRows) {
   for(int32_t j = 0; j < pFillInfo->numOfCols; ++j) {
     SFillColInfo* pCol = &pFillInfo->pFillCol[j];
     if (TSDB_COL_IS_NORMAL_COL(pCol->flag)) {
       continue;
     }
 
-    char* val1 = elePtrAt(data[j]->data, pCol->col.bytes, genRows);
+    char* val1 = elePtrAt(data[j], pCol->col.bytes, genRows);
 
     assert(pCol->tagIndex >= 0 && pCol->tagIndex < pFillInfo->numOfTags);
     SFillTagColInfo* pTag = &pFillInfo->pTags[pCol->tagIndex];
@@ -44,17 +45,17 @@ static void setTagsValue(SFillInfo* pFillInfo, tFilePage** data, int32_t genRows
   }
 }
 
-static void setNullValueForRow(SFillInfo* pFillInfo, tFilePage** data, int32_t numOfCol, int32_t rowIndex) {
+static void setNullValueForRow(SFillInfo* pFillInfo, void** data, int32_t numOfCol, int32_t rowIndex) {
   // the first are always the timestamp column, so start from the second column.
   for (int32_t i = 1; i < numOfCol; ++i) {
     SFillColInfo* pCol = &pFillInfo->pFillCol[i];
 
-    char* output = elePtrAt(data[i]->data, pCol->col.bytes, rowIndex);
+    char* output = elePtrAt(data[i], pCol->col.bytes, rowIndex);
     setNull(output, pCol->col.type, pCol->col.bytes);
   }
 }
 
-static void doFillOneRowResult(SFillInfo* pFillInfo, tFilePage** data, char** srcData, int64_t ts, bool outOfBound) {
+static void doFillOneRowResult(SFillInfo* pFillInfo, void** data, char** srcData, int64_t ts, bool outOfBound) {
   char* prev = pFillInfo->prevValues;
   char* next = pFillInfo->nextValues;
 
@@ -63,7 +64,7 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, tFilePage** data, char** sr
 
   // set the primary timestamp column value
   int32_t index = pFillInfo->numOfCurrent;
-  char* val = elePtrAt(data[0]->data, TSDB_KEYSIZE, index);
+  char* val = elePtrAt(data[0], TSDB_KEYSIZE, index);
   *(TSKEY*) val = pFillInfo->currentKey;
 
   // set the other values
@@ -77,7 +78,7 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, tFilePage** data, char** sr
           continue;
         }
 
-        char* output = elePtrAt(data[i]->data, pCol->col.bytes, index);
+        char* output = elePtrAt(data[i], pCol->col.bytes, index);
         assignVal(output, p + pCol->col.offset, pCol->col.bytes, pCol->col.type);
       }
     } else {  // no prev value yet, set the value for NULL
@@ -93,7 +94,7 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, tFilePage** data, char** sr
           continue;
         }
 
-        char* output = elePtrAt(data[i]->data, pCol->col.bytes, index);
+        char* output = elePtrAt(data[i], pCol->col.bytes, index);
         assignVal(output, p + pCol->col.offset, pCol->col.bytes, pCol->col.type);
       }
     } else { // no prev value yet, set the value for NULL
@@ -111,7 +112,7 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, tFilePage** data, char** sr
         int16_t type  = pCol->col.type;
         int16_t bytes = pCol->col.bytes;
 
-        char *val1 = elePtrAt(data[i]->data, pCol->col.bytes, index);
+        char *val1 = elePtrAt(data[i], pCol->col.bytes, index);
         if (type == TSDB_DATA_TYPE_BINARY|| type == TSDB_DATA_TYPE_NCHAR || type == TSDB_DATA_TYPE_BOOL) {
           setNull(val1, pCol->col.type, bytes);
           continue;
@@ -132,7 +133,7 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, tFilePage** data, char** sr
         continue;
       }
 
-      char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, index);
+      char* val1 = elePtrAt(data[i], pCol->col.bytes, index);
       assignVal(val1, (char*)&pCol->fillVal.i, pCol->col.bytes, pCol->col.type);
     }
   }
@@ -162,7 +163,7 @@ static void copyCurrentRowIntoBuf(SFillInfo* pFillInfo, char** srcData, char* bu
   }
 }
 
-static int32_t fillResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t outputRows) {
+static int32_t fillResultImpl(SFillInfo* pFillInfo, void** data, int32_t outputRows) {
   pFillInfo->numOfCurrent = 0;
 
   char** srcData = pFillInfo->pData;
@@ -213,7 +214,7 @@ static int32_t fillResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t ou
           continue;
         }
 
-        char* output = elePtrAt(data[i]->data, pCol->col.bytes, pFillInfo->numOfCurrent);
+        char* output = elePtrAt(data[i], pCol->col.bytes, pFillInfo->numOfCurrent);
         char* src = elePtrAt(srcData[i], pCol->col.bytes, pFillInfo->index);
 
         if (i == 0 || (pCol->functionId != TSDB_FUNC_COUNT && !isNull(src, pCol->col.type)) ||
@@ -255,7 +256,7 @@ static int32_t fillResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t ou
   return pFillInfo->numOfCurrent;
 }
 
-static int64_t appendFilledResult(SFillInfo* pFillInfo, tFilePage** output, int64_t resultCapacity) {
+static int64_t appendFilledResult(SFillInfo* pFillInfo, void** output, int64_t resultCapacity) {
   /*
    * These data are generated according to fill strategy, since the current timestamp is out of the time window of
    * real result set. Note that we need to keep the direct previous result rows, to generated the filled data.
@@ -278,7 +279,7 @@ static int32_t setTagColumnInfo(SFillInfo* pFillInfo, int32_t numOfCols, int32_t
   int32_t k = 0;
   for (int32_t i = 0; i < numOfCols; ++i) {
     SFillColInfo* pColInfo = &pFillInfo->pFillCol[i];
-    pFillInfo->pData[i] = calloc(1, pColInfo->col.bytes * capacity);
+    pFillInfo->pData[i] = NULL;
 
     if (TSDB_COL_IS_TAG(pColInfo->flag)) {
       bool exists = false;
@@ -356,6 +357,10 @@ SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int3
   pFillInfo->rowSize = setTagColumnInfo(pFillInfo, pFillInfo->numOfCols, pFillInfo->alloc);
   assert(pFillInfo->rowSize > 0);
 
+  for(int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
+    pFillInfo->pData[i] = malloc(pFillInfo->pFillCol[i].col.bytes * pFillInfo->alloc);
+  }
+
   return pFillInfo;
 }
 
@@ -375,11 +380,16 @@ void* taosDestroyFillInfo(SFillInfo* pFillInfo) {
 
   tfree(pFillInfo->prevValues);
   tfree(pFillInfo->nextValues);
-  tfree(pFillInfo->pTags);
-  
+
+  for(int32_t i = 0; i < pFillInfo->numOfTags; ++i) {
+    tfree(pFillInfo->pTags[i].tagVal);
+  }
+
   for(int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
     tfree(pFillInfo->pData[i]);
   }
+
+  tfree(pFillInfo->pTags);
   
   tfree(pFillInfo->pData);
   tfree(pFillInfo->pFillCol);
@@ -413,10 +423,19 @@ void taosFillSetStartInfo(SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey)
   }
 }
 
-// copy the data into source data buffer
-void taosFillSetDataBlockFromFilePage(SFillInfo* pFillInfo, const tFilePage** pInput) {
+void taosFillSetInputDataBlock(SFillInfo* pFillInfo, const SSDataBlock* pInput) {
   for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
-    memcpy(pFillInfo->pData[i], pInput[i]->data, pFillInfo->numOfRows * pFillInfo->pFillCol[i].col.bytes);
+    SColumnInfoData* pColData = taosArrayGet(pInput->pDataBlock, i);
+//    pFillInfo->pData[i] = pColData->pData;
+    if (pInput->info.rows > pFillInfo->alloc) {
+      char* t = realloc(pFillInfo->pData[i], pColData->info.bytes * pInput->info.rows);
+      assert(t != NULL);
+
+      pFillInfo->pData[i] = t;
+      pFillInfo->alloc = pInput->info.rows;
+    }
+
+    memcpy(pFillInfo->pData[i], pColData->pData, pColData->info.bytes * pInput->info.rows);
   }
 }
 
@@ -427,12 +446,20 @@ void taosFillCopyInputDataFromOneFilePage(SFillInfo* pFillInfo, const tFilePage*
     SFillColInfo* pCol = &pFillInfo->pFillCol[i];
 
     const char* data = pInput->data + pCol->col.offset * pInput->num;
-    memcpy(pFillInfo->pData[i], data, (size_t)(pInput->num * pCol->col.bytes));
+    if (pInput->num > pFillInfo->alloc) {
+      char* t = realloc(pFillInfo->pData[i], (size_t)(pCol->col.bytes * pInput->num));
+      assert(t != NULL);
+
+      pFillInfo->pData[i] = t;
+      pFillInfo->alloc = (int32_t)pInput->num;
+    }
+
+    memcpy(pFillInfo->pData[i], data, (size_t)(pCol->col.bytes * pInput->num));
 
     if (TSDB_COL_IS_TAG(pCol->flag)) {  // copy the tag value to tag value buffer
       SFillTagColInfo* pTag = &pFillInfo->pTags[pCol->tagIndex];
       assert (pTag->col.colId == pCol->col.colId);
-      memcpy(pTag->tagVal, data, pCol->col.bytes);
+      memcpy(pTag->tagVal, data, pCol->col.bytes);  // TODO not memcpy??
     }
   }
 }
@@ -490,7 +517,7 @@ int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint*
   return TSDB_CODE_SUCCESS;
 }
 
-int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, tFilePage** output, int32_t capacity) {
+int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, void** output, int32_t capacity) {
   int32_t remain = taosNumOfRemainRows(pFillInfo);
 
   int64_t numOfRes = getNumOfResultsAfterFillGap(pFillInfo, pFillInfo->end, capacity);

src/query/src/qParserImpl.c

@@ -135,28 +135,33 @@ tSQLExpr *tSqlExprIdValueCreate(SStrToken *pToken, int32_t optrType) {
   if (optrType == TK_INTEGER || optrType == TK_STRING || optrType == TK_FLOAT || optrType == TK_BOOL) {
     toTSDBType(pToken->type);
 
-    tVariantCreate(&pSqlExpr->val, pToken);
-    pSqlExpr->nSQLOptr = optrType;
+    tVariantCreate(&pSqlExpr->value, pToken);
+    pSqlExpr->tokenId = optrType;
+    pSqlExpr->type    = SQL_NODE_VALUE;
   } else if (optrType == TK_NOW) {
     // use microsecond by default
-    pSqlExpr->val.i64 = taosGetTimestamp(TSDB_TIME_PRECISION_MICRO);
-    pSqlExpr->val.nType = TSDB_DATA_TYPE_BIGINT;
-    pSqlExpr->nSQLOptr = TK_TIMESTAMP;  // TK_TIMESTAMP used to denote the time value is in microsecond
+    pSqlExpr->value.i64 = taosGetTimestamp(TSDB_TIME_PRECISION_MICRO);
+    pSqlExpr->value.nType = TSDB_DATA_TYPE_BIGINT;
+    pSqlExpr->tokenId = TK_TIMESTAMP;  // TK_TIMESTAMP used to denote the time value is in microsecond
+    pSqlExpr->type    = SQL_NODE_VALUE;
   } else if (optrType == TK_VARIABLE) {
-    int32_t ret = parseAbsoluteDuration(pToken->z, pToken->n, &pSqlExpr->val.i64);
+    int32_t ret = parseAbsoluteDuration(pToken->z, pToken->n, &pSqlExpr->value.i64);
     if (ret != TSDB_CODE_SUCCESS) {
       terrno = TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
     }
 
-    pSqlExpr->val.nType = TSDB_DATA_TYPE_BIGINT;
-    pSqlExpr->nSQLOptr = TK_TIMESTAMP;
-  } else {  // it must be the column name (tk_id) if it is not the number
+    pSqlExpr->value.nType = TSDB_DATA_TYPE_BIGINT;
+    pSqlExpr->tokenId = TK_TIMESTAMP;
+    pSqlExpr->type    = SQL_NODE_VALUE;
+  } else {
+    // Here it must be the column name (tk_id) if it is not a number or string.
     assert(optrType == TK_ID || optrType == TK_ALL);
     if (pToken != NULL) {
       pSqlExpr->colInfo = *pToken;
     }
 
-    pSqlExpr->nSQLOptr = optrType;
+    pSqlExpr->tokenId = optrType;
+    pSqlExpr->type    = SQL_NODE_TABLE_COLUMN;
   }
 
   return pSqlExpr;
@@ -167,19 +172,22 @@ tSQLExpr *tSqlExprIdValueCreate(SStrToken *pToken, int32_t optrType) {
  * function name is denoted by pFunctionToken
  */
 tSQLExpr *tSqlExprCreateFunction(tSQLExprList *pList, SStrToken *pFuncToken, SStrToken *endToken, int32_t optType) {
-  if (pFuncToken == NULL) return NULL;
+  if (pFuncToken == NULL) {
+    return NULL;
+  }
 
   tSQLExpr *pExpr = calloc(1, sizeof(tSQLExpr));
-  pExpr->nSQLOptr = optType;
-  pExpr->pParam = pList;
+  pExpr->tokenId = optType;
+  pExpr->type    = SQL_NODE_SQLFUNCTION;
+  pExpr->pParam  = pList;
 
   int32_t len = (int32_t)((endToken->z + endToken->n) - pFuncToken->z);
-  pExpr->operand.z = pFuncToken->z;
+  pExpr->operand = (*pFuncToken);
 
-  pExpr->operand.n = len;  // raw field name
-  pExpr->operand.type = pFuncToken->type;
+  pExpr->token.n = len;
+  pExpr->token.z = pFuncToken->z;
+  pExpr->token.type = pFuncToken->type;
 
-  pExpr->token = pExpr->operand;
   return pExpr;
 }
 
@@ -190,6 +198,7 @@ tSQLExpr *tSqlExprCreateFunction(tSQLExprList *pList, SStrToken *pFuncToken, SSt
 tSQLExpr *tSqlExprCreate(tSQLExpr *pLeft, tSQLExpr *pRight, int32_t optrType) {
   tSQLExpr *pExpr = calloc(1, sizeof(tSQLExpr));
 
+  pExpr->type = SQL_NODE_EXPR;
   if (pLeft != NULL && pRight != NULL && (optrType != TK_IN)) {
     char* endPos = pRight->token.z + pRight->token.n;
     pExpr->token.z = pLeft->token.z;
@@ -203,32 +212,33 @@ tSQLExpr *tSqlExprCreate(tSQLExpr *pLeft, tSQLExpr *pRight, int32_t optrType) {
      * if a token is noted as the TK_TIMESTAMP, the time precision is microsecond
      * Otherwise, the time precision is adaptive, determined by the time precision from databases.
      */
-    if ((pLeft->nSQLOptr == TK_INTEGER && pRight->nSQLOptr == TK_INTEGER) ||
-        (pLeft->nSQLOptr == TK_TIMESTAMP && pRight->nSQLOptr == TK_TIMESTAMP)) {
-      pExpr->val.nType = TSDB_DATA_TYPE_BIGINT;
-      pExpr->nSQLOptr = pLeft->nSQLOptr;
+    if ((pLeft->tokenId == TK_INTEGER && pRight->tokenId == TK_INTEGER) ||
+        (pLeft->tokenId == TK_TIMESTAMP && pRight->tokenId == TK_TIMESTAMP)) {
+      pExpr->value.nType = TSDB_DATA_TYPE_BIGINT;
+      pExpr->tokenId = pLeft->tokenId;
+      pExpr->type    = SQL_NODE_VALUE;
 
       switch (optrType) {
         case TK_PLUS: {
-          pExpr->val.i64 = pLeft->val.i64 + pRight->val.i64;
+          pExpr->value.i64 = pLeft->value.i64 + pRight->value.i64;
           break;
         }
         case TK_MINUS: {
-          pExpr->val.i64 = pLeft->val.i64 - pRight->val.i64;
+          pExpr->value.i64 = pLeft->value.i64 - pRight->value.i64;
           break;
         }
         case TK_STAR: {
-          pExpr->val.i64 = pLeft->val.i64 * pRight->val.i64;
+          pExpr->value.i64 = pLeft->value.i64 * pRight->value.i64;
           break;
         }
         case TK_DIVIDE: {
-          pExpr->nSQLOptr = TK_FLOAT;
-          pExpr->val.nType = TSDB_DATA_TYPE_DOUBLE;
-          pExpr->val.dKey = (double)pLeft->val.i64 / pRight->val.i64;
+          pExpr->tokenId = TK_FLOAT;
+          pExpr->value.nType = TSDB_DATA_TYPE_DOUBLE;
+          pExpr->value.dKey = (double)pLeft->value.i64 / pRight->value.i64;
           break;
         }
         case TK_REM: {
-          pExpr->val.i64 = pLeft->val.i64 % pRight->val.i64;
+          pExpr->value.i64 = pLeft->value.i64 % pRight->value.i64;
           break;
         }
       }
@@ -236,33 +246,35 @@ tSQLExpr *tSqlExprCreate(tSQLExpr *pLeft, tSQLExpr *pRight, int32_t optrType) {
       tSqlExprDestroy(pLeft);
       tSqlExprDestroy(pRight);
 
-    } else if ((pLeft->nSQLOptr == TK_FLOAT && pRight->nSQLOptr == TK_INTEGER) || (pLeft->nSQLOptr == TK_INTEGER && pRight->nSQLOptr == TK_FLOAT) ||
-        (pLeft->nSQLOptr == TK_FLOAT && pRight->nSQLOptr == TK_FLOAT)) {
-      pExpr->val.nType = TSDB_DATA_TYPE_DOUBLE;
-      pExpr->nSQLOptr  = TK_FLOAT;
+    } else if ((pLeft->tokenId == TK_FLOAT && pRight->tokenId == TK_INTEGER) ||
+               (pLeft->tokenId == TK_INTEGER && pRight->tokenId == TK_FLOAT) ||
+               (pLeft->tokenId == TK_FLOAT && pRight->tokenId == TK_FLOAT)) {
+      pExpr->value.nType = TSDB_DATA_TYPE_DOUBLE;
+      pExpr->tokenId  = TK_FLOAT;
+      pExpr->type     = SQL_NODE_VALUE;
 
-      double left  = (pLeft->val.nType == TSDB_DATA_TYPE_DOUBLE) ? pLeft->val.dKey : pLeft->val.i64;
-      double right = (pRight->val.nType == TSDB_DATA_TYPE_DOUBLE) ? pRight->val.dKey : pRight->val.i64;
+      double left  = (pLeft->value.nType == TSDB_DATA_TYPE_DOUBLE) ? pLeft->value.dKey : pLeft->value.i64;
+      double right = (pRight->value.nType == TSDB_DATA_TYPE_DOUBLE) ? pRight->value.dKey : pRight->value.i64;
 
       switch (optrType) {
         case TK_PLUS: {
-          pExpr->val.dKey = left + right;
+          pExpr->value.dKey = left + right;
           break;
         }
         case TK_MINUS: {
-          pExpr->val.dKey = left - right;
+          pExpr->value.dKey = left - right;
           break;
         }
         case TK_STAR: {
-          pExpr->val.dKey = left * right;
+          pExpr->value.dKey = left * right;
           break;
         }
         case TK_DIVIDE: {
-          pExpr->val.dKey = left / right;
+          pExpr->value.dKey = left / right;
           break;
         }
         case TK_REM: {
-          pExpr->val.dKey = left - ((int64_t)(left / right)) * right;
+          pExpr->value.dKey = left - ((int64_t)(left / right)) * right;
           break;
         }
       }
@@ -271,21 +283,21 @@ tSQLExpr *tSqlExprCreate(tSQLExpr *pLeft, tSQLExpr *pRight, int32_t optrType) {
       tSqlExprDestroy(pRight);
 
     } else {
-      pExpr->nSQLOptr = optrType;
+      pExpr->tokenId = optrType;
       pExpr->pLeft = pLeft;
       pExpr->pRight = pRight;
     }
   } else if (optrType == TK_IN) {
-    pExpr->nSQLOptr = optrType;
+    pExpr->tokenId = optrType;
     pExpr->pLeft = pLeft;
 
     tSQLExpr *pRSub = calloc(1, sizeof(tSQLExpr));
-    pRSub->nSQLOptr = TK_SET;  // TODO refactor .....
+    pRSub->tokenId = TK_SET;  // TODO refactor .....
     pRSub->pParam = (tSQLExprList *)pRight;
 
     pExpr->pRight = pRSub;
   } else {
-    pExpr->nSQLOptr = optrType;
+    pExpr->tokenId = optrType;
     pExpr->pLeft = pLeft;
 
     if (pLeft != NULL && pRight == NULL) {
@@ -325,8 +337,8 @@ void tSqlExprNodeDestroy(tSQLExpr *pExpr) {
     return;
   }
 
-  if (pExpr->nSQLOptr == TK_STRING) {
-    tVariantDestroy(&pExpr->val);
+  if (pExpr->tokenId == TK_STRING) {
+    tVariantDestroy(&pExpr->value);
   }
 
   tSqlExprListDestroy(pExpr->pParam);
@@ -538,8 +550,8 @@ void tSqlSetColumnType(TAOS_FIELD *pField, SStrToken *type) {
 /*
  * extract the select info out of sql string
  */
-SQuerySQL *tSetQuerySqlElems(SStrToken *pSelectToken, tSQLExprList *pSelection, SArray *pFrom, tSQLExpr *pWhere,
-                             SArray *pGroupby, SArray *pSortOrder, SIntervalVal *pInterval,
+SQuerySQL *tSetQuerySqlNode(SStrToken *pSelectToken, tSQLExprList *pSelection, SArray *pFrom, tSQLExpr *pWhere,
+                             SArray *pGroupby, SArray *pSortOrder, SIntervalVal *pInterval, SSessionWindowVal *pSession,
                              SStrToken *pSliding, SArray *pFill, SLimitVal *pLimit, SLimitVal *pGLimit) {
   assert(pSelection != NULL);
 
@@ -562,14 +574,17 @@ SQuerySQL *tSetQuerySqlElems(SStrToken *pSelectToken, tSQLExprList *pSelection,
   }
 
   if (pInterval != NULL) {
-    pQuery->interval = pInterval->interval;
-    pQuery->offset = pInterval->offset;
+    pQuery->interval = *pInterval;
   }
 
   if (pSliding != NULL) {
     pQuery->sliding = *pSliding;
   }
 
+  if (pSession != NULL) {
+    pQuery->sessionVal = *pSession;
+  }
+
   pQuery->fillType = pFill;
   return pQuery;
 }

src/query/src/qPercentile.c

@@ -254,7 +254,7 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
 
   resetSlotInfo(pBucket);
 
-  int32_t ret = createDiskbasedResultBuffer(&pBucket->pBuffer, pBucket->bytes, pBucket->bufPageSize, pBucket->bufPageSize * 512, NULL);
+  int32_t ret = createDiskbasedResultBuffer(&pBucket->pBuffer, pBucket->bufPageSize, pBucket->bufPageSize * 512, NULL);
   if (ret != TSDB_CODE_SUCCESS) {
     tMemBucketDestroy(pBucket);
     return NULL;

src/query/src/qResultbuf.c

@@ -9,8 +9,7 @@
 #define GET_DATA_PAYLOAD(_p) ((char *)(_p)->pData + POINTER_BYTES)
 #define NO_IN_MEM_AVAILABLE_PAGES(_b) (listNEles((_b)->lruList) >= (_b)->inMemPages)
 
-int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t rowSize, int32_t pagesize,
-                                    int32_t inMemBufSize, const void* handle) {
+int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pagesize, int32_t inMemBufSize, const void* handle) {
   *pResultBuf = calloc(1, sizeof(SDiskbasedResultBuf));
 
   SDiskbasedResultBuf* pResBuf = *pResultBuf;
@@ -31,7 +30,6 @@ int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t ro
   // at least more than 2 pages must be in memory
   assert(inMemBufSize >= pagesize * 2);
 
-  pResBuf->numOfRowsPerPage = (pagesize - sizeof(tFilePage)) / rowSize;
   pResBuf->lruList = tdListNew(POINTER_BYTES);
 
   // init id hash table
@@ -387,8 +385,6 @@ void releaseResBufPageInfo(SDiskbasedResultBuf* pResultBuf, SPageInfo* pi) {
   pResultBuf->statis.releasePages += 1;
 }
 
-size_t getNumOfRowsPerPage(const SDiskbasedResultBuf* pResultBuf) { return pResultBuf->numOfRowsPerPage; }
-
 size_t getNumOfResultBufGroupId(const SDiskbasedResultBuf* pResultBuf) { return taosHashGetSize(pResultBuf->groupSet); }
 
 size_t getResBufSize(const SDiskbasedResultBuf* pResultBuf) { return (size_t)pResultBuf->totalBufSize; }

src/vnode/src/vnodeRead.c

@@ -442,4 +442,4 @@ void vnodeWaitReadCompleted(SVnodeObj *pVnode) {
     vTrace("vgId:%d, queued rmsg num:%d", pVnode->vgId, pVnode->queuedRMsg);
     taosMsleep(10);
   }
-}
\ No newline at end of file
+}