taosdata / TDengine / #5029

coverage: 73.003% (+0.02%) from 72.986%
#5029

push

travis-ci

web-flow 
fix(tmq): balance vgroup error (#35183)

273843 of 375111 relevant lines covered (73.0%)

134074995.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.55
/source/libs/function/src/functionMgt.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 









16



#include "functionMgt.h"










17












18



#include "builtins.h"










19



#include "builtinsimpl.h"










20



#include "functionMgtInt.h"










21



#include "nodes.h"










22



#include "querynodes.h"










23



#include "taos.h"










24



#include "taoserror.h"










25



#include "thash.h"










26



#include "tudf.h"










27












28



typedef struct SFuncMgtService {










29



  SHashObj* pFuncNameHashTable;










30



} SFuncMgtService;










31












32



static SFuncMgtService gFunMgtService;










33



static TdThreadOnce    functionHashTableInit = PTHREAD_ONCE_INIT;










34



static int32_t         initFunctionCode = 0;










35












36



static void InitSpecialFuncId() {




1,922,998✔





37



  // just to make sure the funcId of these functions are initialized before used.










38



  (void)fmGetTwstartFuncId();




1,922,998✔





39



  (void)fmGetTwendFuncId();




1,922,998✔





40



  (void)fmGetTwdurationFuncId();




1,922,998✔





41



  (void)fmGetExternalWindowColumnFuncId();




1,922,998✔





42



}




1,922,998✔





43












44



static void doInitFunctionTable() {




1,922,998✔





45



  gFunMgtService.pFuncNameHashTable =




1,922,998✔





46



      taosHashInit(funcMgtBuiltinsNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);




1,922,998✔





47



  if (NULL == gFunMgtService.pFuncNameHashTable) {




1,922,998✔





48



    initFunctionCode = terrno;




×







49



    return;




×







50



  }










51












52



  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {




444,212,538✔





53



    if (TSDB_CODE_SUCCESS != taosHashPut(gFunMgtService.pFuncNameHashTable, funcMgtBuiltins[i].name,




442,289,540✔





54



                                         strlen(funcMgtBuiltins[i].name), &i, sizeof(int32_t))) {




442,289,540✔





55



      initFunctionCode = terrno;




×







56



      return;




×







57



    }










58



  }










59



  InitSpecialFuncId();




1,922,998✔





60



}










61












62



static bool isSpecificClassifyFunc(int32_t funcId, uint64_t classification) {




2,147,483,647✔





63



  if (fmIsUserDefinedFunc(funcId)) {




2,147,483,647✔





64



    return FUNC_MGT_AGG_FUNC == classification










65



               ? FUNC_AGGREGATE_UDF_ID == funcId










66



               : (FUNC_MGT_SCALAR_FUNC == classification ? FUNC_SCALAR_UDF_ID == funcId : false);




8,609,233✔





67



  }










68



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




2,147,483,647✔





69



    return false;




2,147,483,647✔





70



  }










71



  return FUNC_MGT_TEST_MASK(funcMgtBuiltins[funcId].classification, classification);




2,147,483,647✔





72



}










73












74



int32_t fmFuncMgtInit() {




1,922,998✔





75



  (void)taosThreadOnce(&functionHashTableInit, doInitFunctionTable);




1,922,998✔





76



  return initFunctionCode;




1,922,998✔





77



}










78












79



int32_t fmGetFuncInfo(SFunctionNode* pFunc, char* pMsg, int32_t msgLen) {




848,236,660✔





80



  if (NULL != gFunMgtService.pFuncNameHashTable) {




848,236,660✔





81



    void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc->functionName, strlen(pFunc->functionName));




848,068,761✔





82



    if (NULL != pVal) {




848,090,496✔





83



      pFunc->funcId = *(int32_t*)pVal;




847,886,790✔





84



      pFunc->funcType = funcMgtBuiltins[pFunc->funcId].type;




847,886,786✔





85



      return funcMgtBuiltins[pFunc->funcId].translateFunc(pFunc, pMsg, msgLen);




847,887,026✔





86



    }










87



    return TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION;




203,706✔





88



  }










89



  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {




3,259,376✔





90



    if (0 == strcmp(funcMgtBuiltins[i].name, pFunc->functionName)) {




3,259,376✔





91



      pFunc->funcId = i;




175,168✔





92



      pFunc->funcType = funcMgtBuiltins[pFunc->funcId].type;




175,168✔





93



      return funcMgtBuiltins[pFunc->funcId].translateFunc(pFunc, pMsg, msgLen);




175,168✔





94



    }










95



  }










96



  return TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION;




×







97



}










98












99



EFuncReturnRows fmGetFuncReturnRows(SFunctionNode* pFunc) {




17,983,067✔





100



  if (NULL != funcMgtBuiltins[pFunc->funcId].estimateReturnRowsFunc) {




17,983,067✔





101



    return funcMgtBuiltins[pFunc->funcId].estimateReturnRowsFunc(pFunc);




12,706,909✔





102



  }










103



  return (fmIsIndefiniteRowsFunc(pFunc->funcId) || fmIsMultiRowsFunc(pFunc->funcId)) ? FUNC_RETURN_ROWS_INDEFINITE




5,276,158✔





104



                                                                                     : FUNC_RETURN_ROWS_NORMAL;




5,276,158✔





105



}










106












107



bool canCoexistIndefiniteRowsFunc(int32_t funcId1, int32_t funcId2) {




177,924✔





108



  if (fmIsUserDefinedFunc(funcId1) || funcId1 < 0 || funcId1 >= funcMgtBuiltinsNum ||




355,848✔





109



      fmIsUserDefinedFunc(funcId2) || funcId2 < 0 || funcId2 >= funcMgtBuiltinsNum) {




355,848✔





110



    return false;




×







111



  }










112



  if (funcId1 == funcId2) {




177,924✔





113



    return true;




162,451✔





114



  }










115



  if ((funcMgtBuiltins[funcId1].type == FUNCTION_TYPE_LAG || funcMgtBuiltins[funcId1].type == FUNCTION_TYPE_LEAD) &&




15,473✔





116



      (funcMgtBuiltins[funcId2].type == FUNCTION_TYPE_LAG || funcMgtBuiltins[funcId2].type == FUNCTION_TYPE_LEAD)) {




15,034✔





117



    return true;




15,034✔





118



  }










119



  return false;




439✔





120



}










121












122



bool fmIsBuiltinFunc(const char* pFunc) {




16,375✔





123



  return NULL != taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc, strlen(pFunc));




16,375✔





124



}










125












126



EFunctionType fmGetFuncType(const char* pFunc) {




618,306,600✔





127



  void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc, strlen(pFunc));




618,306,600✔





128



  if (NULL != pVal) {




618,321,117✔





129



    return funcMgtBuiltins[*(int32_t*)pVal].type;




618,114,825✔





130



  }










131



  return FUNCTION_TYPE_UDF;




206,292✔





132



}










133












134



EFunctionType fmGetFuncTypeFromId(int32_t funcId) {




23,839,799✔





135



  if (funcId >= 0 && funcId < funcMgtBuiltinsNum) {




23,839,799✔





136



    return funcMgtBuiltins[funcId].type;




23,839,639✔





137



  }










138



  return FUNCTION_TYPE_UDF;




160✔





139



}










140












141



EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {




41,140,696✔





142



  if (fmIsUserDefinedFunc(pFunc->funcId) || pFunc->funcId < 0 || pFunc->funcId >= funcMgtBuiltinsNum) {




41,140,696✔





143



    return FUNC_DATA_REQUIRED_DATA_LOAD;




501✔





144



  }










145



  if (NULL == funcMgtBuiltins[pFunc->funcId].dataRequiredFunc) {




41,146,647✔





146



    return FUNC_DATA_REQUIRED_DATA_LOAD;




×







147



  }










148



  return funcMgtBuiltins[pFunc->funcId].dataRequiredFunc(pFunc, pTimeWindow);




41,146,587✔





149



}










150












151



EFuncDataRequired fmFuncDynDataRequired(int32_t funcId, void* pRes, SDataBlockInfo* pBlockInfo) {




2,147,483,647✔





152



  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {




2,147,483,647✔





153



    return FUNC_DATA_REQUIRED_DATA_LOAD;




243✔





154



  }










155












156



  const char* name = funcMgtBuiltins[funcId].name;




2,147,483,647✔





157



  if ((strcmp(name, "_group_key") == 0) || (strcmp(name, "_select_value") == 0)) {




2,147,483,647✔





158



    return FUNC_DATA_REQUIRED_NOT_LOAD;




337,958✔





159



    ;










160



  }










161












162



  if (funcMgtBuiltins[funcId].dynDataRequiredFunc == NULL) {




2,147,483,647✔





163



    return FUNC_DATA_REQUIRED_DATA_LOAD;




151,373,530✔





164



  } else {










165



    return funcMgtBuiltins[funcId].dynDataRequiredFunc(pRes, pBlockInfo);




2,147,483,647✔





166



  }










167



}










168












169



int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet) {




302,079,972✔





170



  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {




302,079,972✔





171



    return TSDB_CODE_FAILED;




1,334✔





172



  }










173



  pFpSet->getEnv = funcMgtBuiltins[funcId].getEnvFunc;




302,091,407✔





174



  pFpSet->init = funcMgtBuiltins[funcId].initFunc;




302,026,568✔





175



  pFpSet->process = funcMgtBuiltins[funcId].processFunc;




302,044,313✔





176



  pFpSet->finalize = funcMgtBuiltins[funcId].finalizeFunc;




302,023,501✔





177



  pFpSet->combine = funcMgtBuiltins[funcId].combineFunc;




302,043,933✔





178



  pFpSet->processFuncByRow = funcMgtBuiltins[funcId].processFuncByRow;




302,071,341✔





179



  pFpSet->cleanup = funcMgtBuiltins[funcId].cleanupFunc;




302,037,530✔





180



  return TSDB_CODE_SUCCESS;




302,049,876✔





181



}










182












183



int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet) {




37,131✔





184



#ifdef USE_UDF










185



  if (!fmIsUserDefinedFunc(funcId)) {




37,131✔





186



    return TSDB_CODE_FAILED;




×







187



  }










188



  pFpSet->getEnv = udfAggGetEnv;




37,131✔





189



  pFpSet->init = udfAggInit;




37,131✔





190



  pFpSet->process = udfAggProcess;




37,131✔





191



  pFpSet->finalize = udfAggFinalize;




37,131✔





192



  return TSDB_CODE_SUCCESS;




37,131✔





193



#else










194



  TAOS_RETURN(TSDB_CODE_OPS_NOT_SUPPORT);










195



#endif










196



}










197












198



int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet) {




930,903,716✔





199



  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {




930,903,716✔





200



    return TSDB_CODE_FAILED;




57,453✔





201



  }










202



  pFpSet->process = funcMgtBuiltins[funcId].sprocessFunc;




930,931,095✔





203



  pFpSet->getEnv = funcMgtBuiltins[funcId].getEnvFunc;




930,894,564✔





204



  return TSDB_CODE_SUCCESS;




930,910,045✔





205



}










206












207



bool fmIsAggFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_AGG_FUNC); }




2,147,483,647✔





208












209



bool fmIsScalarFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SCALAR_FUNC); }




2,147,483,647✔





210












211



bool fmIsVectorFunc(int32_t funcId) { return !fmIsScalarFunc(funcId) && !fmIsPseudoColumnFunc(funcId); }




1,317,764,180✔





212












213



bool fmIsSelectColsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_COLS_FUNC); }




156,273,025✔





214












215



bool fmIsSelectFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_FUNC); }




1,924,404,422✔





216












217



bool fmIsTimelineFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_TIMELINE_FUNC); }




1,189,942,462✔





218












219



bool fmIsDateTimeFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_DATETIME_FUNC); }




601,818,970✔





220












221



bool fmIsPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PSEUDO_COLUMN_FUNC); }




2,147,483,647✔





222












223



bool fmIsScanPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SCAN_PC_FUNC); }




2,147,483,647✔





224












225



bool fmIsWindowPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_WINDOW_PC_FUNC); }




1,462,324,626✔





226












227



bool fmIsWindowClauseFunc(int32_t funcId) { return fmIsAggFunc(funcId) || fmIsWindowPseudoColumnFunc(funcId); }




62,590,249✔





228












229



bool fmIsWindowIndefRowsFunc(int32_t funcId) { return fmIsVectorFunc(funcId) || fmIsWindowPseudoColumnFunc(funcId); }




394,340✔





230












231



bool fmIsStreamWindowClauseFunc(int32_t funcId) { return fmIsWindowClauseFunc(funcId) || fmIsPlaceHolderFunc(funcId); }




918,498✔





232












233



bool fmIsStreamVectorFunc(int32_t funcId) { return fmIsVectorFunc(funcId) || fmIsPlaceHolderFunc(funcId); }




15,045✔





234












235



bool fmIsIndefiniteRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INDEFINITE_ROWS_FUNC); }




1,614,743,085✔





236












237



bool fmIsSpecialDataRequiredFunc(int32_t funcId) {




93,309,959✔





238



  return isSpecificClassifyFunc(funcId, FUNC_MGT_SPECIAL_DATA_REQUIRED);




93,309,959✔





239



}










240












241



bool fmIsDynamicScanOptimizedFunc(int32_t funcId) {




44,819,101✔





242



  return isSpecificClassifyFunc(funcId, FUNC_MGT_DYNAMIC_SCAN_OPTIMIZED);




44,819,101✔





243



}










244












245



bool fmIsMultiResFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_RES_FUNC); }




1,226,153,800✔





246












247



bool fmIsRepeatScanFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_REPEAT_SCAN_FUNC); }




1,202,773,417✔





248












249



bool fmIsUserDefinedFunc(int32_t funcId) { return funcId > FUNC_UDF_ID_START; }




2,147,483,647✔





250












251



bool fmIsForbidFillFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_FILL_FUNC); }




602,087,256✔





252












253



bool fmIsIntervalInterpoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INTERVAL_INTERPO_FUNC); }




65,158,699✔





254












255



bool fmIsSystemInfoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SYSTEM_INFO_FUNC); }




605,101,496✔





256












257



bool fmIsImplicitTsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_IMPLICIT_TS_FUNC); }




2,147,483,647✔





258












259



bool fmIsClientPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_CLIENT_PC_FUNC); }




612,743,922✔





260












261



bool fmIsMultiRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_ROWS_FUNC); }




1,199,563,839✔





262












263



static bool fmIsKeepOrderFuncId(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_KEEP_ORDER_FUNC); }




727,547,138✔





264












265



bool fmIsKeepOrderFunc(SFunctionNode* pFunc) {




733,934,263✔





266



  if (pFunc->funcType == FUNCTION_TYPE_TOP || pFunc->funcType == FUNCTION_TYPE_BOTTOM ||




733,934,263✔





267



      pFunc->funcType == FUNCTION_TYPE_SAMPLE) {




729,907,835✔





268



    if (pFunc->pParameterList != NULL && pFunc->pParameterList->length >= 2) {




6,390,800✔





269



      SNode* pParam = nodesListGetNode(pFunc->pParameterList, 1);




6,409,476✔





270



      if (pParam != NULL && nodeType(pParam) == QUERY_NODE_VALUE) {




6,409,476✔





271



        SValueNode* pConst = (SValueNode*)pParam;




6,409,476✔





272



        if (pConst->datum.i == 1) {




6,409,476✔





273



          return true;




752,148✔





274



        }










275



      }










276



    }










277



    return false;




5,646,225✔





278



  }










279



  return fmIsKeepOrderFuncId(pFunc->funcId);




727,558,671✔





280



}










281












282



bool fmIsCumulativeFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_CUMULATIVE_FUNC); }




158,791,325✔





283












284



bool fmIsForbidSysTableFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_SYSTABLE_FUNC); }




602,012,021✔





285












286



bool fmIsInterpFunc(int32_t funcId) {




1,233,134,824✔





287



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




1,233,134,824✔





288



    return false;




377,404✔





289



  }










290



  return FUNCTION_TYPE_INTERP == funcMgtBuiltins[funcId].type;




1,232,762,178✔





291



}










292












293



bool fmIsInterpPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INTERP_PC_FUNC); }




1,214,122,119✔





294












295



bool fmIsForecastFunc(int32_t funcId) {




1,183,915,490✔





296



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




1,183,915,490✔





297



    return false;




387,917✔





298



  }










299



  return FUNCTION_TYPE_FORECAST == funcMgtBuiltins[funcId].type;




1,183,528,980✔





300



}










301












302



bool fmIsAnalysisPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_ANALYTICS_PC_FUNC); }




1,217,093,487✔





303












304



bool fmIsImputationCorrelationFunc(int32_t funcId) {




×







305



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




×







306



    return false;




×







307



  }










308












309



  int32_t type = funcMgtBuiltins[funcId].type;




×







310



  return FUNCTION_TYPE_IMPUTATION == type || FUNCTION_TYPE_DTW == type || FUNCTION_TYPE_DTW_PATH == type ||




×







311



         FUNCTION_TYPE_TLCC == type;










312



}










313












314



bool fmIsLastRowFunc(int32_t funcId) {




72,277,661✔





315



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




72,277,661✔





316



    return false;




×







317



  }










318



  return FUNCTION_TYPE_LAST_ROW == funcMgtBuiltins[funcId].type;




72,278,555✔





319



}










320












321



bool fmIsLastFunc(int32_t funcId) {




12,680✔





322



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




12,680✔





323



    return false;




×







324



  }










325



  return FUNCTION_TYPE_LAST == funcMgtBuiltins[funcId].type;




12,680✔





326



}










327












328



bool fmIsNotNullOutputFunc(int32_t funcId) {




455,724,753✔





329



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




455,724,753✔





330



    return false;




89,074✔





331



  }










332



  return FUNCTION_TYPE_LAST == funcMgtBuiltins[funcId].type ||




455,643,887✔





333



         FUNCTION_TYPE_CACHE_LAST == funcMgtBuiltins[funcId].type ||




438,881,256✔





334



         FUNCTION_TYPE_LAST_PARTIAL == funcMgtBuiltins[funcId].type ||




437,614,904✔





335



         FUNCTION_TYPE_LAST_MERGE == funcMgtBuiltins[funcId].type ||




428,600,358✔





336



         FUNCTION_TYPE_FIRST == funcMgtBuiltins[funcId].type ||




424,968,335✔





337



         FUNCTION_TYPE_FIRST_PARTIAL == funcMgtBuiltins[funcId].type ||




413,978,608✔





338



         FUNCTION_TYPE_FIRST_MERGE == funcMgtBuiltins[funcId].type ||




408,813,945✔





339



         FUNCTION_TYPE_COUNT == funcMgtBuiltins[funcId].type ||




406,718,015✔





340



         FUNCTION_TYPE_HYPERLOGLOG == funcMgtBuiltins[funcId].type ||




353,814,955✔





341



         FUNCTION_TYPE_HYPERLOGLOG_PARTIAL == funcMgtBuiltins[funcId].type ||




1,244,474,762✔





342



         FUNCTION_TYPE_HYPERLOGLOG_MERGE == funcMgtBuiltins[funcId].type;




349,947,151✔





343



}










344












345



bool fmIsSelectValueFunc(int32_t funcId) {




344,344,292✔





346



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




344,344,292✔





347



    return false;




×







348



  }










349



  return FUNCTION_TYPE_SELECT_VALUE == funcMgtBuiltins[funcId].type;




344,344,825✔





350



}










351












352



bool fmIsGroupKeyFunc(int32_t funcId) {




262,759,840✔





353



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




262,759,840✔





354



    return false;




×







355



  }










356



  return FUNCTION_TYPE_GROUP_KEY == funcMgtBuiltins[funcId].type;




262,761,652✔





357



}










358












359



bool fmIsHasNullFunc(int32_t funcId) {




×







360



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




×







361



    return false;




×







362



  }










363



  return FUNCTION_TYPE_HAS_NULL == funcMgtBuiltins[funcId].type;




×







364



}










365












366



bool fmisSelectGroupConstValueFunc(int32_t funcId) {




16,742,665✔





367



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




16,742,665✔





368



    return false;




×







369



  }










370



  return FUNCTION_TYPE_GROUP_CONST_VALUE == funcMgtBuiltins[funcId].type;




16,742,665✔





371



}










372












373



bool fmIsElapsedFunc(int32_t funcId) {




10,912,763✔





374



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




10,912,763✔





375



    return false;




×







376



  }










377



  return FUNCTION_TYPE_ELAPSED == funcMgtBuiltins[funcId].type;




10,912,763✔





378



}










379












380



bool fmIsBlockDistFunc(int32_t funcId) {




601,809,627✔





381



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




601,809,627✔





382



    return false;




177,769✔





383



  }










384



  return FUNCTION_TYPE_BLOCK_DIST == funcMgtBuiltins[funcId].type;




601,637,695✔





385



}










386












387



bool fmIsDBUsageFunc(int32_t funcId) {




601,816,388✔





388



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




601,816,388✔





389



    return false;




190,034✔





390



  }










391



  return FUNCTION_TYPE_DB_USAGE == funcMgtBuiltins[funcId].type;




601,628,374✔





392



}










393












394



bool fmIsProcessByRowFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PROCESS_BY_ROW); }




810,895,898✔





395












396



bool fmIsIgnoreNullFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_IGNORE_NULL_FUNC); }




337,189✔





397












398



void fmFuncMgtDestroy() {




1,923,047✔





399



  void* m = gFunMgtService.pFuncNameHashTable;




1,923,047✔





400



  if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {




1,923,047✔





401



    taosHashCleanup(m);




1,922,998✔





402



  }










403



}




1,923,047✔





404












405



#ifdef BUILD_NO_CALL










406



int32_t fmSetNormalFunc(int32_t funcId, SFuncExecFuncs* pFpSet) {










407



  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {










408



    return TSDB_CODE_FAILED;










409



  }










410



  pFpSet->process = funcMgtBuiltins[funcId].processFunc;










411



  return TSDB_CODE_SUCCESS;










412



}










413



#endif










414












415



// function has same input/output type










416



bool fmIsSameInOutType(int32_t funcId) {




49,191,174✔





417



  bool res = false;




49,191,174✔





418



  switch (funcMgtBuiltins[funcId].type) {




49,191,174✔





419



    case FUNCTION_TYPE_MAX:




19,274,942✔





420



    case FUNCTION_TYPE_MIN:










421



    case FUNCTION_TYPE_TOP:










422



    case FUNCTION_TYPE_BOTTOM:










423



    case FUNCTION_TYPE_FIRST:










424



    case FUNCTION_TYPE_LAST:










425



    case FUNCTION_TYPE_SAMPLE:










426



    case FUNCTION_TYPE_TAIL:










427



    case FUNCTION_TYPE_UNIQUE:










428



      res = true;




19,274,942✔





429



      break;




19,274,942✔





430



    default:




29,916,232✔





431



      break;




29,916,232✔





432



  }










433



  return res;




49,191,174✔





434



}










435












436



bool fmIsConstantResFunc(SFunctionNode* pFunc) {




15,015,219✔





437



  SNode* pNode;










438



  FOREACH(pNode, pFunc->pParameterList) {




15,101,643✔





439



    if (nodeType(pNode) != QUERY_NODE_VALUE) {




154,891✔





440



      return false;




68,467✔





441



    }










442



  }










443



  return true;




14,946,752✔





444



}










445












446



bool fmIsSkipScanCheckFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SKIP_SCAN_CHECK_FUNC); }




44,812,924✔





447












448



bool fmIsPrimaryKeyFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PRIMARY_KEY_FUNC); }




524,729,746✔





449












450



bool fmIsPlaceHolderFunc(int32_t funcId) {return isSpecificClassifyFunc(funcId, FUNC_MGT_PLACE_HOLDER_FUNC); }




2,147,483,647✔





451












452



bool fmIsPlaceHolderFuncForExternalWin(int32_t funcId) {




529,254✔





453



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




529,254✔





454



    return false;




×







455



  }










456



  return (funcMgtBuiltins[funcId].type == FUNCTION_TYPE_WSTART || funcMgtBuiltins[funcId].type == FUNCTION_TYPE_WEND ||




885,910✔





457



          funcMgtBuiltins[funcId].type == FUNCTION_TYPE_WDURATION);




356,656✔





458



}










459












460



void getLastCacheDataType(SDataType* pType, int32_t pkBytes) {




×







461



  // TODO: do it later.










462



  pType->bytes = getFirstLastInfoSize(pType->bytes, pkBytes) + VARSTR_HEADER_SIZE;




×







463



  pType->type = TSDB_DATA_TYPE_BINARY;




×







464



}




×







465












466



static int32_t getFuncInfo(SFunctionNode* pFunc) {




157,233,591✔





467



  char msg[128] = {0};




157,233,591✔





468



  return fmGetFuncInfo(pFunc, msg, sizeof(msg));




157,233,591✔





469



}










470












471



int32_t createFunction(const char* pName, SNodeList* pParameterList, SFunctionNode** ppFunc) {




10,120,632✔





472



  int32_t code = nodesMakeNode(QUERY_NODE_FUNCTION, (SNode**)ppFunc);




10,120,632✔





473



  if (NULL == *ppFunc) {




10,120,632✔





474



    return code;




×







475



  }










476



  (void)snprintf((*ppFunc)->functionName, sizeof((*ppFunc)->functionName), "%s", pName);




10,120,632✔





477



  (*ppFunc)->pParameterList = pParameterList;




10,120,632✔





478



  code = getFuncInfo((*ppFunc));




10,120,632✔





479



  if (TSDB_CODE_SUCCESS != code) {




10,120,632✔





480



    (*ppFunc)->pParameterList = NULL;




×







481



    nodesDestroyNode((SNode*)*ppFunc);




×







482



    *ppFunc = NULL;




×







483



    return code;




×







484



  }










485



  return code;




10,120,632✔





486



}










487












488



static void resetOutputChangedFunc(SFunctionNode *pFunc, const SFunctionNode* pSrcFunc) {




147,113,917✔





489



  if (funcMgtBuiltins[pFunc->funcId].type == FUNCTION_TYPE_LAST_MERGE) {




147,113,917✔





490



    pFunc->node.resType = pSrcFunc->node.resType;




22,961,157✔





491



    return;




22,961,157✔





492



  }










493



}










494












495



int32_t createFunctionWithSrcFunc(const char* pName, const SFunctionNode* pSrcFunc, SNodeList* pParameterList, SFunctionNode** ppFunc) {




147,114,934✔





496



  int32_t code = nodesMakeNode(QUERY_NODE_FUNCTION, (SNode**)ppFunc);




147,114,934✔





497



  if (NULL == *ppFunc) {




147,117,751✔





498



    return code;




×







499



  }










500












501



  (*ppFunc)->hasPk = pSrcFunc->hasPk;




147,117,751✔





502



  (*ppFunc)->pkBytes = pSrcFunc->pkBytes;




147,117,707✔





503



  (*ppFunc)->pSrcFuncRef = pSrcFunc;




147,117,707✔





504












505



  (void)snprintf((*ppFunc)->functionName, sizeof((*ppFunc)->functionName), "%s", pName);




147,117,707✔





506



  (*ppFunc)->pParameterList = pParameterList;




147,117,707✔





507



  code = getFuncInfo((*ppFunc));




147,117,751✔





508



  if (TSDB_CODE_SUCCESS != code) {




147,115,941✔





509



    (*ppFunc)->pParameterList = NULL;




×







510



    nodesDestroyNode((SNode*)*ppFunc);




×







511



    *ppFunc = NULL;




×







512



    return code;




×







513



  }










514



  resetOutputChangedFunc(*ppFunc, pSrcFunc);




147,115,941✔





515



  (*ppFunc)->node.relatedTo = pSrcFunc->node.relatedTo;




147,115,235✔





516



  (*ppFunc)->node.bindExprID = pSrcFunc->node.bindExprID;




147,115,235✔





517



  return code;




147,115,235✔





518



}










519












520



static int32_t createColumnByFunc(const SFunctionNode* pFunc, SColumnNode** ppCol) {




97,915,513✔





521



  int32_t code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)ppCol);




97,915,513✔





522



  if (NULL == *ppCol) {




97,921,077✔





523



    return code;




×







524



  }










525



  tstrncpy((*ppCol)->colName, pFunc->node.aliasName, TSDB_COL_NAME_LEN);




97,921,077✔





526



  (*ppCol)->node.resType = pFunc->node.resType;




97,921,077✔





527



  return TSDB_CODE_SUCCESS;




97,921,077✔





528



}










529












530



bool fmIsDistExecFunc(int32_t funcId) {




376,426,911✔





531



  if (fmIsUserDefinedFunc(funcId)) {




376,426,911✔





532



    return false;




104,009✔





533



  }










534



  if (!fmIsVectorFunc(funcId)) {




376,347,306✔





535



    return true;




421,356✔





536



  }










537



  return (NULL != funcMgtBuiltins[funcId].pPartialFunc && NULL != funcMgtBuiltins[funcId].pMergeFunc);




375,934,836✔





538



}










539












540



static int32_t createPartialFunction(const SFunctionNode* pSrcFunc, SFunctionNode** pPartialFunc) {




49,191,021✔





541



  SNodeList* pParameterList = NULL;




49,191,021✔





542



  int32_t    code = nodesCloneList(pSrcFunc->pParameterList, &pParameterList);




49,191,021✔





543



  if (NULL == pParameterList) {




49,194,620✔





544



    return code;




×







545



  }










546



  code =










547



      createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pPartialFunc, pSrcFunc, pParameterList, pPartialFunc);




49,194,620✔





548



  if (TSDB_CODE_SUCCESS != code) {




49,192,252✔





549



    nodesDestroyList(pParameterList);




×







550



    return code;




×







551



  }










552



  (*pPartialFunc)->hasOriginalFunc = true;




49,192,690✔





553



  (*pPartialFunc)->originalFuncId = pSrcFunc->hasOriginalFunc ? pSrcFunc->originalFuncId : pSrcFunc->funcId;




49,192,690✔





554



  char name[TSDB_FUNC_NAME_LEN + TSDB_NAME_DELIMITER_LEN + TSDB_POINTER_PRINT_BYTES + 1] = {0};




49,192,690✔





555












556



  int32_t len = snprintf(name, sizeof(name), "%s.%p", (*pPartialFunc)->functionName, pSrcFunc);




49,192,690✔





557



  if (taosHashBinary(name, len, sizeof(name)) < 0) {




49,194,182✔





558



    return TSDB_CODE_FAILED;




×







559



  }










560



  tstrncpy((*pPartialFunc)->node.aliasName, name, TSDB_COL_NAME_LEN);




49,194,182✔





561



  return TSDB_CODE_SUCCESS;




49,194,182✔





562



}










563












564



static int32_t createMergeFuncPara(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,




97,915,534✔





565



                                   SNodeList** pParameterList) {










566



  SNode*  pRes = NULL;




97,915,534✔





567



  int32_t code = createColumnByFunc(pPartialFunc, (SColumnNode**)&pRes);




97,915,534✔





568



  if (TSDB_CODE_SUCCESS != code) {




97,920,555✔





569



    return code;




×







570



  }










571



  if (NULL != funcMgtBuiltins[pSrcFunc->funcId].createMergeParaFuc) {




97,920,555✔





572



    return funcMgtBuiltins[pSrcFunc->funcId].createMergeParaFuc(pSrcFunc->pParameterList, pRes, pParameterList);




813,192✔





573



  } else {










574



    return nodesListMakeStrictAppend(pParameterList, pRes);




97,107,363✔





575



  }










576



}










577












578



static int32_t createMidFunction(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,




48,723,922✔





579



                                 SFunctionNode** pMidFunc) {










580



  SNodeList*     pParameterList = NULL;




48,723,922✔





581



  SFunctionNode* pFunc = NULL;




48,723,922✔





582












583



  int32_t code = createMergeFuncPara(pSrcFunc, pPartialFunc, &pParameterList);




48,723,922✔





584



  if (TSDB_CODE_SUCCESS == code) {




48,728,359✔





585



    if(funcMgtBuiltins[pSrcFunc->funcId].pMiddleFunc != NULL){




48,728,286✔





586



      code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMiddleFunc, pSrcFunc, pParameterList, &pFunc);




5,632,753✔





587



    }else{










588



      code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pSrcFunc, pParameterList, &pFunc);




43,095,533✔





589



    }










590



  }










591



  if (TSDB_CODE_SUCCESS == code) {




48,727,107✔





592



    tstrncpy(pFunc->node.aliasName, pPartialFunc->node.aliasName, TSDB_COL_NAME_LEN);




48,726,982✔





593



  }










594












595



  if (TSDB_CODE_SUCCESS == code) {




48,727,107✔





596



    *pMidFunc = pFunc;




48,727,107✔





597



  } else {










598



    nodesDestroyList(pParameterList);




×







599



  }










600



  return code;




48,727,076✔





601



}










602












603



static int32_t createMergeFunction(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,




49,191,435✔





604



                                   SFunctionNode** pMergeFunc) {










605



  SNodeList*     pParameterList = NULL;




49,191,435✔





606



  SFunctionNode* pFunc = NULL;




49,191,435✔





607












608



  int32_t code = createMergeFuncPara(pSrcFunc, pPartialFunc, &pParameterList);




49,191,435✔





609



  if (TSDB_CODE_SUCCESS == code) {




49,194,881✔





610



    code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pSrcFunc, pParameterList, &pFunc);




49,194,881✔





611



  }










612



  if (TSDB_CODE_SUCCESS == code) {




49,196,101✔





613



    pFunc->hasOriginalFunc = true;




49,194,881✔





614



    pFunc->originalFuncId = pSrcFunc->hasOriginalFunc ? pSrcFunc->originalFuncId : pSrcFunc->funcId;




49,194,881✔





615



    // overwrite function restype set by translate function










616



    if (fmIsSameInOutType(pSrcFunc->funcId)) {




49,194,881✔





617



      pFunc->node.resType = pSrcFunc->node.resType;




19,275,088✔





618



    }










619



    tstrncpy(pFunc->node.aliasName, pSrcFunc->node.aliasName, TSDB_COL_NAME_LEN);




49,191,939✔





620



  }










621












622



  if (TSDB_CODE_SUCCESS == code) {




49,193,159✔





623



    *pMergeFunc = pFunc;




49,193,159✔





624



  } else {










625



    nodesDestroyList(pParameterList);




×







626



  }










627



  return code;




49,193,253✔





628



}










629












630



int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMidFunc,




49,189,964✔





631



                        SFunctionNode** pMergeFunc) {










632



  if (!fmIsDistExecFunc(pFunc->funcId)) {




49,189,964✔





633



    return TSDB_CODE_FAILED;




×







634



  }










635












636



  int32_t code = createPartialFunction(pFunc, pPartialFunc);




49,193,890✔





637



  if (TSDB_CODE_SUCCESS == code && pMidFunc) {




49,192,273✔





638



    code = createMidFunction(pFunc, *pPartialFunc, pMidFunc);




48,726,805✔





639



  }










640



  if (TSDB_CODE_SUCCESS == code) {




49,192,690✔





641



    code = createMergeFunction(pFunc, *pPartialFunc, pMergeFunc);




49,194,151✔





642



  }










643












644



  if (TSDB_CODE_SUCCESS != code) {




49,190,457✔





645



    nodesDestroyNode((SNode*)*pPartialFunc);




×







646



    if (pMidFunc) nodesDestroyNode((SNode*)*pMidFunc);




×







647



    nodesDestroyNode((SNode*)*pMergeFunc);




×







648



  }










649












650



  return code;




49,192,450✔





651



}










652












653



const char* fmGetFuncName(int32_t funcId) {




343,392✔





654



  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {




343,392✔





655



    return "invalid function";




×







656



  }










657



  return funcMgtBuiltins[funcId].name;




343,392✔





658



}










659












660



/// @param [out] pStateFunc, not changed if error occured or no need to create state func










661



/// @retval 0 for succ, otherwise err occured










662



static int32_t fmCreateStateFunc(const SFunctionNode* pFunc, SFunctionNode** pStateFunc) {




7,470✔





663



  if (funcMgtBuiltins[pFunc->funcId].pStateFunc) {




7,470✔





664



    SNodeList* pParams = NULL;




7,470✔





665



    int32_t    code = nodesCloneList(pFunc->pParameterList, &pParams);




7,470✔





666



    if (!pParams) return code;




7,470✔





667



    code = createFunction(funcMgtBuiltins[pFunc->funcId].pStateFunc, pParams, pStateFunc);




7,470✔





668



    if (TSDB_CODE_SUCCESS != code) {




7,470✔





669



      nodesDestroyList(pParams);




×







670



      return code;




×







671



    }










672



    tstrncpy((*pStateFunc)->node.aliasName, pFunc->node.aliasName, TSDB_COL_NAME_LEN);




7,470✔





673



    tstrncpy((*pStateFunc)->node.userAlias, pFunc->node.userAlias, TSDB_COL_NAME_LEN);




7,470✔





674



  }










675



  return TSDB_CODE_SUCCESS;




7,470✔





676



}










677












678



bool fmIsTSMASupportedFunc(func_id_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_TSMA_FUNC); }




1,255,932✔





679












680



bool fmIsRsmaSupportedFunc(func_id_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_RSMA_FUNC); }




137,608✔





681












682



int32_t fmCreateStateFuncs(SNodeList* pFuncs) {




3,293✔





683



  int32_t code;










684



  SNode*  pNode;










685



  char    buf[128] = {0};




3,293✔





686



  FOREACH(pNode, pFuncs) {




10,763✔





687



    SFunctionNode* pFunc = (SFunctionNode*)pNode;




7,470✔





688



    code = fmGetFuncInfo(pFunc, buf, 128);




7,470✔





689



    if (code) break;




7,470✔





690



    if (fmIsTSMASupportedFunc(pFunc->funcId)) {




7,470✔





691



      SFunctionNode* pNewFunc = NULL;




7,470✔





692



      code = fmCreateStateFunc(pFunc, &pNewFunc);




7,470✔





693



      if (code) {




7,470✔





694



        // error










695



        break;




×







696



      } else if (!pNewFunc) {




7,470✔





697



        // no need state func










698



        continue;




×







699



      } else {










700



        REPLACE_NODE(pNewFunc);




7,470✔





701



        nodesDestroyNode(pNode);




7,470✔





702



      }










703



    }










704



  }










705



  return code;




3,293✔





706



}










707












708



static int32_t fmCreateStateMergeFunc(SFunctionNode* pFunc, SFunctionNode** pStateMergeFunc) {




1,639✔





709



  if (funcMgtBuiltins[pFunc->funcId].pMergeFunc) {




1,639✔





710



    SNodeList* pParams = NULL;




1,341✔





711



    int32_t    code = nodesCloneList(pFunc->pParameterList, &pParams);




1,341✔





712



    if (!pParams) return code;




1,341✔





713



    code = createFunctionWithSrcFunc(funcMgtBuiltins[pFunc->funcId].pMergeFunc, pFunc, pParams, pStateMergeFunc);




1,341✔





714



    if (TSDB_CODE_SUCCESS != code) {




1,341✔





715



      nodesDestroyList(pParams);




×







716



      return code;




×







717



    }










718



    tstrncpy((*pStateMergeFunc)->node.aliasName, pFunc->node.aliasName, TSDB_COL_NAME_LEN);




1,341✔





719



    tstrncpy((*pStateMergeFunc)->node.userAlias, pFunc->node.userAlias, TSDB_COL_NAME_LEN);




1,341✔





720



  }










721



  return TSDB_CODE_SUCCESS;




1,639✔





722



}










723












724



int32_t fmCreateStateMergeFuncs(SNodeList* pFuncs) {




745✔





725



  int32_t code;










726



  SNode*  pNode;










727



  char    buf[128] = {0};




745✔





728



  FOREACH(pNode, pFuncs) {




2,384✔





729



    SFunctionNode* pFunc = (SFunctionNode*)pNode;




1,639✔





730



    if (fmIsTSMASupportedFunc(pFunc->funcId)) {




1,639✔





731



      SFunctionNode* pNewFunc = NULL;




1,639✔





732



      code = fmCreateStateMergeFunc(pFunc, &pNewFunc);




1,639✔





733



      if (code) {




1,639✔





734



        // error










735



        break;




×







736



      } else if (!pNewFunc) {




1,639✔





737



        // no state merge func










738



        continue;




298✔





739



      } else {










740



        REPLACE_NODE(pNewFunc);




1,341✔





741



        nodesDestroyNode(pNode);




1,341✔





742



      }










743



    }










744



  }










745



  return code;




745✔





746



}










747












748



int32_t fmGetFuncId(const char* name) {




9,845,831✔





749



  if (NULL != gFunMgtService.pFuncNameHashTable) {




9,845,831✔





750



    void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, name, strlen(name));




9,845,831✔





751



    if (NULL != pVal) {




9,845,831✔





752



      return *(int32_t*)pVal;




9,845,831✔





753



    }










754



    return -1;




×







755



  }










756



  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {




×







757



    if (0 == strcmp(funcMgtBuiltins[i].name, name)) {




×







758



      return i;




×







759



    }










760



  }










761



  return -1;




×







762



}










763












764



int32_t fmGetTwstartFuncId() {




2,064,112✔





765



  static int32_t twstartFuncId = -2;










766



  if (twstartFuncId < 0) {




2,064,112✔





767



    twstartFuncId = fmGetFuncId("_twstart");




1,922,998✔





768



  }










769



  return twstartFuncId;




2,064,112✔





770



}










771












772



int32_t fmGetTwendFuncId() {




1,954,482✔





773



  static int32_t twendFuncId = -2;










774



  if (twendFuncId < 0) {




1,954,482✔





775



    twendFuncId = fmGetFuncId("_twend");




1,922,998✔





776



  }










777



  return twendFuncId;




1,954,482✔





778



}










779












780



int32_t fmGetTwdurationFuncId() {




1,923,413✔





781



  static int32_t twdurationFuncId = -2;










782



  if (twdurationFuncId < 0) {




1,923,413✔





783



    twdurationFuncId = fmGetFuncId("_twduration");




1,922,998✔





784



  }










785



  return twdurationFuncId;




1,923,413✔





786



}










787












788



int32_t fmGetExternalWindowColumnFuncId() {




1,959,433✔





789



  static int32_t externalWindowColumnFuncId = -2;










790



  if (externalWindowColumnFuncId < 0) {




1,959,433✔





791



    externalWindowColumnFuncId = fmGetFuncId("_external_window_column");




1,922,998✔





792



  }










793



  return externalWindowColumnFuncId;




1,959,433✔





794



}










795












796



bool fmIsMyStateFunc(int32_t funcId, int32_t stateFuncId) {




2,103,433✔





797



  const SBuiltinFuncDefinition* pFunc = &funcMgtBuiltins[funcId];




2,103,433✔





798



  const SBuiltinFuncDefinition* pStateFunc = &funcMgtBuiltins[stateFuncId];




2,103,433✔





799



  if (!pFunc->pStateFunc) {




2,103,433✔





800



    return false;




×







801



  }










802



  if (strcmp(pFunc->pStateFunc, pStateFunc->name) == 0) return true;




2,103,433✔





803



  int32_t stateMergeFuncId = fmGetFuncId(pFunc->pStateFunc);




772,863✔





804



  if (stateMergeFuncId == -1) {




772,863✔





805



    return false;




×







806



  }










807



  const SBuiltinFuncDefinition* pStateMergeFunc = &funcMgtBuiltins[stateMergeFuncId];




772,863✔





808



  return strcmp(pStateFunc->name, pStateMergeFunc->pMergeFunc) == 0;




772,863✔





809



}










810












811



bool fmIsCountLikeFunc(int32_t funcId) {




570,074,297✔





812



  return isSpecificClassifyFunc(funcId, FUNC_MGT_COUNT_LIKE_FUNC);




570,074,297✔





813



}










814












815



bool fmIsRowTsOriginFunc(int32_t funcId) {




5,445,181✔





816



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




5,445,181✔





817



    return false;




×







818



  }










819



  return FUNCTION_TYPE_IROWTS_ORIGIN == funcMgtBuiltins[funcId].type;




5,445,181✔





820



}










821












822



bool fmIsGroupIdFunc(int32_t funcId) {




2,330,590✔





823



  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {




2,330,590✔





824



    return false;




×







825



  }










826



  return FUNCTION_TYPE_GROUP_ID == funcMgtBuiltins[funcId].type;




2,330,590✔





827



}










828












829



const void* fmGetExternalWindowColumnFuncVal(const SStreamRuntimeFuncInfo* pStreamRuntimeFuncInfo, int32_t index) {




10,496,395✔





830



  SSTriggerCalcParam *pParams = taosArrayGet(pStreamRuntimeFuncInfo->pStreamPesudoFuncVals, pStreamRuntimeFuncInfo->curIdx);




10,496,395✔





831



  if (pParams == NULL || pParams->pExternalWindowData == NULL || index < 0) {




10,507,546✔





832



    return NULL;




×







833



  }










834



  return taosArrayGet(pParams->pExternalWindowData, index);




10,507,133✔





835



}










836












837



const void* fmGetStreamPesudoFuncVal(int32_t funcId, const SStreamRuntimeFuncInfo* pStreamRuntimeFuncInfo) {




387,198,893✔





838



  switch (funcMgtBuiltins[funcId].type) {




387,198,893✔





839



    case FUNCTION_TYPE_TGRPID:




137,068✔





840



      return &pStreamRuntimeFuncInfo->groupId;




137,068✔





841



    case FUNCTION_TYPE_PLACEHOLDER_COLUMN:




900,461✔





842



      return pStreamRuntimeFuncInfo->pStreamPartColVals;




900,461✔





843



    case FUNCTION_TYPE_PLACEHOLDER_TBNAME:




2,048,607✔





844



      return pStreamRuntimeFuncInfo->pStreamPartColVals;




2,048,607✔





845



    default:




384,143,700✔





846



      break;




384,143,700✔





847



  }










848












849



  SSTriggerCalcParam *pParams = taosArrayGet(pStreamRuntimeFuncInfo->pStreamPesudoFuncVals, pStreamRuntimeFuncInfo->curIdx);




384,143,700✔





850



  switch (funcMgtBuiltins[funcId].type) {




384,193,475✔





851



    case FUNCTION_TYPE_TPREV_TS:




×







852



      return &pParams->prevTs;




×







853



    case FUNCTION_TYPE_TCURRENT_TS:




1,984,338✔





854



      return &pParams->currentTs;




1,984,338✔





855



    case FUNCTION_TYPE_TNEXT_TS:




×







856



      return &pParams->nextTs;




×







857



    case FUNCTION_TYPE_TWSTART:




331,806,087✔





858



      return &pParams->wstart;




331,806,087✔





859



    case FUNCTION_TYPE_TWEND:




17,270,596✔





860



      return &pParams->wend;




17,270,596✔





861



    case FUNCTION_TYPE_TWDURATION:




1,076,143✔





862



      return &pParams->wduration;




1,076,143✔





863



    case FUNCTION_TYPE_TWROWNUM:




30,700,942✔





864



      return &pParams->wrownum;




30,700,942✔





865



    case FUNCTION_TYPE_TPREV_LOCALTIME:




255,827✔





866



      return &pParams->prevLocalTime;




255,827✔





867



    case FUNCTION_TYPE_TLOCALTIME:




688,675✔





868



      return &pParams->triggerTime;




688,675✔





869



    case FUNCTION_TYPE_TNEXT_LOCALTIME:




393,811✔





870



      return &pParams->nextLocalTime;




393,811✔





871



    case FUNCTION_TYPE_TGRPID:




×







872



      return &pStreamRuntimeFuncInfo->groupId;




×







873



    case FUNCTION_TYPE_PLACEHOLDER_COLUMN:




×







874



      return pStreamRuntimeFuncInfo->pStreamPartColVals;




×







875



    case FUNCTION_TYPE_PLACEHOLDER_TBNAME:




×







876



      return pStreamRuntimeFuncInfo->pStreamPartColVals;




×







877



    case FUNCTION_TYPE_TIDLESTART:




6,048✔





878



      return &pParams->idlestart;




6,048✔





879



    case FUNCTION_TYPE_TIDLEEND:




6,048✔





880



      return &pParams->idleend;




6,048✔





881



    default:




×







882



      break;




×







883



  }










884



  return NULL;




×







885



}










886












887



bool fmIsStreamPesudoColVal(int32_t funcId) {




523,690✔





888



  return funcMgtBuiltins[funcId].type == FUNCTION_TYPE_PLACEHOLDER_COLUMN




523,690✔





889



         || funcMgtBuiltins[funcId].type == FUNCTION_TYPE_PLACEHOLDER_TBNAME;




523,690✔





890



}










891












892



int32_t fmGetStreamPesudoFuncEnv(int32_t funcId, SNodeList* pParamNodes, SFuncExecEnv *pEnv) {




7,369,611✔





893



  if (NULL == pParamNodes || pParamNodes->length < 1) {




7,369,611✔





894



    uError("invalid stream pesudo func param list %p", pParamNodes);




×







895



    return TSDB_CODE_INTERNAL_ERROR;




×







896



  }










897












898



  int32_t firstParamType = nodeType(nodesListGetNode(pParamNodes, 0));




7,370,369✔





899



  if (QUERY_NODE_VALUE != firstParamType) {




7,369,929✔





900



    uError("invalid stream pesudo func first param type %d", firstParamType);




×







901



    return TSDB_CODE_INTERNAL_ERROR;




×







902



  }










903












904



  SValueNode* pResDesc = (SValueNode*)nodesListGetNode(pParamNodes, 0);




7,369,929✔





905



  pEnv->calcMemSize = pResDesc->node.resType.bytes;




7,370,405✔





906












907



  return TSDB_CODE_SUCCESS;




7,370,405✔





908



}










909












910












911



int32_t fmSetStreamPseudoFuncParamVal(int32_t funcId, SNodeList* pParamNodes, const SStreamRuntimeFuncInfo* pStreamRuntimeInfo) {




397,660,609✔





912



  if (!pStreamRuntimeInfo) {




397,660,609✔





913



    uError("internal error, should have pVals for stream pseudo funcs");




×







914



    return TSDB_CODE_INTERNAL_ERROR;




×







915



  }










916



  int32_t code = 0;




397,660,609✔





917



  SArray *pVals1 = NULL;




397,660,609✔





918



  SNode* pFirstParam = nodesListGetNode(pParamNodes, 0);




397,660,609✔





919



  if (nodeType(pFirstParam) != QUERY_NODE_VALUE) {




397,674,886✔





920



    uError("invalid param node type: %d for func: %d", nodeType(pFirstParam), funcId);




×







921



    return TSDB_CODE_INTERNAL_ERROR;




×







922



  }










923












924



  int32_t t = funcMgtBuiltins[funcId].type;




397,728,107✔





925



  uDebug("set stream pseudo func param val, functype: %d, pStreamRuntimeInfo: %p", t, pStreamRuntimeInfo);




397,703,917✔





926



  if (FUNCTION_TYPE_TGRPID == t) {




397,698,406✔





927



    SValue v = {0};




136,826✔





928



    v.type = TSDB_DATA_TYPE_BIGINT;




136,826✔





929



    v.val = *(int64_t*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);




136,826✔





930



    










931



    code = nodesSetValueNodeValue((SValueNode*)pFirstParam, VALUE_GET_DATUM(&v, pFirstParam->type));




137,068✔





932



    if (code != 0) {




137,068✔





933



      uError("failed to set value node value: %s", tstrerror(code));




×







934



      return code;




×







935



    }










936



  } else if (FUNCTION_TYPE_PLACEHOLDER_COLUMN == t) {




397,561,580✔





937



    SNode* pSecondParam = nodesListGetNode(pParamNodes, 1);




900,461✔





938



    if (nodeType(pSecondParam) != QUERY_NODE_VALUE) {




900,610✔





939



      uError("invalid param node type: %d for func: %d", nodeType(pSecondParam), funcId);




×







940



      return TSDB_CODE_INTERNAL_ERROR;




×







941



    }










942



    SArray* pVal = (SArray*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);




900,610✔





943



    int32_t idx = ((SValueNode*)pSecondParam)->datum.i;




900,610✔





944



    if (idx - 1 < 0 || idx - 1 >= taosArrayGetSize(pVal)) {




900,610✔





945



      uError("invalid idx: %d for func: %d, should be in [1, %d]", idx, funcId, (int32_t)taosArrayGetSize(pVal));




149✔





946



      return TSDB_CODE_INTERNAL_ERROR;




×







947



    }










948



    SStreamGroupValue* pValue = taosArrayGet(pVal, idx - 1);




900,610✔





949



    if (pValue == NULL) {




900,610✔





950



      uError("invalid idx: %d for func: %d, should be in [1, %d]", idx, funcId, (int32_t)taosArrayGetSize(pVal));




×







951



      return TSDB_CODE_INTERNAL_ERROR;




×







952



    }










953



    if (!pValue->isNull){




900,610✔





954



      if (pValue->data.type != ((SValueNode*)pFirstParam)->node.resType.type){




894,065✔





955



        uError("invalid value type: %d for func: %d, should be: %d", pValue->data.type, funcId, ((SValueNode*)pFirstParam)->node.resType.type);




×







956



        return TSDB_CODE_INTERNAL_ERROR;




×







957



      }










958



      if (IS_VAR_DATA_TYPE(((SValueNode*)pFirstParam)->node.resType.type)) {




893,821✔





959



        char* tmp = taosMemoryCalloc(1, pValue->data.nData + VARSTR_HEADER_SIZE); 




618,445✔





960



        if (tmp == NULL) {




618,689✔





961



          return TSDB_CODE_OUT_OF_MEMORY;




×







962



        }










963



        taosMemoryFree(((SValueNode*)pFirstParam)->datum.p);




618,689✔





964



        ((SValueNode*)pFirstParam)->datum.p = tmp;




618,296✔





965



        memcpy(varDataVal(((SValueNode*)pFirstParam)->datum.p), pValue->data.pData, pValue->data.nData);




618,445✔





966



        varDataLen(((SValueNode*)pFirstParam)->datum.p) = pValue->data.nData;




618,296✔





967



      } else {










968



        code = nodesSetValueNodeValue((SValueNode*)pFirstParam, VALUE_GET_DATUM(&pValue->data, pValue->data.type));




275,376✔





969



      }










970



    }










971



    ((SValueNode*)pFirstParam)->isNull = pValue->isNull;




900,610✔





972



  } else if (FUNCTION_TYPE_PLACEHOLDER_TBNAME == t) {




396,661,119✔





973



    SArray* pVal = (SArray*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);




1,992,340✔





974



    for (int32_t i = 0; i < taosArrayGetSize(pVal); ++i) {




2,423,141✔





975



      SStreamGroupValue* pValue = taosArrayGet(pVal, i);




2,423,141✔





976



      if (pValue != NULL && pValue->isTbname) {




2,423,141✔





977



        taosMemoryFreeClear(((SValueNode*)pFirstParam)->datum.p);




1,994,021✔





978



        ((SValueNode*)pFirstParam)->datum.p = taosMemoryCalloc(pValue->data.nData + VARSTR_HEADER_SIZE, 1);




1,992,912✔





979



        if (NULL == ((SValueNode*)pFirstParam)->datum.p ) {




1,993,186✔





980



          return terrno;




×







981



        }










982












983



        (void)memcpy(varDataVal(((SValueNode*)pFirstParam)->datum.p), pValue->data.pData, pValue->data.nData);




1,993,187✔





984



        varDataLen(((SValueNode*)pFirstParam)->datum.p) = pValue->data.nData;




1,993,187✔





985



        break;




1,993,748✔





986



      }










987



    }










988



  } else if(FUNCTION_TYPE_EXTERNAL_WINDOW_COLUMN == t) {




394,668,779✔





989



    if (NULL == pParamNodes || LIST_LENGTH(pParamNodes) < 2) {




10,499,286✔





990



      uError("invalid stream external window column param list %p, len: %d", pParamNodes,




413✔





991



             pParamNodes ? LIST_LENGTH(pParamNodes) : 0);










992



      return TSDB_CODE_INTERNAL_ERROR;




×







993



    }










994












995



    SNode* pParamNode = nodesListGetNode(pParamNodes, 1);




10,499,699✔





996



    if (NULL == pParamNode || QUERY_NODE_VALUE != nodeType(pParamNode)) {




10,505,481✔





997



      uError("invalid stream external window column param node %p type %d for func: %d", pParamNode,




1,652✔





998



             pParamNode ? nodeType(pParamNode) : -1, funcId);










999



      return TSDB_CODE_INTERNAL_ERROR;




×







1000



    }










1001












1002



    const SStreamGroupValue* pVal =










1003



        fmGetExternalWindowColumnFuncVal(pStreamRuntimeInfo, ((SValueNode*)pParamNode)->placeholderNo);




10,504,655✔





1004



    if (!pVal) {




10,507,546✔





1005



      uError("failed to set stream pseudo func param val, NULL val for funcId: %d", funcId);




×







1006



      return TSDB_CODE_INTERNAL_ERROR;




×







1007



    }










1008



    if (pVal->isNull) {




10,507,546✔





1009



      ((SValueNode*)pFirstParam)->isNull = true;




826✔





1010



    } else {










1011



      ((SValueNode*)pFirstParam)->isNull = false;




10,506,720✔





1012



      if (IS_VAR_DATA_TYPE(pVal->data.type)) {




10,506,720✔





1013



        // pVal->data.pData already includes the VARSTR header from colDataGetData,










1014



        // so copy it directly as datum.p expects VARSTR format [header][raw data].










1015



        char* tmp = taosMemoryMalloc(pVal->data.nData);




8,260✔





1016



        if (tmp == NULL) {




8,260✔





1017



          return TSDB_CODE_OUT_OF_MEMORY;




×







1018



        }










1019



        taosMemoryFree(((SValueNode*)pFirstParam)->datum.p);




8,260✔





1020



        ((SValueNode*)pFirstParam)->datum.p = tmp;




8,260✔





1021



        memcpy(tmp, pVal->data.pData, pVal->data.nData);




8,260✔





1022



      } else {










1023



        code = nodesSetValueNodeValue((SValueNode*)pFirstParam, (void*)&pVal->data.val);




10,498,460✔





1024



      }










1025



    }










1026



    if (code != 0) {




10,503,416✔





1027



      uError("failed to set value node value: %s", tstrerror(code));




×







1028



      return code;




×







1029



    }










1030



  } else if (LIST_LENGTH(pParamNodes) == 1) {




768,344,246✔





1031



    // twstart, twend










1032



    const void* pVal = fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);




384,184,826✔





1033



    if (!pVal) {




384,186,356✔





1034



      uError("failed to set stream pseudo func param val, NULL val for funcId: %d", funcId);




×







1035



      return TSDB_CODE_INTERNAL_ERROR;




×







1036



    }










1037



    code = nodesSetValueNodeValue((SValueNode*)pFirstParam, (void*)pVal);




384,186,356✔





1038



    if (code != 0) {




384,194,195✔





1039



      uError("failed to set value node value: %s", tstrerror(code));




19,442✔





1040



      return code;




×







1041



    }










1042



  } else {










1043



    uError("invalid placeholder function type: %d", t);




40,187✔





1044



    return TSDB_CODE_INTERNAL_ERROR;




×







1045



  }










1046



  










1047



  return code;




397,726,828✔





1048



}










1049


























    

  • 

    •