• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #5006

29 Mar 2026 04:32AM UTC coverage: 72.274% (+0.1%) from 72.152%
#5006

push

travis-ci

web-flow
refactor: do some internal refactor for TDgpt. (#34955)

253711 of 351039 relevant lines covered (72.27%)

131490495.89 hits per line

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

82.41
/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 doInitFunctionTable() {
1,411,010✔
37
  gFunMgtService.pFuncNameHashTable =
1,411,010✔
38
      taosHashInit(funcMgtBuiltinsNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
1,411,010✔
39
  if (NULL == gFunMgtService.pFuncNameHashTable) {
1,411,010✔
40
    initFunctionCode = terrno;
×
41
    return;
×
42
  }
43

44
  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {
321,710,280✔
45
    if (TSDB_CODE_SUCCESS != taosHashPut(gFunMgtService.pFuncNameHashTable, funcMgtBuiltins[i].name,
320,299,270✔
46
                                         strlen(funcMgtBuiltins[i].name), &i, sizeof(int32_t))) {
320,299,270✔
47
      initFunctionCode = terrno;
×
48
      return;
×
49
    }
50
  }
51
}
52

53
static bool isSpecificClassifyFunc(int32_t funcId, uint64_t classification) {
2,147,483,647✔
54
  if (fmIsUserDefinedFunc(funcId)) {
2,147,483,647✔
55
    return FUNC_MGT_AGG_FUNC == classification
56
               ? FUNC_AGGREGATE_UDF_ID == funcId
57
               : (FUNC_MGT_SCALAR_FUNC == classification ? FUNC_SCALAR_UDF_ID == funcId : false);
8,366,920✔
58
  }
59
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
2,147,483,647✔
60
    return false;
2,147,483,647✔
61
  }
62
  return FUNC_MGT_TEST_MASK(funcMgtBuiltins[funcId].classification, classification);
2,147,483,647✔
63
}
64

65
int32_t fmFuncMgtInit() {
1,411,010✔
66
  (void)taosThreadOnce(&functionHashTableInit, doInitFunctionTable);
1,411,010✔
67
  return initFunctionCode;
1,411,010✔
68
}
69

70
int32_t fmGetFuncInfo(SFunctionNode* pFunc, char* pMsg, int32_t msgLen) {
822,743,145✔
71
  if (NULL != gFunMgtService.pFuncNameHashTable) {
822,743,145✔
72
    void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc->functionName, strlen(pFunc->functionName));
822,579,251✔
73
    if (NULL != pVal) {
822,600,021✔
74
      pFunc->funcId = *(int32_t*)pVal;
822,399,926✔
75
      pFunc->funcType = funcMgtBuiltins[pFunc->funcId].type;
822,399,465✔
76
      return funcMgtBuiltins[pFunc->funcId].translateFunc(pFunc, pMsg, msgLen);
822,398,853✔
77
    }
78
    return TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION;
200,095✔
79
  }
80
  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {
3,175,604✔
81
    if (0 == strcmp(funcMgtBuiltins[i].name, pFunc->functionName)) {
3,176,016✔
82
      pFunc->funcId = i;
170,688✔
83
      pFunc->funcType = funcMgtBuiltins[pFunc->funcId].type;
170,688✔
84
      return funcMgtBuiltins[pFunc->funcId].translateFunc(pFunc, pMsg, msgLen);
170,688✔
85
    }
86
  }
87
  return TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION;
×
88
}
89

90
EFuncReturnRows fmGetFuncReturnRows(SFunctionNode* pFunc) {
17,279,548✔
91
  if (NULL != funcMgtBuiltins[pFunc->funcId].estimateReturnRowsFunc) {
17,279,548✔
92
    return funcMgtBuiltins[pFunc->funcId].estimateReturnRowsFunc(pFunc);
12,180,005✔
93
  }
94
  return (fmIsIndefiniteRowsFunc(pFunc->funcId) || fmIsMultiRowsFunc(pFunc->funcId)) ? FUNC_RETURN_ROWS_INDEFINITE
5,099,543✔
95
                                                                                     : FUNC_RETURN_ROWS_NORMAL;
5,099,543✔
96
}
97

98
bool fmIsBuiltinFunc(const char* pFunc) {
16,276✔
99
  return NULL != taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc, strlen(pFunc));
16,276✔
100
}
101

102
EFunctionType fmGetFuncType(const char* pFunc) {
599,230,776✔
103
  void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc, strlen(pFunc));
599,230,776✔
104
  if (NULL != pVal) {
599,246,681✔
105
    return funcMgtBuiltins[*(int32_t*)pVal].type;
599,030,449✔
106
  }
107
  return FUNCTION_TYPE_UDF;
216,232✔
108
}
109

110
EFunctionType fmGetFuncTypeFromId(int32_t funcId) {
3,039,410✔
111
  if (funcId < funcMgtBuiltinsNum) {
3,039,410✔
112
    return funcMgtBuiltins[funcId].type;
3,039,410✔
113
  }
114
  return FUNCTION_TYPE_UDF;
×
115
}
116

117
EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {
40,537,616✔
118
  if (fmIsUserDefinedFunc(pFunc->funcId) || pFunc->funcId < 0 || pFunc->funcId >= funcMgtBuiltinsNum) {
40,537,616✔
119
    return FUNC_DATA_REQUIRED_DATA_LOAD;
249✔
120
  }
121
  if (NULL == funcMgtBuiltins[pFunc->funcId].dataRequiredFunc) {
40,540,528✔
122
    return FUNC_DATA_REQUIRED_DATA_LOAD;
×
123
  }
124
  return funcMgtBuiltins[pFunc->funcId].dataRequiredFunc(pFunc, pTimeWindow);
40,540,569✔
125
}
126

127
EFuncDataRequired fmFuncDynDataRequired(int32_t funcId, void* pRes, SDataBlockInfo* pBlockInfo) {
2,147,483,647✔
128
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
2,147,483,647✔
129
    return FUNC_DATA_REQUIRED_DATA_LOAD;
42✔
130
  }
131

132
  const char* name = funcMgtBuiltins[funcId].name;
2,147,483,647✔
133
  if ((strcmp(name, "_group_key") == 0) || (strcmp(name, "_select_value") == 0)) {
2,147,483,647✔
134
    return FUNC_DATA_REQUIRED_NOT_LOAD;
275,694✔
135
    ;
136
  }
137

138
  if (funcMgtBuiltins[funcId].dynDataRequiredFunc == NULL) {
2,147,483,647✔
139
    return FUNC_DATA_REQUIRED_DATA_LOAD;
145,108,577✔
140
  } else {
141
    return funcMgtBuiltins[funcId].dynDataRequiredFunc(pRes, pBlockInfo);
2,147,483,647✔
142
  }
143
}
144

145
int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet) {
327,632,727✔
146
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
327,632,727✔
147
    return TSDB_CODE_FAILED;
874✔
148
  }
149
  pFpSet->getEnv = funcMgtBuiltins[funcId].getEnvFunc;
327,641,239✔
150
  pFpSet->init = funcMgtBuiltins[funcId].initFunc;
327,584,615✔
151
  pFpSet->process = funcMgtBuiltins[funcId].processFunc;
327,612,059✔
152
  pFpSet->finalize = funcMgtBuiltins[funcId].finalizeFunc;
327,600,695✔
153
  pFpSet->combine = funcMgtBuiltins[funcId].combineFunc;
327,603,515✔
154
  pFpSet->processFuncByRow = funcMgtBuiltins[funcId].processFuncByRow;
327,593,997✔
155
  pFpSet->cleanup = funcMgtBuiltins[funcId].cleanupFunc;
327,606,797✔
156
  return TSDB_CODE_SUCCESS;
327,635,548✔
157
}
158

159
int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet) {
36,370✔
160
#ifdef USE_UDF
161
  if (!fmIsUserDefinedFunc(funcId)) {
36,370✔
162
    return TSDB_CODE_FAILED;
×
163
  }
164
  pFpSet->getEnv = udfAggGetEnv;
36,370✔
165
  pFpSet->init = udfAggInit;
36,370✔
166
  pFpSet->process = udfAggProcess;
36,370✔
167
  pFpSet->finalize = udfAggFinalize;
36,370✔
168
  return TSDB_CODE_SUCCESS;
36,370✔
169
#else
170
  TAOS_RETURN(TSDB_CODE_OPS_NOT_SUPPORT);
171
#endif
172
}
173

174
int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet) {
854,930,634✔
175
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
854,930,634✔
176
    return TSDB_CODE_FAILED;
55,225✔
177
  }
178
  pFpSet->process = funcMgtBuiltins[funcId].sprocessFunc;
855,010,922✔
179
  pFpSet->getEnv = funcMgtBuiltins[funcId].getEnvFunc;
854,958,331✔
180
  return TSDB_CODE_SUCCESS;
854,955,573✔
181
}
182

183
bool fmIsAggFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_AGG_FUNC); }
2,147,483,647✔
184

185
bool fmIsScalarFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SCALAR_FUNC); }
2,147,483,647✔
186

187
bool fmIsVectorFunc(int32_t funcId) { return !fmIsScalarFunc(funcId) && !fmIsPseudoColumnFunc(funcId); }
1,277,538,433✔
188

189
bool fmIsSelectColsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_COLS_FUNC); }
151,101,929✔
190

191
bool fmIsSelectFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_FUNC); }
1,931,958,741✔
192

193
bool fmIsTimelineFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_TIMELINE_FUNC); }
1,152,701,891✔
194

195
bool fmIsDateTimeFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_DATETIME_FUNC); }
582,959,642✔
196

197
bool fmIsPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PSEUDO_COLUMN_FUNC); }
2,147,483,647✔
198

199
bool fmIsScanPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SCAN_PC_FUNC); }
2,147,483,647✔
200

201
bool fmIsWindowPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_WINDOW_PC_FUNC); }
1,456,229,085✔
202

203
bool fmIsWindowClauseFunc(int32_t funcId) { return fmIsAggFunc(funcId) || fmIsWindowPseudoColumnFunc(funcId); }
58,543,697✔
204

205
bool fmIsStreamWindowClauseFunc(int32_t funcId) { return fmIsWindowClauseFunc(funcId) || fmIsPlaceHolderFunc(funcId); }
334,972✔
206

207
bool fmIsStreamVectorFunc(int32_t funcId) { return fmIsVectorFunc(funcId) || fmIsPlaceHolderFunc(funcId); }
10,472✔
208

209
bool fmIsIndefiniteRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INDEFINITE_ROWS_FUNC); }
1,544,123,480✔
210

211
bool fmIsSpecialDataRequiredFunc(int32_t funcId) {
91,036,394✔
212
  return isSpecificClassifyFunc(funcId, FUNC_MGT_SPECIAL_DATA_REQUIRED);
91,036,394✔
213
}
214

215
bool fmIsDynamicScanOptimizedFunc(int32_t funcId) {
43,375,647✔
216
  return isSpecificClassifyFunc(funcId, FUNC_MGT_DYNAMIC_SCAN_OPTIMIZED);
43,375,647✔
217
}
218

219
bool fmIsMultiResFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_RES_FUNC); }
1,186,904,465✔
220

221
bool fmIsRepeatScanFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_REPEAT_SCAN_FUNC); }
1,164,779,659✔
222

223
bool fmIsUserDefinedFunc(int32_t funcId) { return funcId > FUNC_UDF_ID_START; }
2,147,483,647✔
224

225
bool fmIsForbidFillFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_FILL_FUNC); }
583,205,339✔
226

227
bool fmIsIntervalInterpoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INTERVAL_INTERPO_FUNC); }
62,448,736✔
228

229
bool fmIsSystemInfoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SYSTEM_INFO_FUNC); }
586,127,773✔
230

231
bool fmIsImplicitTsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_IMPLICIT_TS_FUNC); }
2,147,483,647✔
232

233
bool fmIsClientPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_CLIENT_PC_FUNC); }
585,867,583✔
234

235
bool fmIsMultiRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_ROWS_FUNC); }
1,161,990,442✔
236

237
static bool fmIsKeepOrderFuncId(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_KEEP_ORDER_FUNC); }
704,279,983✔
238

239
bool fmIsKeepOrderFunc(SFunctionNode* pFunc) {
710,492,922✔
240
  if (pFunc->funcType == FUNCTION_TYPE_TOP || pFunc->funcType == FUNCTION_TYPE_BOTTOM ||
710,492,922✔
241
      pFunc->funcType == FUNCTION_TYPE_SAMPLE) {
706,568,133✔
242
    if (pFunc->pParameterList != NULL && pFunc->pParameterList->length >= 2) {
6,213,582✔
243
      SNode* pParam = nodesListGetNode(pFunc->pParameterList, 1);
6,229,682✔
244
      if (pParam != NULL && nodeType(pParam) == QUERY_NODE_VALUE) {
6,229,682✔
245
        SValueNode* pConst = (SValueNode*)pParam;
6,229,682✔
246
        if (pConst->datum.i == 1) {
6,229,682✔
247
          return true;
730,168✔
248
        }
249
      }
250
    }
251
    return false;
5,490,180✔
252
  }
253
  return fmIsKeepOrderFuncId(pFunc->funcId);
704,285,064✔
254
}
255

256
bool fmIsCumulativeFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_CUMULATIVE_FUNC); }
153,417,751✔
257

258
bool fmIsForbidSysTableFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_SYSTABLE_FUNC); }
583,138,759✔
259

260
bool fmIsInterpFunc(int32_t funcId) {
1,194,694,482✔
261
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
1,194,694,482✔
262
    return false;
370,264✔
263
  }
264
  return FUNCTION_TYPE_INTERP == funcMgtBuiltins[funcId].type;
1,194,328,701✔
265
}
266

267
bool fmIsInterpPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INTERP_PC_FUNC); }
1,176,027,806✔
268

269
bool fmIsForecastFunc(int32_t funcId) {
1,146,957,252✔
270
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
1,146,957,252✔
271
    return false;
378,356✔
272
  }
273
  return FUNCTION_TYPE_FORECAST == funcMgtBuiltins[funcId].type;
1,146,581,113✔
274
}
275

276
bool fmIsAnalysisPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_ANALYTICS_PC_FUNC); }
1,178,915,447✔
277

278
bool fmIsImputationCorrelationFunc(int32_t funcId) {
×
279
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
×
280
    return false;
×
281
  }
282

283
  int32_t type = funcMgtBuiltins[funcId].type;
×
284
  return FUNCTION_TYPE_IMPUTATION == type || FUNCTION_TYPE_DTW == type || FUNCTION_TYPE_DTW_PATH == type ||
×
285
         FUNCTION_TYPE_TLCC == type;
286
}
287

288
bool fmIsLastRowFunc(int32_t funcId) {
69,198,762✔
289
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
69,198,762✔
290
    return false;
10✔
291
  }
292
  return FUNCTION_TYPE_LAST_ROW == funcMgtBuiltins[funcId].type;
69,198,996✔
293
}
294

295
bool fmIsLastFunc(int32_t funcId) {
12,300✔
296
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
12,300✔
297
    return false;
×
298
  }
299
  return FUNCTION_TYPE_LAST == funcMgtBuiltins[funcId].type;
12,300✔
300
}
301

302
bool fmIsNotNullOutputFunc(int32_t funcId) {
488,563,045✔
303
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
488,563,045✔
304
    return false;
85,967✔
305
  }
306
  return FUNCTION_TYPE_LAST == funcMgtBuiltins[funcId].type ||
488,485,496✔
307
         FUNCTION_TYPE_CACHE_LAST == funcMgtBuiltins[funcId].type ||
470,020,858✔
308
         FUNCTION_TYPE_LAST_PARTIAL == funcMgtBuiltins[funcId].type ||
468,745,504✔
309
         FUNCTION_TYPE_LAST_MERGE == funcMgtBuiltins[funcId].type ||
459,649,545✔
310
         FUNCTION_TYPE_FIRST == funcMgtBuiltins[funcId].type ||
456,116,241✔
311
         FUNCTION_TYPE_FIRST_PARTIAL == funcMgtBuiltins[funcId].type ||
443,272,915✔
312
         FUNCTION_TYPE_FIRST_MERGE == funcMgtBuiltins[funcId].type ||
437,801,445✔
313
         FUNCTION_TYPE_COUNT == funcMgtBuiltins[funcId].type ||
435,842,268✔
314
         FUNCTION_TYPE_HYPERLOGLOG == funcMgtBuiltins[funcId].type ||
381,242,086✔
315
         FUNCTION_TYPE_HYPERLOGLOG_PARTIAL == funcMgtBuiltins[funcId].type ||
1,336,083,038✔
316
         FUNCTION_TYPE_HYPERLOGLOG_MERGE == funcMgtBuiltins[funcId].type;
377,503,848✔
317
}
318

319
bool fmIsSelectValueFunc(int32_t funcId) {
346,595,349✔
320
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
346,595,349✔
321
    return false;
×
322
  }
323
  return FUNCTION_TYPE_SELECT_VALUE == funcMgtBuiltins[funcId].type;
346,596,219✔
324
}
325

326
bool fmIsGroupKeyFunc(int32_t funcId) {
268,182,259✔
327
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
268,182,259✔
328
    return false;
×
329
  }
330
  return FUNCTION_TYPE_GROUP_KEY == funcMgtBuiltins[funcId].type;
268,183,114✔
331
}
332

333
bool fmIsHasNullFunc(int32_t funcId) {
×
334
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
×
335
    return false;
×
336
  }
337
  return FUNCTION_TYPE_HAS_NULL == funcMgtBuiltins[funcId].type;
×
338
}
339

340
bool fmisSelectGroupConstValueFunc(int32_t funcId) {
16,839,998✔
341
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
16,839,998✔
342
    return false;
×
343
  }
344
  return FUNCTION_TYPE_GROUP_CONST_VALUE == funcMgtBuiltins[funcId].type;
16,839,998✔
345
}
346

347
bool fmIsElapsedFunc(int32_t funcId) {
10,635,755✔
348
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
10,635,755✔
349
    return false;
×
350
  }
351
  return FUNCTION_TYPE_ELAPSED == funcMgtBuiltins[funcId].type;
10,635,755✔
352
}
353

354
bool fmIsBlockDistFunc(int32_t funcId) {
582,955,664✔
355
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
582,955,664✔
356
    return false;
177,772✔
357
  }
358
  return FUNCTION_TYPE_BLOCK_DIST == funcMgtBuiltins[funcId].type;
582,782,384✔
359
}
360

361
bool fmIsDBUsageFunc(int32_t funcId) {
582,954,238✔
362
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
582,954,238✔
363
    return false;
184,325✔
364
  }
365
  return FUNCTION_TYPE_DB_USAGE == funcMgtBuiltins[funcId].type;
582,772,239✔
366
}
367

368
bool fmIsProcessByRowFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PROCESS_BY_ROW); }
626,756,018✔
369

370
bool fmIsIgnoreNullFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_IGNORE_NULL_FUNC); }
330,345✔
371

372
void fmFuncMgtDestroy() {
1,411,055✔
373
  void* m = gFunMgtService.pFuncNameHashTable;
1,411,055✔
374
  if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {
1,411,055✔
375
    taosHashCleanup(m);
1,411,010✔
376
  }
377
}
1,411,055✔
378

379
#ifdef BUILD_NO_CALL
380
int32_t fmSetNormalFunc(int32_t funcId, SFuncExecFuncs* pFpSet) {
381
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
382
    return TSDB_CODE_FAILED;
383
  }
384
  pFpSet->process = funcMgtBuiltins[funcId].processFunc;
385
  return TSDB_CODE_SUCCESS;
386
}
387
#endif
388

389
// function has same input/output type
390
bool fmIsSameInOutType(int32_t funcId) {
48,179,694✔
391
  bool res = false;
48,179,694✔
392
  switch (funcMgtBuiltins[funcId].type) {
48,179,694✔
393
    case FUNCTION_TYPE_MAX:
19,041,373✔
394
    case FUNCTION_TYPE_MIN:
395
    case FUNCTION_TYPE_TOP:
396
    case FUNCTION_TYPE_BOTTOM:
397
    case FUNCTION_TYPE_FIRST:
398
    case FUNCTION_TYPE_LAST:
399
    case FUNCTION_TYPE_SAMPLE:
400
    case FUNCTION_TYPE_TAIL:
401
    case FUNCTION_TYPE_UNIQUE:
402
      res = true;
19,041,373✔
403
      break;
19,041,373✔
404
    default:
29,138,321✔
405
      break;
29,138,321✔
406
  }
407
  return res;
48,179,694✔
408
}
409

410
bool fmIsConstantResFunc(SFunctionNode* pFunc) {
14,613,596✔
411
  SNode* pNode;
412
  FOREACH(pNode, pFunc->pParameterList) {
14,697,524✔
413
    if (nodeType(pNode) != QUERY_NODE_VALUE) {
156,568✔
414
      return false;
72,640✔
415
    }
416
  }
417
  return true;
14,540,956✔
418
}
419

420
bool fmIsSkipScanCheckFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SKIP_SCAN_CHECK_FUNC); }
43,370,518✔
421

422
bool fmIsPrimaryKeyFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PRIMARY_KEY_FUNC); }
507,828,921✔
423

424
bool fmIsPlaceHolderFunc(int32_t funcId) {return isSpecificClassifyFunc(funcId, FUNC_MGT_PLACE_HOLDER_FUNC); }
2,147,483,647✔
425

426
void getLastCacheDataType(SDataType* pType, int32_t pkBytes) {
×
427
  // TODO: do it later.
428
  pType->bytes = getFirstLastInfoSize(pType->bytes, pkBytes) + VARSTR_HEADER_SIZE;
×
429
  pType->type = TSDB_DATA_TYPE_BINARY;
×
430
}
×
431

432
static int32_t getFuncInfo(SFunctionNode* pFunc) {
153,947,551✔
433
  char msg[128] = {0};
153,947,551✔
434
  return fmGetFuncInfo(pFunc, msg, sizeof(msg));
153,947,551✔
435
}
436

437
int32_t createFunction(const char* pName, SNodeList* pParameterList, SFunctionNode** ppFunc) {
9,882,406✔
438
  int32_t code = nodesMakeNode(QUERY_NODE_FUNCTION, (SNode**)ppFunc);
9,882,406✔
439
  if (NULL == *ppFunc) {
9,882,406✔
440
    return code;
×
441
  }
442
  (void)snprintf((*ppFunc)->functionName, sizeof((*ppFunc)->functionName), "%s", pName);
9,882,406✔
443
  (*ppFunc)->pParameterList = pParameterList;
9,882,406✔
444
  code = getFuncInfo((*ppFunc));
9,882,406✔
445
  if (TSDB_CODE_SUCCESS != code) {
9,882,406✔
446
    (*ppFunc)->pParameterList = NULL;
×
447
    nodesDestroyNode((SNode*)*ppFunc);
×
448
    *ppFunc = NULL;
×
449
    return code;
×
450
  }
451
  return code;
9,882,406✔
452
}
453

454
static void resetOutputChangedFunc(SFunctionNode *pFunc, const SFunctionNode* pSrcFunc) {
144,064,110✔
455
  if (funcMgtBuiltins[pFunc->funcId].type == FUNCTION_TYPE_LAST_MERGE) {
144,064,110✔
456
    pFunc->node.resType = pSrcFunc->node.resType;
22,774,434✔
457
    return;
22,774,434✔
458
  }
459
}
460

461
int32_t createFunctionWithSrcFunc(const char* pName, const SFunctionNode* pSrcFunc, SNodeList* pParameterList, SFunctionNode** ppFunc) {
144,063,609✔
462
  int32_t code = nodesMakeNode(QUERY_NODE_FUNCTION, (SNode**)ppFunc);
144,063,609✔
463
  if (NULL == *ppFunc) {
144,066,858✔
464
    return code;
×
465
  }
466

467
  (*ppFunc)->hasPk = pSrcFunc->hasPk;
144,066,858✔
468
  (*ppFunc)->pkBytes = pSrcFunc->pkBytes;
144,066,858✔
469
  (*ppFunc)->pSrcFuncRef = pSrcFunc;
144,066,858✔
470

471
  (void)snprintf((*ppFunc)->functionName, sizeof((*ppFunc)->functionName), "%s", pName);
144,066,816✔
472
  (*ppFunc)->pParameterList = pParameterList;
144,066,774✔
473
  code = getFuncInfo((*ppFunc));
144,066,816✔
474
  if (TSDB_CODE_SUCCESS != code) {
144,066,308✔
475
    (*ppFunc)->pParameterList = NULL;
×
476
    nodesDestroyNode((SNode*)*ppFunc);
×
477
    *ppFunc = NULL;
×
478
    return code;
×
479
  }
480
  resetOutputChangedFunc(*ppFunc, pSrcFunc);
144,066,308✔
481
  (*ppFunc)->node.relatedTo = pSrcFunc->node.relatedTo;
144,064,935✔
482
  (*ppFunc)->node.bindExprID = pSrcFunc->node.bindExprID;
144,064,977✔
483
  return code;
144,065,019✔
484
}
485

486
static int32_t createColumnByFunc(const SFunctionNode* pFunc, SColumnNode** ppCol) {
95,876,383✔
487
  int32_t code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)ppCol);
95,876,383✔
488
  if (NULL == *ppCol) {
95,881,542✔
489
    return code;
×
490
  }
491
  tstrncpy((*ppCol)->colName, pFunc->node.aliasName, TSDB_COL_NAME_LEN);
95,881,542✔
492
  (*ppCol)->node.resType = pFunc->node.resType;
95,881,542✔
493
  return TSDB_CODE_SUCCESS;
95,881,542✔
494
}
495

496
bool fmIsDistExecFunc(int32_t funcId) {
366,928,084✔
497
  if (fmIsUserDefinedFunc(funcId)) {
366,928,084✔
498
    return false;
102,520✔
499
  }
500
  if (!fmIsVectorFunc(funcId)) {
366,851,713✔
501
    return true;
180,575✔
502
  }
503
  return (NULL != funcMgtBuiltins[funcId].pPartialFunc && NULL != funcMgtBuiltins[funcId].pMergeFunc);
366,677,675✔
504
}
505

506
static int32_t createPartialFunction(const SFunctionNode* pSrcFunc, SFunctionNode** pPartialFunc) {
48,181,093✔
507
  SNodeList* pParameterList = NULL;
48,181,093✔
508
  int32_t    code = nodesCloneList(pSrcFunc->pParameterList, &pParameterList);
48,181,093✔
509
  if (NULL == pParameterList) {
48,183,519✔
510
    return code;
×
511
  }
512
  code =
513
      createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pPartialFunc, pSrcFunc, pParameterList, pPartialFunc);
48,183,519✔
514
  if (TSDB_CODE_SUCCESS != code) {
48,181,340✔
515
    nodesDestroyList(pParameterList);
×
516
    return code;
×
517
  }
518
  (*pPartialFunc)->hasOriginalFunc = true;
48,181,594✔
519
  (*pPartialFunc)->originalFuncId = pSrcFunc->hasOriginalFunc ? pSrcFunc->originalFuncId : pSrcFunc->funcId;
48,181,594✔
520
  char name[TSDB_FUNC_NAME_LEN + TSDB_NAME_DELIMITER_LEN + TSDB_POINTER_PRINT_BYTES + 1] = {0};
48,181,594✔
521

522
  int32_t len = snprintf(name, sizeof(name), "%s.%p", (*pPartialFunc)->functionName, pSrcFunc);
48,181,552✔
523
  if (taosHashBinary(name, len, sizeof(name)) < 0) {
48,182,857✔
524
    return TSDB_CODE_FAILED;
×
525
  }
526
  tstrncpy((*pPartialFunc)->node.aliasName, name, TSDB_COL_NAME_LEN);
48,182,857✔
527
  return TSDB_CODE_SUCCESS;
48,182,857✔
528
}
529

530
static int32_t createMergeFuncPara(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,
95,876,061✔
531
                                   SNodeList** pParameterList) {
532
  SNode*  pRes = NULL;
95,876,061✔
533
  int32_t code = createColumnByFunc(pPartialFunc, (SColumnNode**)&pRes);
95,876,061✔
534
  if (TSDB_CODE_SUCCESS != code) {
95,881,822✔
535
    return code;
×
536
  }
537
  if (NULL != funcMgtBuiltins[pSrcFunc->funcId].createMergeParaFuc) {
95,881,822✔
538
    return funcMgtBuiltins[pSrcFunc->funcId].createMergeParaFuc(pSrcFunc->pParameterList, pRes, pParameterList);
817,858✔
539
  } else {
540
    return nodesListMakeStrictAppend(pParameterList, pRes);
95,064,006✔
541
  }
542
}
543

544
static int32_t createMidFunction(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,
47,694,985✔
545
                                 SFunctionNode** pMidFunc) {
546
  SNodeList*     pParameterList = NULL;
47,694,985✔
547
  SFunctionNode* pFunc = NULL;
47,694,985✔
548

549
  int32_t code = createMergeFuncPara(pSrcFunc, pPartialFunc, &pParameterList);
47,694,943✔
550
  if (TSDB_CODE_SUCCESS == code) {
47,699,694✔
551
    if(funcMgtBuiltins[pSrcFunc->funcId].pMiddleFunc != NULL){
47,699,694✔
552
      code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMiddleFunc, pSrcFunc, pParameterList, &pFunc);
5,377,513✔
553
    }else{
554
      code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pSrcFunc, pParameterList, &pFunc);
42,322,139✔
555
    }
556
  }
557
  if (TSDB_CODE_SUCCESS == code) {
47,699,304✔
558
    tstrncpy(pFunc->node.aliasName, pPartialFunc->node.aliasName, TSDB_COL_NAME_LEN);
47,699,279✔
559
  }
560

561
  if (TSDB_CODE_SUCCESS == code) {
47,699,262✔
562
    *pMidFunc = pFunc;
47,699,262✔
563
  } else {
564
    nodesDestroyList(pParameterList);
×
565
  }
566
  return code;
47,699,787✔
567
}
568

569
static int32_t createMergeFunction(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,
48,181,237✔
570
                                   SFunctionNode** pMergeFunc) {
571
  SNodeList*     pParameterList = NULL;
48,181,237✔
572
  SFunctionNode* pFunc = NULL;
48,181,237✔
573

574
  int32_t code = createMergeFuncPara(pSrcFunc, pPartialFunc, &pParameterList);
48,181,237✔
575
  if (TSDB_CODE_SUCCESS == code) {
48,183,655✔
576
    code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pSrcFunc, pParameterList, &pFunc);
48,183,655✔
577
  }
578
  if (TSDB_CODE_SUCCESS == code) {
48,184,893✔
579
    pFunc->hasOriginalFunc = true;
48,183,723✔
580
    pFunc->originalFuncId = pSrcFunc->hasOriginalFunc ? pSrcFunc->originalFuncId : pSrcFunc->funcId;
48,183,723✔
581
    // overwrite function restype set by translate function
582
    if (fmIsSameInOutType(pSrcFunc->funcId)) {
48,183,723✔
583
      pFunc->node.resType = pSrcFunc->node.resType;
19,041,763✔
584
    }
585
    tstrncpy(pFunc->node.aliasName, pSrcFunc->node.aliasName, TSDB_COL_NAME_LEN);
48,180,907✔
586
  }
587

588
  if (TSDB_CODE_SUCCESS == code) {
48,182,077✔
589
    *pMergeFunc = pFunc;
48,182,077✔
590
  } else {
591
    nodesDestroyList(pParameterList);
×
592
  }
593
  return code;
48,182,171✔
594
}
595

596
int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMidFunc,
48,179,047✔
597
                        SFunctionNode** pMergeFunc) {
598
  if (!fmIsDistExecFunc(pFunc->funcId)) {
48,179,047✔
599
    return TSDB_CODE_FAILED;
×
600
  }
601

602
  int32_t code = createPartialFunction(pFunc, pPartialFunc);
48,183,816✔
603
  if (TSDB_CODE_SUCCESS == code && pMidFunc) {
48,181,694✔
604
    code = createMidFunction(pFunc, *pPartialFunc, pMidFunc);
47,699,193✔
605
  }
606
  if (TSDB_CODE_SUCCESS == code) {
48,182,356✔
607
    code = createMergeFunction(pFunc, *pPartialFunc, pMergeFunc);
48,183,723✔
608
  }
609

610
  if (TSDB_CODE_SUCCESS != code) {
48,179,115✔
611
    nodesDestroyNode((SNode*)*pPartialFunc);
×
612
    if (pMidFunc) nodesDestroyNode((SNode*)*pMidFunc);
×
613
    nodesDestroyNode((SNode*)*pMergeFunc);
×
614
  }
615

616
  return code;
48,181,762✔
617
}
618

619
const char* fmGetFuncName(int32_t funcId) {
334,686✔
620
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
334,686✔
621
    return "invalid function";
×
622
  }
623
  return funcMgtBuiltins[funcId].name;
334,686✔
624
}
625

626
/// @param [out] pStateFunc, not changed if error occured or no need to create state func
627
/// @retval 0 for succ, otherwise err occured
628
static int32_t fmCreateStateFunc(const SFunctionNode* pFunc, SFunctionNode** pStateFunc) {
7,420✔
629
  if (funcMgtBuiltins[pFunc->funcId].pStateFunc) {
7,420✔
630
    SNodeList* pParams = NULL;
7,420✔
631
    int32_t    code = nodesCloneList(pFunc->pParameterList, &pParams);
7,420✔
632
    if (!pParams) return code;
7,420✔
633
    code = createFunction(funcMgtBuiltins[pFunc->funcId].pStateFunc, pParams, pStateFunc);
7,420✔
634
    if (TSDB_CODE_SUCCESS != code) {
7,420✔
635
      nodesDestroyList(pParams);
×
636
      return code;
×
637
    }
638
    tstrncpy((*pStateFunc)->node.aliasName, pFunc->node.aliasName, TSDB_COL_NAME_LEN);
7,420✔
639
    tstrncpy((*pStateFunc)->node.userAlias, pFunc->node.userAlias, TSDB_COL_NAME_LEN);
7,420✔
640
  }
641
  return TSDB_CODE_SUCCESS;
7,420✔
642
}
643

644
bool fmIsTSMASupportedFunc(func_id_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_TSMA_FUNC); }
1,267,336✔
645

646
bool fmIsRsmaSupportedFunc(func_id_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_RSMA_FUNC); }
134,118✔
647

648
int32_t fmCreateStateFuncs(SNodeList* pFuncs) {
3,271✔
649
  int32_t code;
650
  SNode*  pNode;
651
  char    buf[128] = {0};
3,271✔
652
  FOREACH(pNode, pFuncs) {
10,691✔
653
    SFunctionNode* pFunc = (SFunctionNode*)pNode;
7,420✔
654
    code = fmGetFuncInfo(pFunc, buf, 128);
7,420✔
655
    if (code) break;
7,420✔
656
    if (fmIsTSMASupportedFunc(pFunc->funcId)) {
7,420✔
657
      SFunctionNode* pNewFunc = NULL;
7,420✔
658
      code = fmCreateStateFunc(pFunc, &pNewFunc);
7,420✔
659
      if (code) {
7,420✔
660
        // error
661
        break;
×
662
      } else if (!pNewFunc) {
7,420✔
663
        // no need state func
664
        continue;
×
665
      } else {
666
        REPLACE_NODE(pNewFunc);
7,420✔
667
        nodesDestroyNode(pNode);
7,420✔
668
      }
669
    }
670
  }
671
  return code;
3,271✔
672
}
673

674
static int32_t fmCreateStateMergeFunc(SFunctionNode* pFunc, SFunctionNode** pStateMergeFunc) {
1,628✔
675
  if (funcMgtBuiltins[pFunc->funcId].pMergeFunc) {
1,628✔
676
    SNodeList* pParams = NULL;
1,332✔
677
    int32_t    code = nodesCloneList(pFunc->pParameterList, &pParams);
1,332✔
678
    if (!pParams) return code;
1,332✔
679
    code = createFunctionWithSrcFunc(funcMgtBuiltins[pFunc->funcId].pMergeFunc, pFunc, pParams, pStateMergeFunc);
1,332✔
680
    if (TSDB_CODE_SUCCESS != code) {
1,332✔
681
      nodesDestroyList(pParams);
×
682
      return code;
×
683
    }
684
    tstrncpy((*pStateMergeFunc)->node.aliasName, pFunc->node.aliasName, TSDB_COL_NAME_LEN);
1,332✔
685
    tstrncpy((*pStateMergeFunc)->node.userAlias, pFunc->node.userAlias, TSDB_COL_NAME_LEN);
1,332✔
686
  }
687
  return TSDB_CODE_SUCCESS;
1,628✔
688
}
689

690
int32_t fmCreateStateMergeFuncs(SNodeList* pFuncs) {
740✔
691
  int32_t code;
692
  SNode*  pNode;
693
  char    buf[128] = {0};
740✔
694
  FOREACH(pNode, pFuncs) {
2,368✔
695
    SFunctionNode* pFunc = (SFunctionNode*)pNode;
1,628✔
696
    if (fmIsTSMASupportedFunc(pFunc->funcId)) {
1,628✔
697
      SFunctionNode* pNewFunc = NULL;
1,628✔
698
      code = fmCreateStateMergeFunc(pFunc, &pNewFunc);
1,628✔
699
      if (code) {
1,628✔
700
        // error
701
        break;
×
702
      } else if (!pNewFunc) {
1,628✔
703
        // no state merge func
704
        continue;
296✔
705
      } else {
706
        REPLACE_NODE(pNewFunc);
1,332✔
707
        nodesDestroyNode(pNode);
1,332✔
708
      }
709
    }
710
  }
711
  return code;
740✔
712
}
713

714
int32_t fmGetFuncId(const char* name) {
1,884,255✔
715
  if (NULL != gFunMgtService.pFuncNameHashTable) {
1,884,255✔
716
    void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, name, strlen(name));
1,884,255✔
717
    if (NULL != pVal) {
1,884,255✔
718
      return *(int32_t*)pVal;
1,884,255✔
719
    }
720
    return -1;
×
721
  }
722
  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {
×
723
    if (0 == strcmp(funcMgtBuiltins[i].name, name)) {
×
724
      return i;
×
725
    }
726
  }
727
  return -1;
×
728
}
729

730
bool fmIsMyStateFunc(int32_t funcId, int32_t stateFuncId) {
2,091,240✔
731
  const SBuiltinFuncDefinition* pFunc = &funcMgtBuiltins[funcId];
2,091,240✔
732
  const SBuiltinFuncDefinition* pStateFunc = &funcMgtBuiltins[stateFuncId];
2,091,240✔
733
  if (!pFunc->pStateFunc) {
2,091,240✔
734
    return false;
×
735
  }
736
  if (strcmp(pFunc->pStateFunc, pStateFunc->name) == 0) return true;
2,091,240✔
737
  int32_t stateMergeFuncId = fmGetFuncId(pFunc->pStateFunc);
761,164✔
738
  if (stateMergeFuncId == -1) {
761,164✔
739
    return false;
×
740
  }
741
  const SBuiltinFuncDefinition* pStateMergeFunc = &funcMgtBuiltins[stateMergeFuncId];
761,164✔
742
  return strcmp(pStateFunc->name, pStateMergeFunc->pMergeFunc) == 0;
761,164✔
743
}
744

745
bool fmIsCountLikeFunc(int32_t funcId) {
576,654,564✔
746
  return isSpecificClassifyFunc(funcId, FUNC_MGT_COUNT_LIKE_FUNC);
576,654,564✔
747
}
748

749
bool fmIsRowTsOriginFunc(int32_t funcId) {
5,260,009✔
750
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
5,260,009✔
751
    return false;
428✔
752
  }
753
  return FUNCTION_TYPE_IROWTS_ORIGIN == funcMgtBuiltins[funcId].type;
5,259,581✔
754
}
755

756
bool fmIsGroupIdFunc(int32_t funcId) {
×
757
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
×
758
    return false;
×
759
  }
760
  return FUNCTION_TYPE_GROUP_ID == funcMgtBuiltins[funcId].type;
×
761
}
762

763
const void* fmGetStreamPesudoFuncVal(int32_t funcId, const SStreamRuntimeFuncInfo* pStreamRuntimeFuncInfo) {
296,291,866✔
764
  SSTriggerCalcParam *pParams = taosArrayGet(pStreamRuntimeFuncInfo->pStreamPesudoFuncVals, pStreamRuntimeFuncInfo->curIdx);
296,291,866✔
765
  switch (funcMgtBuiltins[funcId].type) {
296,293,542✔
766
    case FUNCTION_TYPE_TPREV_TS:
×
767
      return &pParams->prevTs;
×
768
    case FUNCTION_TYPE_TCURRENT_TS:
1,964,208✔
769
      return &pParams->currentTs;
1,964,208✔
770
    case FUNCTION_TYPE_TNEXT_TS:
×
771
      return &pParams->nextTs;
×
772
    case FUNCTION_TYPE_TWSTART:
272,577,931✔
773
      return &pParams->wstart;
272,577,931✔
774
    case FUNCTION_TYPE_TWEND:
16,797,446✔
775
      return &pParams->wend;
16,797,446✔
776
    case FUNCTION_TYPE_TWDURATION:
1,068,451✔
777
      return &pParams->wduration;
1,068,451✔
778
    case FUNCTION_TYPE_TWROWNUM:
98,366✔
779
      return &pParams->wrownum;
98,366✔
780
    case FUNCTION_TYPE_TPREV_LOCALTIME:
162,075✔
781
      return &pParams->prevLocalTime;
162,075✔
782
    case FUNCTION_TYPE_TLOCALTIME:
455,332✔
783
      return &pParams->triggerTime;
455,332✔
784
    case FUNCTION_TYPE_TNEXT_LOCALTIME:
250,280✔
785
      return &pParams->nextLocalTime;
250,280✔
786
    case FUNCTION_TYPE_TGRPID:
124,509✔
787
      return &pStreamRuntimeFuncInfo->groupId;
124,509✔
788
    case FUNCTION_TYPE_PLACEHOLDER_COLUMN:
857,764✔
789
      return pStreamRuntimeFuncInfo->pStreamPartColVals;
857,764✔
790
    case FUNCTION_TYPE_PLACEHOLDER_TBNAME:
1,939,961✔
791
      return pStreamRuntimeFuncInfo->pStreamPartColVals;
1,939,961✔
792
    case FUNCTION_TYPE_TIDLESTART:
5,832✔
793
      return &pParams->idlestart;
5,832✔
794
    case FUNCTION_TYPE_TIDLEEND:
5,832✔
795
      return &pParams->idleend;
5,832✔
796
    default:
×
797
      break;
×
798
  }
799
  return NULL;
×
800
}
801

802
bool fmIsStreamPesudoColVal(int32_t funcId) {
514,974✔
803
  return funcMgtBuiltins[funcId].type == FUNCTION_TYPE_PLACEHOLDER_COLUMN
514,974✔
804
         || funcMgtBuiltins[funcId].type == FUNCTION_TYPE_PLACEHOLDER_TBNAME;
514,974✔
805
}
806

807
int32_t fmGetStreamPesudoFuncEnv(int32_t funcId, SNodeList* pParamNodes, SFuncExecEnv *pEnv) {
6,488,552✔
808
  if (NULL == pParamNodes || pParamNodes->length < 1) {
6,488,552✔
809
    uError("invalid stream pesudo func param list %p", pParamNodes);
×
810
    return TSDB_CODE_INTERNAL_ERROR;
×
811
  }
812

813
  int32_t firstParamType = nodeType(nodesListGetNode(pParamNodes, 0));
6,489,241✔
814
  if (QUERY_NODE_VALUE != firstParamType) {
6,489,009✔
815
    uError("invalid stream pesudo func first param type %d", firstParamType);
×
816
    return TSDB_CODE_INTERNAL_ERROR;
×
817
  }
818

819
  SValueNode* pResDesc = (SValueNode*)nodesListGetNode(pParamNodes, 0);
6,489,009✔
820
  pEnv->calcMemSize = pResDesc->node.resType.bytes;
6,488,326✔
821

822
  return TSDB_CODE_SUCCESS;
6,488,861✔
823
}
824

825

826
int32_t fmSetStreamPseudoFuncParamVal(int32_t funcId, SNodeList* pParamNodes, const SStreamRuntimeFuncInfo* pStreamRuntimeInfo) {
296,238,086✔
827
  if (!pStreamRuntimeInfo) {
296,238,086✔
828
    uError("internal error, should have pVals for stream pseudo funcs");
×
829
    return TSDB_CODE_INTERNAL_ERROR;
×
830
  }
831
  int32_t code = 0;
296,238,086✔
832
  SArray *pVals1 = NULL;
296,238,086✔
833
  SNode* pFirstParam = nodesListGetNode(pParamNodes, 0);
296,238,086✔
834
  if (nodeType(pFirstParam) != QUERY_NODE_VALUE) {
296,292,483✔
835
    uError("invalid param node type: %d for func: %d", nodeType(pFirstParam), funcId);
×
836
    return TSDB_CODE_INTERNAL_ERROR;
×
837
  }
838

839
  int32_t t = funcMgtBuiltins[funcId].type;
296,321,153✔
840
  uDebug("set stream pseudo func param val, functype: %d, pStreamRuntimeInfo: %p", t, pStreamRuntimeInfo);
296,291,995✔
841
  if (FUNCTION_TYPE_TGRPID == t) {
296,316,315✔
842
    SValue v = {0};
124,509✔
843
    v.type = TSDB_DATA_TYPE_BIGINT;
124,509✔
844
    v.val = *(int64_t*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
124,509✔
845
    
846
    code = nodesSetValueNodeValue((SValueNode*)pFirstParam, VALUE_GET_DATUM(&v, pFirstParam->type));
124,509✔
847
    if (code != 0) {
124,509✔
848
      uError("failed to set value node value: %s", tstrerror(code));
×
849
      return code;
×
850
    }
851
  } else if (FUNCTION_TYPE_PLACEHOLDER_COLUMN == t) {
296,191,806✔
852
    SNode* pSecondParam = nodesListGetNode(pParamNodes, 1);
857,764✔
853
    if (nodeType(pSecondParam) != QUERY_NODE_VALUE) {
857,764✔
854
      uError("invalid param node type: %d for func: %d", nodeType(pSecondParam), funcId);
×
855
      return TSDB_CODE_INTERNAL_ERROR;
×
856
    }
857
    SArray* pVal = (SArray*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
857,764✔
858
    int32_t idx = ((SValueNode*)pSecondParam)->datum.i;
857,764✔
859
    if (idx - 1 < 0 || idx - 1 >= taosArrayGetSize(pVal)) {
857,764✔
860
      uError("invalid idx: %d for func: %d, should be in [1, %d]", idx, funcId, (int32_t)taosArrayGetSize(pVal));
×
861
      return TSDB_CODE_INTERNAL_ERROR;
×
862
    }
863
    SStreamGroupValue* pValue = taosArrayGet(pVal, idx - 1);
857,581✔
864
    if (pValue == NULL) {
857,345✔
865
      uError("invalid idx: %d for func: %d, should be in [1, %d]", idx, funcId, (int32_t)taosArrayGetSize(pVal));
×
866
      return TSDB_CODE_INTERNAL_ERROR;
×
867
    }
868
    if (!pValue->isNull){
857,345✔
869
      if (pValue->data.type != ((SValueNode*)pFirstParam)->node.resType.type){
851,123✔
870
        uError("invalid value type: %d for func: %d, should be: %d", pValue->data.type, funcId, ((SValueNode*)pFirstParam)->node.resType.type);
×
871
        return TSDB_CODE_INTERNAL_ERROR;
×
872
      }
873
      if (IS_VAR_DATA_TYPE(((SValueNode*)pFirstParam)->node.resType.type)) {
851,123✔
874
        char* tmp = taosMemoryCalloc(1, pValue->data.nData + VARSTR_HEADER_SIZE); 
584,490✔
875
        if (tmp == NULL) {
584,726✔
876
          return TSDB_CODE_OUT_OF_MEMORY;
×
877
        }
878
        taosMemoryFree(((SValueNode*)pFirstParam)->datum.p);
584,726✔
879
        ((SValueNode*)pFirstParam)->datum.p = tmp;
584,726✔
880
        memcpy(varDataVal(((SValueNode*)pFirstParam)->datum.p), pValue->data.pData, pValue->data.nData);
584,726✔
881
        varDataLen(((SValueNode*)pFirstParam)->datum.p) = pValue->data.nData;
584,726✔
882
      } else {
883
        code = nodesSetValueNodeValue((SValueNode*)pFirstParam, VALUE_GET_DATUM(&pValue->data, pValue->data.type));
266,633✔
884
      }
885
    }
886
    ((SValueNode*)pFirstParam)->isNull = pValue->isNull;
857,581✔
887
  } else if (FUNCTION_TYPE_PLACEHOLDER_TBNAME == t) {
295,334,042✔
888
    SArray* pVal = (SArray*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
1,939,412✔
889
    for (int32_t i = 0; i < taosArrayGetSize(pVal); ++i) {
2,366,478✔
890
      SStreamGroupValue* pValue = taosArrayGet(pVal, i);
2,365,652✔
891
      if (pValue != NULL && pValue->isTbname) {
2,365,929✔
892
        taosMemoryFreeClear(((SValueNode*)pFirstParam)->datum.p);
1,940,238✔
893
        ((SValueNode*)pFirstParam)->datum.p = taosMemoryCalloc(pValue->data.nData + VARSTR_HEADER_SIZE, 1);
1,939,694✔
894
        if (NULL == ((SValueNode*)pFirstParam)->datum.p ) {
1,939,145✔
895
          return terrno;
×
896
        }
897

898
        (void)memcpy(varDataVal(((SValueNode*)pFirstParam)->datum.p), pValue->data.pData, pValue->data.nData);
1,939,684✔
899
        varDataLen(((SValueNode*)pFirstParam)->datum.p) = pValue->data.nData;
1,939,971✔
900
        break;
1,940,238✔
901
      }
902
    }
903
  } else if (LIST_LENGTH(pParamNodes) == 1) {
586,714,894✔
904
    // twstart, twend
905
    const void* pVal = fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
293,369,667✔
906
    if (!pVal) {
293,390,667✔
907
      uError("failed to set stream pseudo func param val, NULL val for funcId: %d", funcId);
×
908
      return TSDB_CODE_INTERNAL_ERROR;
×
909
    }
910
    code = nodesSetValueNodeValue((SValueNode*)pFirstParam, (void*)pVal);
293,390,667✔
911
    if (code != 0) {
293,392,939✔
912
      uError("failed to set value node value: %s", tstrerror(code));
72,675✔
913
      return code;
×
914
    }
915
  } else {
916
    uError("invalid placeholder function type: %d", t);
84,493✔
917
    return TSDB_CODE_INTERNAL_ERROR;
×
918
  }
919
  
920
  return code;
296,294,737✔
921
}
922

923

924

STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc