add mode function code

src/client/src/tscGlobalmerge.c

@@ -615,11 +615,9 @@ static void doMergeResultImpl(SOperatorInfo* pInfo, SQLFunctionCtx *pCtx, int32_
       aAggs[functionId].mergeFunc(&pCtx[j]);
     }
 
-    if (functionId == TSDB_FUNC_UNIQUE &&
-        (GET_RES_INFO(&(pCtx[j]))->numOfRes > MAX_UNIQUE_RESULT_ROWS || GET_RES_INFO(&(pCtx[j]))->numOfRes == -1)){
-      tscError("Unique result num is too large. num: %d, limit: %d",
-             GET_RES_INFO(&(pCtx[j]))->numOfRes, MAX_UNIQUE_RESULT_ROWS);
-      longjmp(pInfo->pRuntimeEnv->env, TSDB_CODE_QRY_UNIQUE_RESULT_TOO_LARGE);
+    if (GET_RES_INFO(&(pCtx[j]))->numOfRes == -1){
+      tscError("result num is too large.");
+      longjmp(pInfo->pRuntimeEnv->env, TSDB_CODE_QRY_RESULT_TOO_LARGE);
     }
   }
 }

src/client/src/tscSQLParser.c

@@ -2693,7 +2693,7 @@ int32_t addExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t col
   const char* msg26 = "start param cannot be 0 with 'log_bin'";
   const char* msg27 = "factor param cannot be negative or equal to 0/1";
   const char* msg28 = "the second paramter of diff should be 0 or 1";
-  const char* msg29 = "key timestamp column cannot be used to unique function";
+  const char* msg29 = "key timestamp column cannot be used to unique/mode function";
 
   switch (functionId) {
     case TSDB_FUNC_COUNT: {
@@ -2791,7 +2791,8 @@ int32_t addExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t col
     case TSDB_FUNC_CSUM:
     case TSDB_FUNC_STDDEV:
     case TSDB_FUNC_LEASTSQR:
-    case TSDB_FUNC_ELAPSED: {
+    case TSDB_FUNC_ELAPSED:
+    case TSDB_FUNC_MODE: {
       // 1. valid the number of parameters
       int32_t numOfParams =
           (pItem->pNode->Expr.paramList == NULL) ? 0 : (int32_t)taosArrayGetSize(pItem->pNode->Expr.paramList);
@@ -2852,7 +2853,9 @@ int32_t addExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t col
       // 2. check if sql function can be applied on this column data type
       SSchema* pSchema = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, index.columnIndex);
 
-      if (!IS_NUMERIC_TYPE(pSchema->type) && (functionId != TSDB_FUNC_ELAPSED)) {
+      if (functionId == TSDB_FUNC_MODE && pColumnSchema->colId == PRIMARYKEY_TIMESTAMP_COL_INDEX ){
+        return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg29);
+      } else if (!IS_NUMERIC_TYPE(pSchema->type) && (functionId != TSDB_FUNC_ELAPSED)) {
         return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg1);
       } else if (IS_UNSIGNED_NUMERIC_TYPE(pSchema->type) &&
                  (functionId == TSDB_FUNC_DIFF || functionId == TSDB_FUNC_DERIVATIVE)) {

src/inc/taoserror.h

@@ -293,7 +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")
+#define TSDB_CODE_QRY_RESULT_TOO_LARGE          TAOS_DEF_ERROR_CODE(0, 0x0711)  //"result num is too large")
 
 // grant
 #define TSDB_CODE_GRANT_EXPIRED                 TAOS_DEF_ERROR_CODE(0, 0x0800)  //"License expired"

src/query/inc/qAggMain.h

@@ -79,8 +79,9 @@ extern "C" {
 #define TSDB_FUNC_ELAPSED      37
 #define TSDB_FUNC_HISTOGRAM    38
 #define TSDB_FUNC_UNIQUE       39
+#define TSDB_FUNC_MODE         40
 
-#define TSDB_FUNC_MAX_NUM    40
+#define TSDB_FUNC_MAX_NUM    41
 
 #define TSDB_FUNCSTATE_SO           0x1u    // single output
 #define TSDB_FUNCSTATE_MO           0x2u    // dynamic number of output, not multinumber of output e.g., TOP/BOTTOM
@@ -148,7 +149,7 @@ typedef struct SResultRowCellInfo {
   int8_t   hasResult;       // result generated, not NULL value
   bool     initialized;     // output buffer has been initialized
   bool     complete;        // query has completed
-  uint32_t numOfRes;        // num of output result in current buffer
+  int32_t  numOfRes;        // num of output result in current buffer
 } SResultRowCellInfo;
 
 typedef struct SPoint1 {
@@ -203,6 +204,7 @@ typedef struct SQLFunctionCtx {
   SPoint1      end;
 
   SHashObj     **pUniqueSet;   // for unique function
+  SHashObj     **pModeSet;     // for mode function
 } SQLFunctionCtx;
 
 typedef struct SAggFunctionInfo {

src/query/inc/qExecutor.h

@@ -91,6 +91,7 @@ typedef struct SResultRow {
   STimeWindow   win;
   char         *key;               // start key of current result row
   SHashObj     *uniqueHash;  // for unique function
+  SHashObj     *modeHash;  // for unique function
 } SResultRow;
 
 typedef struct SResultRowCell {

src/query/inc/qResultbuf.h

@@ -80,6 +80,8 @@ typedef struct SDiskbasedResultBuf {
 #define PAGE_INFO_INITIALIZER         (SPageDiskInfo){-1, -1}
 #define MAX_UNIQUE_RESULT_ROWS (1000)
 #define MAX_UNIQUE_RESULT_SIZE (1024*1024*1)
+#define MAX_MODE_INNER_RESULT_ROWS (1000000)
+#define MAX_MODE_INNER_RESULT_SIZE (1024*1024*10)
 /**
  * create disk-based result buffer
  * @param pResultBuf

src/query/src/qExecutor.c

@@ -1009,11 +1009,9 @@ static void doApplyFunctions(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx
       }
     }
 
-    if (functionId == TSDB_FUNC_UNIQUE &&
-        (GET_RES_INFO(&(pCtx[k]))->numOfRes > MAX_UNIQUE_RESULT_ROWS || GET_RES_INFO(&(pCtx[k]))->numOfRes == -1)){
-      qError("Unique result num is too large. num: %d, limit: %d",
-             GET_RES_INFO(&(pCtx[k]))->numOfRes, MAX_UNIQUE_RESULT_ROWS);
-      longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_UNIQUE_RESULT_TOO_LARGE);
+    if (GET_RES_INFO(&(pCtx[k]))->numOfRes == -1){
+      qError("result num is too large.");
+      longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_RESULT_TOO_LARGE);
     }
 
     // restore it
@@ -1276,11 +1274,9 @@ static void doAggregateImpl(SOperatorInfo* pOperator, TSKEY startTs, SQLFunction
         assert(0);
       }
 
-      if (functionId == TSDB_FUNC_UNIQUE &&
-          (GET_RES_INFO(&(pCtx[k]))->numOfRes > MAX_UNIQUE_RESULT_ROWS || GET_RES_INFO(&(pCtx[k]))->numOfRes == -1)){
-        qError("Unique result num is too large. num: %d, limit: %d",
-               GET_RES_INFO(&(pCtx[k]))->numOfRes, MAX_UNIQUE_RESULT_ROWS);
-        longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_UNIQUE_RESULT_TOO_LARGE);
+      if (GET_RES_INFO(&(pCtx[k]))->numOfRes == -1){
+        qError("Mode inner result num is too large");
+        longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_RESULT_TOO_LARGE);
       }
     }
   }
@@ -3690,6 +3686,8 @@ void setDefaultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SOptrBasicInfo *pInfo, i
     pCtx[i].resultInfo   = pCellInfo;
     if (pCtx[i].functionId == TSDB_FUNC_UNIQUE) {
       pCtx[i].pUniqueSet = &pRow->uniqueHash;
+    }else if (pCtx[i].functionId == TSDB_FUNC_MODE) {
+      pCtx[i].pUniqueSet = &pRow->modeHash;
     }
     pCtx[i].pOutput      = pData->pData;
     pCtx[i].currentStage = stage;
@@ -4027,6 +4025,8 @@ void setResultRowOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pRe
     pCtx[i].resultInfo = getResultCell(pResult, i, rowCellInfoOffset);
     if (pCtx[i].functionId == TSDB_FUNC_UNIQUE){
       pCtx[i].pUniqueSet = &pResult->uniqueHash;
+    }else if (pCtx[i].functionId == TSDB_FUNC_MODE){
+      pCtx[i].pUniqueSet = &pResult->modeHash;
     }
 
     SResultRowCellInfo* pResInfo = pCtx[i].resultInfo;
@@ -4123,6 +4123,8 @@ void setResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult, SQLF
     pCtx[i].resultInfo = getResultCell(pResult, i, rowCellInfoOffset);
     if (pCtx[i].functionId == TSDB_FUNC_UNIQUE) {
       pCtx[i].pUniqueSet = &pResult->uniqueHash;
+    }else if (pCtx[i].functionId == TSDB_FUNC_MODE) {
+      pCtx[i].pUniqueSet = &pResult->modeHash;
     }
   }
 }

src/util/src/terror.c

@@ -299,7 +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")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_RESULT_TOO_LARGE,         "result num is too large")
 
 // grant
 TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED,                "License expired")