Merge pull request #11117 from taosdata/feature/3.0_liaohj

Feature/3.0 liaohj

include/common/tdatablock.h

@@ -133,7 +133,8 @@ static FORCE_INLINE int32_t colDataAppendInt32(SColumnInfoData* pColumnInfoData,
 }
 
 static FORCE_INLINE int32_t colDataAppendInt64(SColumnInfoData* pColumnInfoData, uint32_t currentRow, int64_t* v) {
-  ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_BIGINT || pColumnInfoData->info.type == TSDB_DATA_TYPE_UBIGINT);
+  int32_t type = pColumnInfoData->info.type;
+  ASSERT(type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_UBIGINT || type == TSDB_DATA_TYPE_TIMESTAMP);
   char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
   *(int64_t*)p = *(int64_t*)v;
 }
@@ -175,17 +176,16 @@ size_t blockDataGetRowSize(SSDataBlock* pBlock);
 double blockDataGetSerialRowSize(const SSDataBlock* pBlock);
 size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock);
 
-SSchema* blockDataExtractSchema(const SSDataBlock* pBlock, int32_t* numOfCols);
-
 int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo);
 int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
 
-int32_t      blockDataEnsureColumnCapacity(SColumnInfoData* pColumn, uint32_t numOfRows);
-int32_t      blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
-void         blockDataCleanup(SSDataBlock* pDataBlock);
+int32_t blockDataEnsureColumnCapacity(SColumnInfoData* pColumn, uint32_t numOfRows);
+int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
+void    blockDataCleanup(SSDataBlock* pDataBlock);
+size_t  blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
+void*   blockDataDestroy(SSDataBlock* pBlock);
+
 SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock);
-size_t       blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
-void*        blockDataDestroy(SSDataBlock* pBlock);
 
 void blockDebugShowData(const SArray* dataBlocks);
 

include/common/tmsg.h

@@ -469,8 +469,7 @@ typedef struct {
   int32_t tz;  // query client timezone
   char    intervalUnit;
   char    slidingUnit;
-  char
-      offsetUnit;  // TODO Remove it, the offset is the number of precision tickle, and it must be a immutable duration.
+  char    offsetUnit;  // TODO Remove it, the offset is the number of precision tickle, and it must be a immutable duration.
   int8_t  precision;
   int64_t interval;
   int64_t sliding;

include/libs/function/function.h

@@ -41,6 +41,7 @@ typedef void (*FExecFinalize)(struct SqlFunctionCtx *pCtx);
 typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
 
 typedef struct SScalarFuncExecFuncs {
+  FExecGetEnv getEnv;
   FScalarExecProcess process;
 } SScalarFuncExecFuncs;
 
@@ -241,7 +242,6 @@ typedef struct tExprNode {
   };
 } tExprNode;
 
-void exprTreeToBinary(SBufferWriter* bw, tExprNode* pExprTree);
 void tExprTreeDestroy(tExprNode *pNode, void (*fp)(void *));
 
 typedef struct SAggFunctionInfo {
@@ -267,28 +267,6 @@ struct SScalarParam {
   int32_t            numOfRows;
 };
 
-typedef struct SMultiFunctionsDesc {
-  bool stableQuery;
-  bool groupbyColumn;
-  bool agg;
-  bool arithmeticOnAgg;
-  bool projectionQuery;
-  bool hasFilter;
-  bool onlyTagQuery;
-  bool orderProjectQuery;
-  bool globalMerge;
-  bool multigroupResult;
-  bool blockDistribution;
-  bool stateWindow;
-  bool timewindow;
-  bool sessionWindow;
-  bool topbotQuery;
-  bool interpQuery;
-  bool distinct;
-  bool join;
-  bool continueQuery;
-} SMultiFunctionsDesc;
-
 int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, SResultDataInfo* pInfo, int16_t extLength,
                           bool isSuperTable);
 
@@ -296,8 +274,6 @@ bool qIsValidUdf(SArray* pUdfInfo, const char* name, int32_t len, int32_t* funct
 
 tExprNode* exprTreeFromBinary(const void* data, size_t size);
 
-void extractFunctionDesc(SArray* pFunctionIdList, SMultiFunctionsDesc* pDesc);
-
 tExprNode* exprdup(tExprNode* pTree);
 
 void resetResultRowEntryResult(SqlFunctionCtx* pCtx, int32_t num);

include/libs/scalar/scalar.h

@@ -58,6 +58,14 @@ int32_t ceilFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
 int32_t floorFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
 int32_t roundFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
 
+bool getTimePseudoFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+
+int32_t winStartTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+int32_t winEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+int32_t winDurFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+int32_t qStartTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+int32_t qEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+
 #ifdef __cplusplus
 }
 #endif

source/common/src/tdatablock.c

@@ -331,7 +331,6 @@ int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock) {
     return 0;
   }
 
-  ASSERT(pColInfoData->nullbitmap == NULL);
   pDataBlock->info.window.skey = *(TSKEY*)colDataGetData(pColInfoData, 0);
   pDataBlock->info.window.ekey = *(TSKEY*)colDataGetData(pColInfoData, (pDataBlock->info.rows - 1));
   return 0;
@@ -609,22 +608,6 @@ size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock) {
   return sizeof(int32_t) + pBlock->info.numOfCols * sizeof(int32_t);
 }
 
-SSchema* blockDataExtractSchema(const SSDataBlock* pBlock, int32_t* numOfCols) {
-  SSchema* pSchema = taosMemoryCalloc(pBlock->info.numOfCols, sizeof(SSchema));
-  for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
-    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
-    pSchema[i].bytes = pColInfoData->info.bytes;
-    pSchema[i].type = pColInfoData->info.type;
-    pSchema[i].colId = pColInfoData->info.colId;
-  }
-
-  if (numOfCols != NULL) {
-    *numOfCols = pBlock->info.numOfCols;
-  }
-
-  return pSchema;
-}
-
 double blockDataGetSerialRowSize(const SSDataBlock* pBlock) {
   ASSERT(pBlock != NULL);
   double rowSize = 0;

source/libs/executor/inc/executorimpl.h

@@ -161,20 +161,8 @@ typedef struct STaskCostInfo {
 typedef struct SOperatorCostInfo {
   uint64_t openCost;
   uint64_t execCost;
-//  uint64_t totalRows;
-//  uint64_t totalBytes;
 } SOperatorCostInfo;
 
-typedef struct {
-  int64_t vgroupLimit;
-  int64_t ts;
-} SOrderedPrjQueryInfo;
-
-typedef struct {
-  char*   tags;
-  SArray* pResult;  // SArray<SStddevInterResult>
-} SInterResult;
-
 // The basic query information extracted from the SQueryInfo tree to support the
 // execution of query in a data node.
 typedef struct STaskAttr {
@@ -230,7 +218,6 @@ typedef struct STaskAttr {
   SColumnInfo*         tagColList;
   int32_t              numOfFilterCols;
   int64_t*             fillVal;
-  SOrderedPrjQueryInfo prjInfo;  // limit value for each vgroup, only available in global order projection query.
 
   SSingleColumnFilterInfo* pFilterInfo;
   //  SFilterInfo     *pFilters;
@@ -245,6 +232,7 @@ struct SOperatorInfo;
 
 typedef void (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char **result, int32_t *length);
 typedef bool (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char *result, int32_t length);
+
 typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* param);
 typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* param, bool* newgroup);
 typedef void (*__optr_close_fn_t)(void* param, int32_t num);
@@ -330,11 +318,12 @@ typedef struct SOperatorInfo {
   SResultInfo            resultInfo;
   struct SOperatorInfo** pDownstream;      // downstram pointer list
   int32_t                numOfDownstream;  // number of downstream. The value is always ONE expect for join operator
+  __optr_open_fn_t       _openFn;          // DO NOT invoke this function directly
   __optr_fn_t            getNextFn;
-  __optr_fn_t            cleanupFn;
+  __optr_fn_t            getStreamResFn;   // execute the aggregate in the stream model.
+  __optr_fn_t            cleanupFn;        // call this function to release the allocated resources ASAP
   __optr_close_fn_t      closeFn;
-  __optr_open_fn_t       _openFn;          // DO NOT invoke this function directly
-  __optr_encode_fn_t     encodeResultRow;           //
+  __optr_encode_fn_t     encodeResultRow;
   __optr_decode_fn_t     decodeResultRow;
 } SOperatorInfo;
 
@@ -363,18 +352,18 @@ typedef struct SQInfo {
   STaskCostInfo   summary;
 } SQInfo;
 
-enum {
-  DATA_NOT_READY = 0x1,
-  DATA_READY     = 0x2,
-  DATA_EXHAUSTED = 0x3,
-};
+typedef enum {
+  EX_SOURCE_DATA_NOT_READY = 0x1,
+  EX_SOURCE_DATA_READY     = 0x2,
+  EX_SOURCE_DATA_EXHAUSTED = 0x3,
+} EX_SOURCE_STATUS;
 
 typedef struct SSourceDataInfo {
   struct SExchangeInfo *pEx;
-  int32_t            index;
-  SRetrieveTableRsp *pRsp;
-  uint64_t           totalRows;
-  int32_t            status;
+  int32_t               index;
+  SRetrieveTableRsp    *pRsp;
+  uint64_t              totalRows;
+  EX_SOURCE_STATUS      status;
 } SSourceDataInfo;
 
 typedef struct SLoadRemoteDataInfo {
@@ -383,12 +372,6 @@ typedef struct SLoadRemoteDataInfo {
   uint64_t           totalElapsed;  // total elapsed time
 } SLoadRemoteDataInfo;
 
-enum {
-  EX_SOURCE_DATA_NOT_READY = 0x1,
-  EX_SOURCE_DATA_READY     = 0x2,
-  EX_SOURCE_DATA_EXHAUSTED = 0x3,
-};
-
 typedef struct SExchangeInfo {
   SArray*            pSources;
   SArray*            pSourceDataInfo;
@@ -483,17 +466,24 @@ typedef struct SAggSupporter {
   int32_t            resultRowSize;        // the result buffer size for each result row, with the meta data size for each row
 } SAggSupporter;
 
+typedef enum {
+  OPTR_EXEC_MODEL_BATCH  = 0x1,
+  OPTR_EXEC_MODEL_STREAM = 0x2,
+} OPTR_EXEC_MODEL;
+
 typedef struct STableIntervalOperatorInfo {
-  SOptrBasicInfo     binfo;
-  SGroupResInfo      groupResInfo;
-  SInterval          interval;
-  STimeWindow        win;
-  int32_t            precision;
-  bool               timeWindowInterpo;
-  char             **pRow;
-  SAggSupporter      aggSup;
-  STableQueryInfo   *pCurrent;
-  int32_t            order;
+  SOptrBasicInfo     binfo;                // basic info
+  SGroupResInfo      groupResInfo;         // multiple results build supporter
+  SInterval          interval;             // interval info
+  STimeWindow        win;                  // query time range
+  bool               timeWindowInterpo;    // interpolation needed or not
+  char             **pRow;                 // previous row/tuple of already processed datablock
+  SAggSupporter      aggSup;               // aggregate supporter
+  STableQueryInfo   *pCurrent;             // current tableQueryInfo struct
+  int32_t            order;                // current SSDataBlock scan order
+  OPTR_EXEC_MODEL    execModel;            // operator execution model [batch model|stream model]
+  SArray            *pUpdatedWindow;       // updated time window due to the input data block from the downstream operator.
+  SColumnInfoData    timeWindowData;      // query time window info for scalar function execution.
 } STableIntervalOperatorInfo;
 
 typedef struct SAggOperatorInfo {
@@ -695,12 +685,6 @@ SOperatorInfo* createSLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorI
 SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pdownstream, int32_t numOfDownstream, SSchema* pSchema,
                                       int32_t numOfOutput);
 
-void doSetFilterColumnInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, SSDataBlock* pBlock);
-bool doFilterDataBlock(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, int32_t numOfRows, int8_t* p);
-void doCompactSDataBlock(SSDataBlock* pBlock, int32_t numOfRows, int8_t* p);
-
-SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numOfRows);
-
 void* doDestroyFilterInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols);
 
 void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);

source/libs/function/src/builtins.c

@@ -315,31 +315,31 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "_qstartts",
     .type = FUNCTION_TYPE_QSTARTTS,
-    .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
+    .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_WINDOW_PC_FUNC,
     .checkFunc    = stubCheckAndGetResultType,
-    .getEnvFunc   = NULL,
+    .getEnvFunc   = getTimePseudoFuncEnv,
     .initFunc     = NULL,
-    .sprocessFunc = NULL,
+    .sprocessFunc = qStartTsFunction,
     .finalizeFunc = NULL
   },
   {
     .name = "_qendts",
     .type = FUNCTION_TYPE_QENDTS,
-    .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
+    .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_WINDOW_PC_FUNC,
     .checkFunc    = stubCheckAndGetResultType,
-    .getEnvFunc   = NULL,
+    .getEnvFunc   = getTimePseudoFuncEnv,
     .initFunc     = NULL,
-    .sprocessFunc = NULL,
+    .sprocessFunc = qEndTsFunction,
     .finalizeFunc = NULL
   },
   {
     .name = "_wstartts",
-    .type = FUNCTION_TYPE_QSTARTTS,
+    .type = FUNCTION_TYPE_WSTARTTS,
     .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_WINDOW_PC_FUNC,
     .checkFunc    = stubCheckAndGetResultType,
-    .getEnvFunc   = NULL,
+    .getEnvFunc   = getTimePseudoFuncEnv,
     .initFunc     = NULL,
-    .sprocessFunc = NULL,
+    .sprocessFunc = winStartTsFunction,
     .finalizeFunc = NULL
   },
   {
@@ -347,9 +347,9 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .type = FUNCTION_TYPE_QENDTS,
     .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_WINDOW_PC_FUNC,
     .checkFunc    = stubCheckAndGetResultType,
-    .getEnvFunc   = NULL,
+    .getEnvFunc   = getTimePseudoFuncEnv,
     .initFunc     = NULL,
-    .sprocessFunc = NULL,
+    .sprocessFunc = winEndTsFunction,
     .finalizeFunc = NULL
   },
   {
@@ -357,9 +357,9 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .type = FUNCTION_TYPE_WDURATION,
     .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_WINDOW_PC_FUNC,
     .checkFunc    = stubCheckAndGetResultType,
-    .getEnvFunc   = NULL,
+    .getEnvFunc   = getTimePseudoFuncEnv,
     .initFunc     = NULL,
-    .sprocessFunc = NULL,
+    .sprocessFunc = winDurFunction,
     .finalizeFunc = NULL
   }
 };
@@ -368,6 +368,7 @@ const int32_t funcMgtBuiltinsNum = (sizeof(funcMgtBuiltins) / sizeof(SBuiltinFun
 
 int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
   switch(pFunc->funcType) {
+    case FUNCTION_TYPE_WDURATION:
     case FUNCTION_TYPE_COUNT:
       pFunc->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
       break;
@@ -400,14 +401,18 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
     }
     case FUNCTION_TYPE_CONCAT:
     case FUNCTION_TYPE_ROWTS:
-    case FUNCTION_TYPE_TBNAME:
-    case FUNCTION_TYPE_QSTARTTS:
+    case FUNCTION_TYPE_TBNAME: {
+      // todo
+      break;
+    }
+
     case FUNCTION_TYPE_QENDTS:
-    case FUNCTION_TYPE_WSTARTTS:
+    case FUNCTION_TYPE_QSTARTTS:
     case FUNCTION_TYPE_WENDTS:
-    case FUNCTION_TYPE_WDURATION:
-      // todo
+    case FUNCTION_TYPE_WSTARTTS: {
+      pFunc->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_TIMESTAMP};
       break;
+    }
 
     case FUNCTION_TYPE_ABS:
     case FUNCTION_TYPE_CEIL:

source/libs/function/src/functionMgt.c

@@ -92,6 +92,7 @@ int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet) {
     return TSDB_CODE_FAILED;
   }
   pFpSet->process = funcMgtBuiltins[funcId].sprocessFunc;
+  pFpSet->getEnv  = funcMgtBuiltins[funcId].getEnvFunc;
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/function/src/texpr.c

@@ -116,42 +116,6 @@ bool exprTreeApplyFilter(tExprNode *pExpr, const void *pItem, SExprTraverseSupp
   return param->nodeFilterFn(pItem, pExpr->_node.info);
 }
 
-
-
-static void exprTreeToBinaryImpl(SBufferWriter* bw, tExprNode* expr) {
-  tbufWriteUint8(bw, expr->nodeType);
-  
-  if (expr->nodeType == TEXPR_VALUE_NODE) {
-    SVariant* pVal = expr->pVal;
-    
-    tbufWriteUint32(bw, pVal->nType);
-    if (pVal->nType == TSDB_DATA_TYPE_BINARY) {
-      tbufWriteInt32(bw, pVal->nLen);
-      tbufWrite(bw, pVal->pz, pVal->nLen);
-    } else {
-      tbufWriteInt64(bw, pVal->i);
-    }
-    
-  } else if (expr->nodeType == TEXPR_COL_NODE) {
-    SSchema* pSchema = expr->pSchema;
-    tbufWriteInt16(bw, pSchema->colId);
-    tbufWriteInt16(bw, pSchema->bytes);
-    tbufWriteUint8(bw, pSchema->type);
-    tbufWriteString(bw, pSchema->name);
-    
-  } else if (expr->nodeType == TEXPR_BINARYEXPR_NODE) {
-    tbufWriteUint8(bw, expr->_node.optr);
-    exprTreeToBinaryImpl(bw, expr->_node.pLeft);
-    exprTreeToBinaryImpl(bw, expr->_node.pRight);
-  }
-}
-
-void exprTreeToBinary(SBufferWriter* bw, tExprNode* expr) {
-  if (expr != NULL) {
-    exprTreeToBinaryImpl(bw, expr);
-  }
-}
-
 // TODO: these three functions should be made global
 static void* exception_calloc(size_t nmemb, size_t size) {
   void* p = taosMemoryCalloc(nmemb, size);
@@ -230,97 +194,6 @@ tExprNode* exprTreeFromBinary(const void* data, size_t size) {
   return exprTreeFromBinaryImpl(&br);
 }
 
-tExprNode* exprTreeFromTableName(const char* tbnameCond) {
-  if (!tbnameCond) {
-    return NULL;
-  }
-
-  int32_t anchor = CLEANUP_GET_ANCHOR();
-
-  tExprNode* expr = exception_calloc(1, sizeof(tExprNode));
-  CLEANUP_PUSH_VOID_PTR_PTR(true, tExprTreeDestroy, expr, NULL);
-
-  expr->nodeType = TEXPR_BINARYEXPR_NODE;
-
-  tExprNode* left = exception_calloc(1, sizeof(tExprNode));
-  expr->_node.pLeft = left;
-
-  left->nodeType = TEXPR_COL_NODE;
-  SSchema* pSchema = exception_calloc(1, sizeof(SSchema));
-  left->pSchema = pSchema;
-
-//  *pSchema = NULL;//*tGetTbnameColumnSchema();
-
-  tExprNode* right = exception_calloc(1, sizeof(tExprNode));
-  expr->_node.pRight = right;
-
-  if (strncmp(tbnameCond, QUERY_COND_REL_PREFIX_LIKE, QUERY_COND_REL_PREFIX_LIKE_LEN) == 0) {
-    right->nodeType = TEXPR_VALUE_NODE;
-    expr->_node.optr = OP_TYPE_LIKE;
-    SVariant* pVal = exception_calloc(1, sizeof(SVariant));
-    right->pVal = pVal;
-    size_t len = strlen(tbnameCond + QUERY_COND_REL_PREFIX_LIKE_LEN) + 1;
-    pVal->pz = exception_malloc(len);
-    memcpy(pVal->pz, tbnameCond + QUERY_COND_REL_PREFIX_LIKE_LEN, len);
-    pVal->nType = TSDB_DATA_TYPE_BINARY;
-    pVal->nLen = (int32_t)len;
-
-  } else if (strncmp(tbnameCond, QUERY_COND_REL_PREFIX_MATCH, QUERY_COND_REL_PREFIX_MATCH_LEN) == 0) {
-    right->nodeType = TEXPR_VALUE_NODE;
-    expr->_node.optr = OP_TYPE_MATCH;
-    SVariant* pVal = exception_calloc(1, sizeof(SVariant));
-    right->pVal = pVal;
-    size_t len = strlen(tbnameCond + QUERY_COND_REL_PREFIX_MATCH_LEN) + 1;
-    pVal->pz = exception_malloc(len);
-    memcpy(pVal->pz, tbnameCond + QUERY_COND_REL_PREFIX_MATCH_LEN, len);
-    pVal->nType = TSDB_DATA_TYPE_BINARY;
-    pVal->nLen = (int32_t)len;
-  } else if (strncmp(tbnameCond, QUERY_COND_REL_PREFIX_NMATCH, QUERY_COND_REL_PREFIX_NMATCH_LEN) == 0) {
-    right->nodeType = TEXPR_VALUE_NODE;
-    expr->_node.optr = OP_TYPE_NMATCH;
-    SVariant* pVal = exception_calloc(1, sizeof(SVariant));
-    right->pVal = pVal;
-    size_t len = strlen(tbnameCond + QUERY_COND_REL_PREFIX_NMATCH_LEN) + 1;
-    pVal->pz = exception_malloc(len);
-    memcpy(pVal->pz, tbnameCond + QUERY_COND_REL_PREFIX_NMATCH_LEN, len);
-    pVal->nType = TSDB_DATA_TYPE_BINARY;
-    pVal->nLen = (int32_t)len;
-  } else if (strncmp(tbnameCond, QUERY_COND_REL_PREFIX_IN, QUERY_COND_REL_PREFIX_IN_LEN) == 0) {
-    right->nodeType = TEXPR_VALUE_NODE;
-    expr->_node.optr = OP_TYPE_IN;
-    SVariant* pVal = exception_calloc(1, sizeof(SVariant));
-    right->pVal = pVal;
-    pVal->nType = TSDB_DATA_TYPE_POINTER_ARRAY;
-    pVal->arr = taosArrayInit(2, POINTER_BYTES);
-
-    const char* cond = tbnameCond + QUERY_COND_REL_PREFIX_IN_LEN;
-    for (const char *e = cond; *e != 0; e++) {
-      if (*e == TS_PATH_DELIMITER[0]) {
-        cond = e + 1;
-      } else if (*e == ',') {
-        size_t len = e - cond;
-        char* p = exception_malloc(len + VARSTR_HEADER_SIZE);
-        STR_WITH_SIZE_TO_VARSTR(p, cond, (VarDataLenT)len);
-        cond += len;
-        taosArrayPush(pVal->arr, &p);
-      }
-    }
-
-    if (*cond != 0) {
-      size_t len = strlen(cond) + VARSTR_HEADER_SIZE;
-      
-      char* p = exception_malloc(len);
-      STR_WITH_SIZE_TO_VARSTR(p, cond, (VarDataLenT)(len - VARSTR_HEADER_SIZE));
-      taosArrayPush(pVal->arr, &p);
-    }
-
-    taosArraySortString(pVal->arr, taosArrayCompareString);
-  }
-
-  CLEANUP_EXECUTE_TO(anchor, false);
-  return expr;
-}
-
 void buildFilterSetFromBinary(void **q, const char *buf, int32_t len) {
   SBufferReader br = tbufInitReader(buf, len, false); 
   uint32_t type  = tbufReadUint32(&br);     

source/libs/function/src/tunaryoperator.c

-#include "tunaryoperator.h"
-
-
-
-
-// TODO dynamic define these functions
-//_unary_scalar_fn_t getUnaryScalarOperatorFn(int32_t operator) {
-//  assert(0);
-//}
-
-//bool isStringOperatorFn(int32_t op) {
-//  return op == FUNCTION_LENGTH;
-//}

source/libs/scalar/src/scalar.c

@@ -288,7 +288,7 @@ _return:
   SCL_RET(code);
 }
 
-int32_t sclExecFuncion(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *output) {
+int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *output) {
   if (NULL == node->pParameterList || node->pParameterList->length <= 0) {
     sclError("invalid function parameter list, list:%p, paramNum:%d", node->pParameterList, node->pParameterList ? node->pParameterList->length : 0);
     SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
@@ -420,7 +420,7 @@ EDealRes sclRewriteFunction(SNode** pNode, SScalarCtx *ctx) {
   SFunctionNode *node = (SFunctionNode *)*pNode;
   SScalarParam output = {0};
   
-  ctx->code = sclExecFuncion(node, ctx, &output);
+  ctx->code = sclExecFunction(node, ctx, &output);
   if (ctx->code) {
     return DEAL_RES_ERROR;
   }
@@ -547,7 +547,7 @@ EDealRes sclWalkFunction(SNode* pNode, SScalarCtx *ctx) {
   SFunctionNode *node = (SFunctionNode *)pNode;
   SScalarParam output = {0};
   
-  ctx->code = sclExecFuncion(node, ctx, &output);
+  ctx->code = sclExecFunction(node, ctx, &output);
   if (ctx->code) {
     return DEAL_RES_ERROR;
   }
@@ -667,7 +667,7 @@ int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes) {
 
   int32_t code = 0;
   SScalarCtx ctx = {0};
-  ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
+  ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
   if (NULL == ctx.pRes) {
     sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
     SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
@@ -689,7 +689,7 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
 
   int32_t code = 0;
   SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList};
-  
+  // TODO: OPT performance
   ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
   if (NULL == ctx.pRes) {
     sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
@@ -716,6 +716,3 @@ _return:
   sclFreeRes(ctx.pRes);
   return code;
 }
-
-
-

source/libs/scalar/src/sclfunc.c

@@ -377,3 +377,34 @@ static void reverseCopy(char* dest, const char* src, int16_t type, int32_t numOf
   }
 }
 
+bool getTimePseudoFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
+  pEnv->calcMemSize = sizeof(int64_t);
+  return true;
+}
+
+int32_t qStartTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
+  ASSERT(inputNum == 1);
+  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t *)colDataGetData(pInput->columnData, 0));
+}
+
+int32_t qEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
+  ASSERT(inputNum == 1);
+  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t *)colDataGetData(pInput->columnData, 1));
+}
+
+int32_t winDurFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
+  ASSERT(inputNum == 1);
+  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t *)colDataGetData(pInput->columnData, 2));
+}
+
+int32_t winStartTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
+  ASSERT(inputNum == 1);
+  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t*) colDataGetData(pInput->columnData, 3));
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t winEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
+  ASSERT(inputNum == 1);
+  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t*) colDataGetData(pInput->columnData, 4));
+  return TSDB_CODE_SUCCESS;
+}
\ No newline at end of file