add unique logic

src/client/src/tscGlobalmerge.c

@@ -447,7 +447,7 @@ int32_t tscCreateGlobalMergerEnv(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBu
     }
     capacity = (*nBufferSizes) / rlen;
   }
-  
+
   pModel = createColumnModel(pSchema, (int32_t)size, capacity);
   tfree(pSchema);
   if (pModel == NULL){
@@ -456,7 +456,7 @@ int32_t tscCreateGlobalMergerEnv(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBu
 
   int32_t pg = DEFAULT_PAGE_SIZE;
   int32_t overhead = sizeof(tFilePage);
-  while((pg - overhead) < pModel->rowSize * 2) {
+  while((pg - overhead) < rlen * 2) {
     pg *= 2;
   }
 
@@ -656,7 +656,7 @@ static void doExecuteFinalMerge(SOperatorInfo* pOperator, int32_t numOfExpr, SSD
         for(int32_t j = 0; j < numOfExpr; ++j) {
           pCtx[j].pOutput += (pCtx[j].outputBytes * numOfRows);
           if (pCtx[j].functionId == TSDB_FUNC_TOP || pCtx[j].functionId == TSDB_FUNC_BOTTOM ||
-              pCtx[j].functionId == TSDB_FUNC_SAMPLE) {
+              pCtx[j].functionId == TSDB_FUNC_SAMPLE || pCtx[j].functionId == TSDB_FUNC_UNIQUE) {
             if(j > 0) pCtx[j].ptsOutputBuf = pCtx[j - 1].pOutput;
           }
         }

src/client/src/tscServer.c

@@ -1045,8 +1045,8 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
   
   SGroupbyExpr *pGroupbyExpr = query.pGroupbyExpr;
   if (pGroupbyExpr != NULL && pGroupbyExpr->numOfGroupCols > 0) {
-    pQueryMsg->orderByIdx = htons(pGroupbyExpr->orderIndex);
-    pQueryMsg->orderType = htons(pGroupbyExpr->orderType);
+    //pQueryMsg->orderByIdx = htons(pGroupbyExpr->orderIndex);
+    pQueryMsg->groupOrderType = htons(pGroupbyExpr->orderType);
 
     for (int32_t j = 0; j < pGroupbyExpr->numOfGroupCols; ++j) {
       SColIndex* pCol = taosArrayGet(pGroupbyExpr->columnInfo, j);

src/client/src/tscUtil.c

@@ -74,11 +74,11 @@ int32_t converToStr(char *str, int type, void *buf, int32_t bufSize, int32_t *le
       break;
 
     case TSDB_DATA_TYPE_UINT:
-      n = sprintf(str, "%d", *(uint32_t*)buf);
+      n = sprintf(str, "%u", *(uint32_t*)buf);
       break;
 
     case TSDB_DATA_TYPE_UBIGINT:
-      n = sprintf(str, "%" PRId64, *(uint64_t*)buf);
+      n = sprintf(str, "%" PRIu64, *(uint64_t*)buf);
       break;
 
     case TSDB_DATA_TYPE_FLOAT:
@@ -304,7 +304,7 @@ bool tscNonOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableI
     return false;
   }
 
-  // order by columnIndex exists, not a non-ordered projection query
+  // order by columnIndex not exists, not a ordered projection query
   return pQueryInfo->order.orderColId < 0;
 }
 
@@ -313,7 +313,7 @@ bool tscOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableInde
     return false;
   }
 
-  // order by columnIndex exists, a non-ordered projection query
+  // order by columnIndex exists, a ordered projection query
   return pQueryInfo->order.orderColId >= 0;
 }
 
@@ -689,7 +689,8 @@ bool isSimpleAggregateRv(SQueryInfo* pQueryInfo) {
         (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM ||
          functionId == TSDB_FUNC_TS_COMP ||
          functionId == TSDB_FUNC_SAMPLE ||
-         functionId == TSDB_FUNC_HISTOGRAM)) {
+         functionId == TSDB_FUNC_HISTOGRAM ||
+         functionId == TSDB_FUNC_UNIQUE)) {
       return true;
     }
   }
@@ -5403,7 +5404,7 @@ int parseJsontoTagData(char* json, SKVRowBuilder* kvRowBuilder, char* errMsg, in
   // set json real data
   cJSON *root = cJSON_Parse(json);
   if (root == NULL){
-    tscError("json parse error");
+    tscError("json parse error:%s", json);
     return tscSQLSyntaxErrMsg(errMsg, "json parse error", NULL);
   }
 

src/inc/taoserror.h

@@ -293,6 +293,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_QRY_SYS_ERROR                 TAOS_DEF_ERROR_CODE(0, 0x070D)  //"System error")
 #define TSDB_CODE_QRY_INVALID_TIME_CONDITION    TAOS_DEF_ERROR_CODE(0, 0x070E)  //"invalid time condition")
 #define TSDB_CODE_QRY_INVALID_SCHEMA_VERSION    TAOS_DEF_ERROR_CODE(0, 0x0710)  //"invalid schema version")
+#define TSDB_CODE_QRY_UNIQUE_RESULT_TOO_LARGE   TAOS_DEF_ERROR_CODE(0, 0x0711)  //"unique result num is too large")
 
 // grant
 #define TSDB_CODE_GRANT_EXPIRED                 TAOS_DEF_ERROR_CODE(0, 0x0800)  //"License expired"

src/inc/taosmsg.h

@@ -503,8 +503,8 @@ typedef struct {
   uint32_t    tagCondLen;       // tag length in current query
   int32_t    colCondLen;       // column length in current query
   int16_t     numOfGroupCols;   // num of group by columns
-  int16_t     orderByIdx;
-  int16_t     orderType;        // used in group by xx order by xxx
+  int16_t     orderByIdx;       // useless
+  int16_t     groupOrderType;   // used for group order
   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;

src/query/inc/qAggMain.h

@@ -78,8 +78,9 @@ extern "C" {
 
 #define TSDB_FUNC_ELAPSED      37
 #define TSDB_FUNC_HISTOGRAM    38
+#define TSDB_FUNC_UNIQUE       39
 
-#define TSDB_FUNC_MAX_NUM    39
+#define TSDB_FUNC_MAX_NUM    40
 
 #define TSDB_FUNCSTATE_SO           0x1u    // single output
 #define TSDB_FUNCSTATE_MO           0x2u    // dynamic number of output, not multinumber of output e.g., TOP/BOTTOM
@@ -200,6 +201,8 @@ typedef struct SQLFunctionCtx {
   SExtTagsInfo tagInfo;
   SPoint1      start;
   SPoint1      end;
+
+  int32_t      maxUniqueResult;
 } SQLFunctionCtx;
 
 typedef struct SAggFunctionInfo {
@@ -249,7 +252,7 @@ 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[40];
+extern struct SAggFunctionInfo aAggs[TSDB_FUNC_MAX_NUM];
 
 extern int32_t functionCompatList[]; // compatible check array list
 

src/query/inc/qExecutor.h

@@ -221,6 +221,7 @@ typedef struct SQueryAttr {
 
   bool             stableQuery;      // super table query or not
   bool             topBotQuery;      // TODO used bitwise flag
+  bool             uniqueQuery;
   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
@@ -281,6 +282,7 @@ typedef struct SQueryAttr {
   STableGroupInfo  tableGroupInfo;       // table <tid, last_key> list  SArray<STableKeyInfo>
   int32_t          vgId;
   SArray          *pUdfInfo;             // no need to free
+  int32_t          maxUniqueResult;
 } SQueryAttr;
 
 typedef SSDataBlock* (*__operator_fn_t)(void* param, bool* newgroup);
@@ -730,4 +732,5 @@ void addTableReadRows(SQueryRuntimeEnv* pEnv, int32_t tid, int32_t rows);
 // tsdb scan table callback table or query is over. param is SQueryRuntimeEnv*
 bool qReadOverCB(void* param, int8_t type, int32_t tid);
 
+bool isUniqueQuery(int32_t numOfOutput, SExprInfo* pExprs);
 #endif  // TDENGINE_QEXECUTOR_H

src/query/inc/qExtbuffer.h

@@ -53,14 +53,14 @@ typedef struct tFlushoutInfo {
 } tFlushoutInfo;
 
 typedef struct tFlushoutData {
-  uint32_t       nAllocSize;
-  uint32_t       nLength;
-  tFlushoutInfo *pFlushoutInfo;
+  uint32_t       nAllocSize;    // capacity
+  uint32_t       nLength;       // size
+  tFlushoutInfo *pFlushoutInfo; // dynamic allocate
 } tFlushoutData;
 
 typedef struct SExtFileInfo {
-  uint32_t      nFileSize;  // in pages
-  uint32_t      pageSize;
+  uint32_t      nFileSize;  // how many pages in file
+  //uint32_t      pageSize; // useless
   uint32_t      numOfElemsInFile;
   tFlushoutData flushoutData;
 } SExtFileInfo;

src/query/inc/qResultbuf.h

@@ -78,7 +78,7 @@ typedef struct SDiskbasedResultBuf {
 
 #define DEFAULT_INTERN_BUF_PAGE_SIZE  (1024L)                          // in bytes
 #define PAGE_INFO_INITIALIZER         (SPageDiskInfo){-1, -1}
-
+#define MAX_UNIQUE_RESULT_SIZE (1000000)
 /**
  * create disk-based result buffer
  * @param pResultBuf

src/query/inc/qTableMeta.h

@@ -50,7 +50,7 @@ typedef struct SGroupbyExpr {
   int16_t tableIndex;
   SArray* columnInfo;  // SArray<SColIndex>, group by columns information
   int16_t numOfGroupCols;  // todo remove it
-  int16_t orderIndex;  // order by column index
+  //int16_t orderIndex; // order by column index, rm useless orderIndex
   int16_t orderType;   // order by type: asc/desc
 } SGroupbyExpr;
 

src/query/src/qAggMain.c

@@ -223,6 +223,21 @@ typedef struct{
   SHistogramFuncBin* orderedBins;
 } SHistogramFuncInfo;
 
+typedef struct {
+  int64_t  timestamp;
+  char *   pTags;
+} UniqueUnit;
+
+typedef struct {
+  SHashObj     *pSet;
+  int32_t      num;
+  char         res[];
+} SUniqueFuncInfo;
+
+void freeUniqueUnit(void* unit){
+  tfree(((UniqueUnit *)unit)->pTags);
+}
+
 int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, int16_t *type,
                           int32_t *bytes, int32_t *interBytes, int16_t extLength, bool isSuperTable, SUdfInfo* pUdfInfo) {
   if (!isValidDataType(dataType)) {
@@ -353,6 +368,12 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI
       *bytes = (sizeof(STopBotInfo) + (sizeof(tValuePair) + POINTER_BYTES + extLength) * param);
       *interBytes = *bytes;
 
+      return TSDB_CODE_SUCCESS;
+    } else if (functionId == TSDB_FUNC_UNIQUE) {
+      *type = TSDB_DATA_TYPE_BINARY;
+      *bytes = (sizeof(SUniqueFuncInfo) + (sizeof(tValuePair) + POINTER_BYTES + extLength) * param);
+      *interBytes = *bytes;
+
       return TSDB_CODE_SUCCESS;
     } else if (functionId == TSDB_FUNC_SAMPLE) {
       *type = TSDB_DATA_TYPE_BINARY;
@@ -479,6 +500,14 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI
 
     size_t size = sizeof(STopBotInfo) + (sizeof(tValuePair) + POINTER_BYTES + extLength) * param;
 
+    // the output column may be larger than sizeof(STopBotInfo)
+    *interBytes = (int32_t)size;
+  } else if (functionId == TSDB_FUNC_UNIQUE) {
+    *type = (int16_t)dataType;
+    *bytes = dataBytes;
+
+    size_t size = sizeof(SUniqueFuncInfo) + (sizeof(tValuePair) + POINTER_BYTES + extLength) * param;
+
     // the output column may be larger than sizeof(STopBotInfo)
     *interBytes = (int32_t)size;
   } else if (functionId == TSDB_FUNC_SAMPLE) {
@@ -5097,6 +5126,206 @@ static void histogram_func_finalizer(SQLFunctionCtx *pCtx) {
   doFinalizer(pCtx);
 }
 
+// unique use the intermediate result buffer to keep the intermediate result
+static SUniqueFuncInfo *getUniqueOutputInfo(SQLFunctionCtx *pCtx) {
+  SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
+
+  // only the first_stage_merge is directly written data into final output buffer
+  if (pCtx->stableQuery && pCtx->currentStage != MERGE_STAGE) {
+    return (SUniqueFuncInfo*) pCtx->pOutput;
+  } else { // during normal table query and super table at the secondary_stage, result is written to intermediate buffer
+    return GET_ROWCELL_INTERBUF(pResInfo);
+  }
+}
+
+// unique
+static void copyUniqueRes(SQLFunctionCtx *pCtx, int32_t type) {
+  SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
+  SUniqueFuncInfo *pRes = GET_ROWCELL_INTERBUF(pResInfo);
+
+  size_t size = sizeof(tValuePair) + pCtx->tagInfo.tagsLen;
+  char *tvp = pRes->res;
+
+  int32_t step = QUERY_ASC_FORWARD_STEP;
+  int32_t len = (int32_t)(GET_RES_INFO(pCtx)->numOfRes);
+
+  switch (type) {
+    case TSDB_DATA_TYPE_UINT:
+    case TSDB_DATA_TYPE_INT: {
+      int32_t *output = (int32_t *)pCtx->pOutput;
+      for (int32_t i = 0; i < len; ++i, output += step) {
+        *output = ((tValuePair *)tvp)->v.i64;
+        tvp += size;
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_UBIGINT:
+    case TSDB_DATA_TYPE_BIGINT: {
+      int64_t *output = (int64_t *)pCtx->pOutput;
+      for (int32_t i = 0; i < len; ++i, output += step) {
+        *output = ((tValuePair *)tvp)->v.i64;
+        tvp += size;
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_DOUBLE: {
+      double *output = (double *)pCtx->pOutput;
+      for (int32_t i = 0; i < len; ++i, output += step) {
+        *output = ((tValuePair *)tvp)->v.dKey;
+        tvp += size;
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_FLOAT: {
+      float *output = (float *)pCtx->pOutput;
+      for (int32_t i = 0; i < len; ++i, output += step) {
+        *output = ((tValuePair *)tvp)->v.dKey;
+        tvp += size;
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_USMALLINT:
+    case TSDB_DATA_TYPE_SMALLINT: {
+      int16_t *output = (int16_t *)pCtx->pOutput;
+      for (int32_t i = 0; i < len; ++i, output += step) {
+        *output = ((tValuePair *)tvp)->v.i64;
+        tvp += size;
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_UTINYINT:
+    case TSDB_DATA_TYPE_TINYINT: {
+      int8_t *output = (int8_t *)pCtx->pOutput;
+      for (int32_t i = 0; i < len; ++i, output += step) {
+        *output = ((tValuePair *)tvp)->v.i64;
+        tvp += size;
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_BINARY:
+    {
+
+    }
+    case TSDB_DATA_TYPE_NCHAR:
+    {
+
+    }
+    default: {
+      qError("top/bottom function not support data type:%d", pCtx->inputType);
+      return;
+    }
+  }
+
+  // set the output timestamp of each record.
+  TSKEY *output = pCtx->ptsOutputBuf;
+  for (int32_t i = 0; i < len; ++i, output += step) {
+    *output = ((tValuePair *)tvp)->timestamp;
+    tvp += size;
+  }
+
+  // set the corresponding tag data for each record
+  // todo check malloc failure
+  if (pCtx->tagInfo.numOfTagCols == 0) {
+    return ;
+  }
+
+  char **pData = calloc(pCtx->tagInfo.numOfTagCols, POINTER_BYTES);
+  for (int32_t i = 0; i < pCtx->tagInfo.numOfTagCols; ++i) {
+    pData[i] = pCtx->tagInfo.pTagCtxList[i]->pOutput;
+  }
+
+  for (int32_t i = 0; i < len; ++i, output += step) {
+    int16_t offset = 0;
+    for (int32_t j = 0; j < pCtx->tagInfo.numOfTagCols; ++j) {
+      memcpy(pData[j], ((tValuePair *)tvp)->pTags + offset, (size_t)pCtx->tagInfo.pTagCtxList[j]->outputBytes);
+      offset += pCtx->tagInfo.pTagCtxList[j]->outputBytes;
+      pData[j] += pCtx->tagInfo.pTagCtxList[j]->outputBytes;
+      tvp += size;
+    }
+  }
+
+  tfree(pData);
+}
+
+static bool unique_function_setup(SQLFunctionCtx *pCtx, SResultRowCellInfo* pResInfo) {
+  if (!function_setup(pCtx, pResInfo)) {
+    return false;
+  }
+
+  SUniqueFuncInfo *uniqueInfo = getUniqueOutputInfo(pCtx);
+  uniqueInfo->pSet = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+  taosHashSetFreeFp(uniqueInfo->pSet, freeUniqueUnit);
+
+  return true;
+}
+
+static void unique_function(SQLFunctionCtx *pCtx) {
+  SUniqueFuncInfo *pInfo = getUniqueOutputInfo(pCtx);
+
+  for (int32_t i = 0; i < pCtx->size; i++) {
+    char *pData = GET_INPUT_DATA(pCtx, i);
+    TSKEY k = 0;
+    if (pCtx->ptsList != NULL) {
+      k = GET_TS_DATA(pCtx, i);
+    }
+    tValuePair *unique = taosHashGet(pInfo->pSet, pData, pCtx->inputBytes);
+    if (unique == NULL) {
+      size_t size = sizeof(tValuePair) + pCtx->tagInfo.tagsLen;
+      tValuePair *tmp = (tValuePair *)(pInfo->res + pInfo->num * size);
+      if (pCtx->inputType == TSDB_DATA_TYPE_BINARY || pCtx->inputType == TSDB_DATA_TYPE_NCHAR) {
+        tVariantCreateFromBinary(&tmp->v, varDataVal(pData), varDataLen(pData), pCtx->inputType);
+      }else{
+        tVariantCreateFromBinary(&tmp->v, pData, 0, pCtx->inputType);
+      }
+      tmp->timestamp = k;
+
+      int32_t offset = 0;
+      for (int32_t j = 0; j < pCtx->tagInfo.numOfTagCols; ++j) {
+        SQLFunctionCtx *tagCtx = pCtx->tagInfo.pTagCtxList[j];
+        if (tagCtx->functionId == TSDB_FUNC_TAG_DUMMY) {
+          aAggs[TSDB_FUNC_TAG].xFunction(tagCtx);
+          memcpy(tmp->pTags + offset, tagCtx->pOutput, tagCtx->outputBytes);
+          offset += tagCtx->outputBytes;
+        }
+      }
+
+      taosHashPut(pInfo->pSet, pData, pCtx->inputBytes, &tmp, sizeof(tValuePair*));
+      pInfo->num++;
+    }else if(unique->timestamp > k){
+      unique->timestamp = k;
+    }
+  }
+
+}
+
+static void unique_function_merge(SQLFunctionCtx *pCtx) {
+  //SUniqueFuncInfo *pInput = (SUniqueFuncInfo *)GET_INPUT_DATA_LIST(pCtx);
+  //SUniqueFuncInfo *pOutput = getUniqueOutputInfo(pCtx);
+  // the intermediate result is binary, we only use the output data type
+//  for (int32_t i = 0; i < pInput->num; ++i) {
+//    int16_t type = (pCtx->outputType == TSDB_DATA_TYPE_FLOAT)? TSDB_DATA_TYPE_DOUBLE:pCtx->outputType;
+//    do_top_function_add(pOutput, (int32_t)pCtx->param[0].i64, &pInput->res[i]->v.i64, pInput->res[i]->timestamp,
+//                        type, &pCtx->tagInfo, pInput->res[i]->pTags, pCtx->currentStage);
+//  }
+//
+//  SET_VAL(pCtx, pInput->num, pOutput->num);
+//
+//  if (pOutput->num > 0) {
+//    SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
+//    pResInfo->hasResult = DATA_SET_FLAG;
+//  }
+}
+
+static void unique_func_finalizer(SQLFunctionCtx *pCtx) {
+  SUniqueFuncInfo *pInfo = getUniqueOutputInfo(pCtx);
+
+  GET_RES_INFO(pCtx)->numOfRes = pInfo->num;
+  GET_TRUE_DATA_TYPE();
+  copyUniqueRes(pCtx, type);
+  doFinalizer(pCtx);
+}
+
+
 /////////////////////////////////////////////////////////////////////////////////////////////
 /*
  * function compatible list.
@@ -5117,11 +5346,11 @@ int32_t functionCompatList[] = {
     1,          1,        1,         1,       -1,      1,          1,           1,          5,          1,      1,
     // tid_tag, deriv,    csum,       mavg,        sample,
     6,          8,        -1,         -1,          -1,
-    // block_info,elapsed,histogram
-    7,          1,        -1
+    // block_info,elapsed,histogram,unique
+    7,          1,        -1,        -1
 };
 
-SAggFunctionInfo aAggs[40] = {{
+SAggFunctionInfo aAggs[TSDB_FUNC_MAX_NUM] = {{
                               // 0, count function does not invoke the finalize function
                               "count",
                               TSDB_FUNC_COUNT,
@@ -5591,5 +5820,17 @@ SAggFunctionInfo aAggs[40] = {{
                               histogram_func_finalizer,
                               histogram_func_merge,
                               dataBlockRequired,
+                          },
+                          {
+                              // 39
+                              "unique",
+                              TSDB_FUNC_UNIQUE,
+                              TSDB_FUNC_INVALID_ID,
+                              TSDB_FUNCSTATE_MO | TSDB_FUNCSTATE_STABLE | TSDB_FUNCSTATE_SELECTIVITY,
+                              unique_function_setup,
+                              unique_function,
+                              unique_func_finalizer,
+                              unique_function_merge,
+                              dataBlockRequired,
                           }
 };

src/query/src/qExecutor.c

@@ -289,9 +289,8 @@ static int compareRowData(const void *a, const void *b, const void *userData) {
 }
 
 static void sortGroupResByOrderList(SGroupResInfo *pGroupResInfo, SQueryRuntimeEnv *pRuntimeEnv, SSDataBlock* pDataBlock, SQLFunctionCtx *pCtx) {
-  SArray *columnOrderList = getOrderCheckColumns(pRuntimeEnv->pQueryAttr);
-  size_t size = taosArrayGetSize(columnOrderList);
-  taosArrayDestroy(&columnOrderList);
+  int32_t size = pRuntimeEnv->pQueryAttr->pGroupbyExpr == NULL? 0: pRuntimeEnv->pQueryAttr->pGroupbyExpr->numOfGroupCols;
+  if (pRuntimeEnv->pQueryAttr->interval.interval > 0) size++;
 
   if (size <= 0) {
     return;
@@ -1004,6 +1003,13 @@ static void doApplyFunctions(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx
       }
     }
 
+    if (functionId == TSDB_FUNC_UNIQUE && GET_RES_INFO(&(pCtx[k]))->numOfRes > pQueryAttr->maxUniqueResult){
+      qError("Unique result num is too large. num: %d, limit: %d",
+             GET_RES_INFO(&(pCtx[k]))->numOfRes, pQueryAttr->maxUniqueResult);
+      aAggs[functionId].xFinalize(&pCtx[k]);
+      longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_UNIQUE_RESULT_TOO_LARGE);
+    }
+
     // restore it
     pCtx[k].preAggVals.isSet = hasAggregates;
     pCtx[k].pInput = start;
@@ -1263,6 +1269,14 @@ static void doAggregateImpl(SOperatorInfo* pOperator, TSKEY startTs, SQLFunction
       } else {
         assert(0);
       }
+
+      SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
+      if (functionId == TSDB_FUNC_UNIQUE && GET_RES_INFO(&(pCtx[k]))->numOfRes > pQueryAttr->maxUniqueResult){
+        qError("Unique result num is too large. num: %d, limit: %d",
+               GET_RES_INFO(&(pCtx[k]))->numOfRes, pQueryAttr->maxUniqueResult);
+        aAggs[functionId].xFinalize(&pCtx[k]);
+        longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_UNIQUE_RESULT_TOO_LARGE);
+      }
     }
   }
 }
@@ -1893,7 +1907,7 @@ static int32_t setCtxTagColumnInfo(SQLFunctionCtx *pCtx, int32_t numOfOutput) {
       continue;
     }
 
-    if (functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) {
+    if (functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) { //ts_select ts,top(col,2)
       tagLen += pCtx[i].outputBytes;
       pTagCtx[num++] = &pCtx[i];
     } else if ((aAggs[functionId].status & TSDB_FUNCSTATE_SELECTIVITY) != 0) {
@@ -1962,6 +1976,9 @@ static SQLFunctionCtx* createSQLFunctionCtx(SQueryRuntimeEnv* pRuntimeEnv, SExpr
     pCtx->end.key      = INT64_MIN;
     pCtx->startTs      = INT64_MIN;
 
+    if (pCtx->functionId == TSDB_FUNC_UNIQUE){
+      pCtx->maxUniqueResult = pQueryAttr->maxUniqueResult;
+    }
     pCtx->numOfParams  = pSqlExpr->numOfParams;
     for (int32_t j = 0; j < pCtx->numOfParams; ++j) {
       int16_t type = pSqlExpr->param[j].nType;
@@ -2754,7 +2771,15 @@ static void getIntermediateBufInfo(SQueryRuntimeEnv* pRuntimeEnv, int32_t* ps, i
   SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
   int32_t MIN_ROWS_PER_PAGE = 4;
 
-  *rowsize = (int32_t)(pQueryAttr->resultRowSize * getRowNumForMultioutput(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery));
+  if (pQueryAttr->uniqueQuery) {
+    int64_t rowSize = pQueryAttr->resultRowSize;
+    while(rowSize*pQueryAttr->maxUniqueResult > 1024*1024*100){
+      pQueryAttr->maxUniqueResult = pQueryAttr->maxUniqueResult >> 1u;
+    }
+    *rowsize = (int32_t)(rowSize*pQueryAttr->maxUniqueResult);
+  }else{
+    *rowsize = (int32_t)(pQueryAttr->resultRowSize * getRowNumForMultioutput(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery));
+  }
   int32_t overhead = sizeof(tFilePage);
 
   // one page contains at least two rows
@@ -3177,7 +3202,7 @@ int32_t loadDataBlockOnDemand(SQueryRuntimeEnv* pRuntimeEnv, STableScanInfo* pTa
                                   pTableScanInfo->rowCellInfoOffset) != TSDB_CODE_SUCCESS) {
         longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
-    } else if (pQueryAttr->stableQuery && (!pQueryAttr->tsCompQuery) && (!pQueryAttr->diffQuery) && (!pQueryAttr->pointInterpQuery)) { // stable aggregate, not interval aggregate or normal column aggregate
+    } else if (pQueryAttr->stableQuery && (!pQueryAttr->tsCompQuery) && (!pQueryAttr->diffQuery) && (!pQueryAttr->pointInterpQuery) && (!pQueryAttr->uniqueQuery)) { // stable aggregate, not interval aggregate or normal column aggregate
       doSetTableGroupOutputBuf(pRuntimeEnv, pTableScanInfo->pResultRowInfo, pTableScanInfo->pCtx,
                                pTableScanInfo->rowCellInfoOffset, pTableScanInfo->numOfOutput,
                                pRuntimeEnv->current->groupIndex);
@@ -3671,7 +3696,7 @@ void setDefaultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SOptrBasicInfo *pInfo, i
     // set the timestamp output buffer for top/bottom/diff query
     int32_t fid = pCtx[i].functionId;
     if (fid == TSDB_FUNC_TOP || fid == TSDB_FUNC_BOTTOM || fid == TSDB_FUNC_DIFF || fid == TSDB_FUNC_DERIVATIVE ||
-        fid == TSDB_FUNC_SAMPLE || fid == TSDB_FUNC_MAVG || fid == TSDB_FUNC_CSUM) {
+        fid == TSDB_FUNC_SAMPLE || fid == TSDB_FUNC_MAVG || fid == TSDB_FUNC_CSUM || fid == TSDB_FUNC_UNIQUE) {
       if (i > 0) pCtx[i].ptsOutputBuf = pCtx[i-1].pOutput;
     } else if (fid == TSDB_FUNC_INTERP) {
       assert(pCtx[0].functionId == TSDB_FUNC_TS_DUMMY || pCtx[0].functionId == TSDB_FUNC_TS);
@@ -3742,7 +3767,7 @@ void updateOutputBuf(SOptrBasicInfo* pBInfo, int32_t *bufCapacity, int32_t numOf
     if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM ||
         functionId == TSDB_FUNC_DIFF || functionId == TSDB_FUNC_DERIVATIVE ||
         functionId == TSDB_FUNC_CSUM || functionId == TSDB_FUNC_MAVG ||
-        functionId == TSDB_FUNC_SAMPLE ) {
+        functionId == TSDB_FUNC_SAMPLE || functionId == TSDB_FUNC_UNIQUE) {
       if (i > 0) pBInfo->pCtx[i].ptsOutputBuf = pBInfo->pCtx[i-1].pOutput;
     } else if (functionId == TSDB_FUNC_INTERP) {
       assert(pBInfo->pCtx[0].functionId == TSDB_FUNC_TS_DUMMY || pBInfo->pCtx[0].functionId == TSDB_FUNC_TS);
@@ -3918,6 +3943,15 @@ void finalizeQueryResult(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SResult
   }
 }
 
+bool isUniqueQuery(int32_t numOfOutput, SExprInfo* pExprs) {
+  for (int32_t i = 0; i < numOfOutput; ++i) {
+    if (pExprs[i].base.functionId == TSDB_FUNC_UNIQUE) {
+      return true;
+    }
+  }
+  return false;
+}
+
 static bool hasMainOutput(SQueryAttr *pQueryAttr) {
   for (int32_t i = 0; i < pQueryAttr->numOfOutput; ++i) {
     int32_t functionId = pQueryAttr->pExpr1[i].base.functionId;
@@ -4002,7 +4036,8 @@ void setResultRowOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pRe
     int32_t functionId = pCtx[i].functionId;
 
     if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM || functionId == TSDB_FUNC_DIFF ||
-        functionId == TSDB_FUNC_CSUM || functionId == TSDB_FUNC_MAVG || functionId == TSDB_FUNC_SAMPLE) {
+        functionId == TSDB_FUNC_CSUM || functionId == TSDB_FUNC_MAVG || functionId == TSDB_FUNC_SAMPLE ||
+        functionId == TSDB_FUNC_UNIQUE) {
       if(i > 0) pCtx[i].ptsOutputBuf = pCtx[i-1].pOutput;
     }
 
@@ -4071,7 +4106,8 @@ void setResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult, SQLF
     int32_t functionId = pCtx[i].functionId;
     if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM ||
         functionId == TSDB_FUNC_DIFF || functionId == TSDB_FUNC_DERIVATIVE ||
-        functionId == TSDB_FUNC_SAMPLE || functionId == TSDB_FUNC_MAVG || functionId == TSDB_FUNC_CSUM) {
+        functionId == TSDB_FUNC_SAMPLE || functionId == TSDB_FUNC_MAVG ||
+        functionId == TSDB_FUNC_CSUM || functionId == TSDB_FUNC_UNIQUE) {
       if(i > 0) pCtx[i].ptsOutputBuf = pCtx[i-1].pOutput;
     }
 
@@ -5511,10 +5547,6 @@ SArray* getOrderCheckColumns(SQueryAttr* pQuery) {
   }
 
   if (pQuery->interval.interval > 0) {
-    if (pOrderColumns == NULL) {
-      pOrderColumns = taosArrayInit(1, sizeof(SColIndex));
-    }
-
     SColIndex colIndex = {.colIndex = 0, .colId = 0, .flag = TSDB_COL_NORMAL};
     taosArrayPush(pOrderColumns, &colIndex);
   }
@@ -8753,8 +8785,8 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) {
       pMsg += tListLen(param->pGroupColIndex[i].name);
     }
 
-    pQueryMsg->orderByIdx = htons(pQueryMsg->orderByIdx);
-    pQueryMsg->orderType = htons(pQueryMsg->orderType);
+    //pQueryMsg->orderByIdx = htons(pQueryMsg->orderByIdx);
+    pQueryMsg->groupOrderType = htons(pQueryMsg->groupOrderType);
   }
 
   pQueryMsg->fillType = htons(pQueryMsg->fillType);
@@ -9331,8 +9363,8 @@ SGroupbyExpr *createGroupbyExprFromMsg(SQueryTableMsg *pQueryMsg, SColIndex *pCo
   }
 
   pGroupbyExpr->numOfGroupCols = pQueryMsg->numOfGroupCols;
-  pGroupbyExpr->orderType = pQueryMsg->orderType;
-  pGroupbyExpr->orderIndex = pQueryMsg->orderByIdx;
+  pGroupbyExpr->orderType = pQueryMsg->groupOrderType;
+  //pGroupbyExpr->orderIndex = pQueryMsg->orderByIdx;
 
   pGroupbyExpr->columnInfo = taosArrayInit(pQueryMsg->numOfGroupCols, sizeof(SColIndex));
   for(int32_t i = 0; i < pQueryMsg->numOfGroupCols; ++i) {
@@ -9547,7 +9579,9 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SGroupbyExpr* pGroupbyExpr, S
   pQueryAttr->vgId            = vgId;
   pQueryAttr->pFilters        = pFilters;
   pQueryAttr->range           = pQueryMsg->range;
-  
+  pQueryAttr->uniqueQuery     = isUniqueQuery(numOfOutput, pExprs);
+  pQueryAttr->maxUniqueResult = MAX_UNIQUE_RESULT_SIZE;
+
   pQueryAttr->tableCols = calloc(numOfCols, sizeof(SSingleColumnFilterInfo));
   if (pQueryAttr->tableCols == NULL) {
     goto _cleanup;

src/query/src/qExtbuffer.c

@@ -46,7 +46,7 @@ tExtMemBuffer* createExtMemBuffer(int32_t inMemSize, int32_t elemSize, int32_t p
   
   SExtFileInfo *pFMeta = &pMemBuffer->fileMeta;
 
-  pFMeta->pageSize = DEFAULT_PAGE_SIZE;
+  //pFMeta->pageSize = DEFAULT_PAGE_SIZE;
 
   pFMeta->flushoutData.nAllocSize = 4;
   pFMeta->flushoutData.nLength = 0;

src/query/src/qPlan.c

@@ -77,7 +77,7 @@ static SQueryNode* createQueryNode(int32_t type, const char* name, SQueryNode**
 
       pGroupbyExpr->tableIndex = p->tableIndex;
       pGroupbyExpr->orderType  = p->orderType;
-      pGroupbyExpr->orderIndex = p->orderIndex;
+      //pGroupbyExpr->orderIndex = p->orderIndex;
       pGroupbyExpr->numOfGroupCols = p->numOfGroupCols;
       pGroupbyExpr->columnInfo = taosArrayDup(p->columnInfo);
       pNode->pExtInfo = pGroupbyExpr;

src/query/src/qResultbuf.c

@@ -20,7 +20,7 @@ int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pa
   pResBuf->pageSize     = pagesize;
   pResBuf->numOfPages   = 0;                        // all pages are in buffer in the first place
   pResBuf->totalBufSize = 0;
-  pResBuf->inMemPages   = inMemBufSize/pagesize;    // maximum allowed pages, it is a soft limit.
+  pResBuf->inMemPages   = inMemBufSize/pagesize + 1;    // maximum allowed pages, it is a soft limit.
   pResBuf->allocateId   = -1;
   pResBuf->comp         = true;
   pResBuf->file         = NULL;
@@ -28,7 +28,7 @@ int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pa
   pResBuf->fileSize     = 0;
 
   // at least more than 2 pages must be in memory
-  assert(inMemBufSize >= pagesize * 2);
+  // assert(inMemBufSize >= pagesize * 2);
 
   pResBuf->lruList = tdListNew(POINTER_BYTES);
 
@@ -257,7 +257,7 @@ static char* evicOneDataPage(SDiskbasedResultBuf* pResultBuf) {
     int32_t prev = pResultBuf->inMemPages;
 
     // increase by 50% of previous mem pages
-    pResultBuf->inMemPages = (int32_t)(pResultBuf->inMemPages * 1.5f);
+    pResultBuf->inMemPages = (int32_t)(pResultBuf->inMemPages * 1.5f) + 1;  // if pResultBuf->inMemPages == 1, *1.5 always == 1
 
     qWarn("%p in memory buf page not sufficient, expand from %d to %d, page size:%d", pResultBuf, prev,
           pResultBuf->inMemPages, pResultBuf->pageSize);

src/query/src/qUtil.c

@@ -44,6 +44,9 @@ int32_t getRowNumForMultioutput(SQueryAttr* pQueryAttr, bool topBottomQuery, boo
     }
   }
 
+  if (pQueryAttr->uniqueQuery){
+    return pQueryAttr->maxUniqueResult;
+  }
   return 1;
 }
 
@@ -154,7 +157,7 @@ void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResultRow, int16
     for (int32_t i = 0; i < pRuntimeEnv->pQueryAttr->numOfOutput; ++i) {
       SResultRowCellInfo *pResultInfo = &pResultRow->pCellInfo[i];
 
-      int16_t size = pRuntimeEnv->pQueryAttr->pExpr1[i].base.resType;
+      int16_t size = pRuntimeEnv->pQueryAttr->pExpr1[i].base.resBytes;
       char * s = getPosInResultPage(pRuntimeEnv->pQueryAttr, page, pResultRow->offset, offset);
       memset(s, 0, size);
 
@@ -217,7 +220,6 @@ SResultRow* getNewResultRow(SResultRowPool* p) {
   }
 
   p->position.pos = (p->position.pos + 1)%p->numOfElemPerBlock;
-  initResultRow(ptr);
 
   return ptr;
 }
@@ -451,9 +453,7 @@ int32_t tsDescOrder(const void* p1, const void* p2) {
   }
 }
 
-void
-
-orderTheResultRows(SQueryRuntimeEnv* pRuntimeEnv) {
+void orderTheResultRows(SQueryRuntimeEnv* pRuntimeEnv) {
   __compar_fn_t  fn = NULL;
   if (pRuntimeEnv->pQueryAttr->order.order == TSDB_ORDER_ASC) {
     fn = tsAscOrder;

src/util/src/terror.c

@@ -299,6 +299,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_NOT_ENOUGH_BUFFER,        "Query buffer limit ha
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INCONSISTAN,              "File inconsistance in replica")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_TIME_CONDITION,   "One valid time range condition expected")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_SYS_ERROR,                "System error")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_UNIQUE_RESULT_TOO_LARGE,  "Unique result num is too large")
 
 // grant
 TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED,                "License expired")