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

taosdata / TDengine / #4978

06 Mar 2026 09:48AM UTC coverage: 68.439% (-0.02%) from 68.456%
#4978

push

travis-ci

web-flow
feat(TDgpt): support multiple input data columns for anomaly detection. (#34606)

0 of 93 new or added lines in 9 files covered. (0.0%)

3130 existing lines in 120 files now uncovered.

211124 of 308486 relevant lines covered (68.44%)

136029500.43 hits per line

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

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

44
  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {
302,061,430✔
45
    if (TSDB_CODE_SUCCESS != taosHashPut(gFunMgtService.pFuncNameHashTable, funcMgtBuiltins[i].name,
300,724,875✔
46
                                         strlen(funcMgtBuiltins[i].name), &i, sizeof(int32_t))) {
300,724,875✔
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);
7,955,353✔
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,336,555✔
66
  (void)taosThreadOnce(&functionHashTableInit, doInitFunctionTable);
1,336,555✔
67
  return initFunctionCode;
1,336,555✔
68
}
69

70
int32_t fmGetFuncInfo(SFunctionNode* pFunc, char* pMsg, int32_t msgLen) {
752,183,384✔
71
  if (NULL != gFunMgtService.pFuncNameHashTable) {
752,183,384✔
72
    void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc->functionName, strlen(pFunc->functionName));
752,032,776✔
73
    if (NULL != pVal) {
752,053,022✔
74
      pFunc->funcId = *(int32_t*)pVal;
751,862,790✔
75
      pFunc->funcType = funcMgtBuiltins[pFunc->funcId].type;
751,862,762✔
76
      return funcMgtBuiltins[pFunc->funcId].translateFunc(pFunc, pMsg, msgLen);
751,862,645✔
77
    }
78
    return TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION;
190,232✔
79
  }
80
  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {
3,054,676✔
81
    if (0 == strcmp(funcMgtBuiltins[i].name, pFunc->functionName)) {
3,055,144✔
82
      pFunc->funcId = i;
164,192✔
83
      pFunc->funcType = funcMgtBuiltins[pFunc->funcId].type;
164,192✔
84
      return funcMgtBuiltins[pFunc->funcId].translateFunc(pFunc, pMsg, msgLen);
164,192✔
85
    }
86
  }
UNCOV
87
  return TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION;
×
88
}
89

90
EFuncReturnRows fmGetFuncReturnRows(SFunctionNode* pFunc) {
16,540,873✔
91
  if (NULL != funcMgtBuiltins[pFunc->funcId].estimateReturnRowsFunc) {
16,540,873✔
92
    return funcMgtBuiltins[pFunc->funcId].estimateReturnRowsFunc(pFunc);
11,609,840✔
93
  }
94
  return (fmIsIndefiniteRowsFunc(pFunc->funcId) || fmIsMultiRowsFunc(pFunc->funcId)) ? FUNC_RETURN_ROWS_INDEFINITE
4,931,033✔
95
                                                                                     : FUNC_RETURN_ROWS_NORMAL;
4,931,033✔
96
}
97

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

102
EFunctionType fmGetFuncType(const char* pFunc) {
550,705,060✔
103
  void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc, strlen(pFunc));
550,705,060✔
104
  if (NULL != pVal) {
550,725,615✔
105
    return funcMgtBuiltins[*(int32_t*)pVal].type;
550,531,048✔
106
  }
107
  return FUNCTION_TYPE_UDF;
194,567✔
108
}
109

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

117
EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {
35,230,937✔
118
  if (fmIsUserDefinedFunc(pFunc->funcId) || pFunc->funcId < 0 || pFunc->funcId >= funcMgtBuiltinsNum) {
35,230,937✔
UNCOV
119
    return FUNC_DATA_REQUIRED_DATA_LOAD;
×
120
  }
121
  if (NULL == funcMgtBuiltins[pFunc->funcId].dataRequiredFunc) {
35,237,735✔
122
    return FUNC_DATA_REQUIRED_DATA_LOAD;
×
123
  }
124
  return funcMgtBuiltins[pFunc->funcId].dataRequiredFunc(pFunc, pTimeWindow);
35,237,757✔
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;
161,555✔
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;
1,514,691✔
135
    ;
136
  }
137

138
  if (funcMgtBuiltins[funcId].dynDataRequiredFunc == NULL) {
2,147,483,647✔
139
    return FUNC_DATA_REQUIRED_DATA_LOAD;
204,933,500✔
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) {
337,189,611✔
146
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
337,189,611✔
147
    return TSDB_CODE_FAILED;
31,213✔
148
  }
149
  pFpSet->getEnv = funcMgtBuiltins[funcId].getEnvFunc;
337,171,139✔
150
  pFpSet->init = funcMgtBuiltins[funcId].initFunc;
337,154,670✔
151
  pFpSet->process = funcMgtBuiltins[funcId].processFunc;
337,175,322✔
152
  pFpSet->finalize = funcMgtBuiltins[funcId].finalizeFunc;
337,174,116✔
153
  pFpSet->combine = funcMgtBuiltins[funcId].combineFunc;
337,164,962✔
154
  pFpSet->processFuncByRow = funcMgtBuiltins[funcId].processFuncByRow;
337,152,465✔
155
  pFpSet->cleanup = funcMgtBuiltins[funcId].cleanupFunc;
337,187,716✔
156
  return TSDB_CODE_SUCCESS;
337,145,485✔
157
}
158

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

174
int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet) {
1,024,160,971✔
175
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
1,024,160,971✔
176
    return TSDB_CODE_FAILED;
219,933✔
177
  }
178
  pFpSet->process = funcMgtBuiltins[funcId].sprocessFunc;
1,024,184,869✔
179
  pFpSet->getEnv = funcMgtBuiltins[funcId].getEnvFunc;
1,024,351,365✔
180
  return TSDB_CODE_SUCCESS;
1,024,408,582✔
181
}
182

183
bool fmIsAggFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_AGG_FUNC); }
2,088,230,231✔
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,168,830,282✔
188

189
bool fmIsSelectColsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_COLS_FUNC); }
134,703,738✔
190

191
bool fmIsSelectFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_FUNC); }
1,869,217,392✔
192

193
bool fmIsTimelineFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_TIMELINE_FUNC); }
1,066,012,535✔
194

195
bool fmIsDateTimeFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_DATETIME_FUNC); }
539,306,118✔
196

197
bool fmIsPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PSEUDO_COLUMN_FUNC); }
2,050,524,624✔
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,386,048,182✔
202

203
bool fmIsWindowClauseFunc(int32_t funcId) { return fmIsAggFunc(funcId) || fmIsWindowPseudoColumnFunc(funcId); }
56,475,461✔
204

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

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

209
bool fmIsIndefiniteRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INDEFINITE_ROWS_FUNC); }
1,447,080,261✔
210

211
bool fmIsSpecialDataRequiredFunc(int32_t funcId) {
81,603,623✔
212
  return isSpecificClassifyFunc(funcId, FUNC_MGT_SPECIAL_DATA_REQUIRED);
81,603,623✔
213
}
214

215
bool fmIsDynamicScanOptimizedFunc(int32_t funcId) {
39,545,768✔
216
  return isSpecificClassifyFunc(funcId, FUNC_MGT_DYNAMIC_SCAN_OPTIMIZED);
39,545,768✔
217
}
218

219
bool fmIsMultiResFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_RES_FUNC); }
1,092,992,163✔
220

221
bool fmIsRepeatScanFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_REPEAT_SCAN_FUNC); }
1,077,126,279✔
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); }
539,535,821✔
226

227
bool fmIsIntervalInterpoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INTERVAL_INTERPO_FUNC); }
58,570,968✔
228

229
bool fmIsSystemInfoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SYSTEM_INFO_FUNC); }
542,321,122✔
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); }
542,079,806✔
234

235
bool fmIsMultiRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_ROWS_FUNC); }
1,074,856,179✔
236

237
static bool fmIsKeepOrderFuncId(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_KEEP_ORDER_FUNC); }
644,505,305✔
238

239
bool fmIsKeepOrderFunc(SFunctionNode* pFunc) {
650,295,327✔
240
  if (pFunc->funcType == FUNCTION_TYPE_TOP || pFunc->funcType == FUNCTION_TYPE_BOTTOM ||
650,295,327✔
241
      pFunc->funcType == FUNCTION_TYPE_SAMPLE) {
646,668,896✔
242
    if (pFunc->pParameterList != NULL && pFunc->pParameterList->length >= 2) {
5,797,430✔
243
      SNode* pParam = nodesListGetNode(pFunc->pParameterList, 1);
5,809,758✔
244
      if (pParam != NULL && nodeType(pParam) == QUERY_NODE_VALUE) {
5,809,758✔
245
        SValueNode* pConst = (SValueNode*)pParam;
5,809,758✔
246
        if (pConst->datum.i == 1) {
5,809,758✔
247
          return true;
693,444✔
248
        }
249
      }
250
    }
251
    return false;
5,110,101✔
252
  }
253
  return fmIsKeepOrderFuncId(pFunc->funcId);
644,518,293✔
254
}
255

256
bool fmIsCumulativeFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_CUMULATIVE_FUNC); }
136,891,741✔
257

258
bool fmIsForbidSysTableFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_SYSTABLE_FUNC); }
539,483,756✔
259

260
bool fmIsInterpFunc(int32_t funcId) {
1,106,105,722✔
261
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
1,106,105,722✔
262
    return false;
354,561✔
263
  }
264
  return FUNCTION_TYPE_INTERP == funcMgtBuiltins[funcId].type;
1,105,752,583✔
265
}
266

267
bool fmIsInterpPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INTERP_PC_FUNC); }
1,087,118,487✔
268

269
bool fmIsForecastFunc(int32_t funcId) {
1,060,469,476✔
270
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
1,060,469,476✔
271
    return false;
358,302✔
272
  }
273
  return FUNCTION_TYPE_FORECAST == funcMgtBuiltins[funcId].type;
1,060,112,285✔
274
}
275

276
bool fmIsAnalysisPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_ANALYTICS_PC_FUNC); }
1,089,867,724✔
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) {
61,657,907✔
289
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
61,657,907✔
290
    return false;
×
291
  }
292
  return FUNCTION_TYPE_LAST_ROW == funcMgtBuiltins[funcId].type;
61,658,345✔
293
}
294

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

302
bool fmIsNotNullOutputFunc(int32_t funcId) {
497,400,698✔
303
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
497,400,698✔
304
    return false;
60,889✔
305
  }
306
  return FUNCTION_TYPE_LAST == funcMgtBuiltins[funcId].type ||
497,356,173✔
307
         FUNCTION_TYPE_CACHE_LAST == funcMgtBuiltins[funcId].type ||
477,539,425✔
308
         FUNCTION_TYPE_LAST_PARTIAL == funcMgtBuiltins[funcId].type ||
477,076,711✔
309
         FUNCTION_TYPE_LAST_MERGE == funcMgtBuiltins[funcId].type ||
465,889,381✔
310
         FUNCTION_TYPE_FIRST == funcMgtBuiltins[funcId].type ||
461,009,126✔
311
         FUNCTION_TYPE_FIRST_PARTIAL == funcMgtBuiltins[funcId].type ||
446,761,130✔
312
         FUNCTION_TYPE_FIRST_MERGE == funcMgtBuiltins[funcId].type ||
440,486,990✔
313
         FUNCTION_TYPE_COUNT == funcMgtBuiltins[funcId].type ||
437,890,813✔
314
         FUNCTION_TYPE_HYPERLOGLOG == funcMgtBuiltins[funcId].type ||
385,979,064✔
315
         FUNCTION_TYPE_HYPERLOGLOG_PARTIAL == funcMgtBuiltins[funcId].type ||
1,357,355,285✔
316
         FUNCTION_TYPE_HYPERLOGLOG_MERGE == funcMgtBuiltins[funcId].type;
382,447,827✔
317
}
318

319
bool fmIsSelectValueFunc(int32_t funcId) {
394,141,242✔
320
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
394,141,242✔
321
    return false;
×
322
  }
323
  return FUNCTION_TYPE_SELECT_VALUE == funcMgtBuiltins[funcId].type;
394,141,651✔
324
}
325

326
bool fmIsGroupKeyFunc(int32_t funcId) {
317,980,076✔
327
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
317,980,076✔
328
    return false;
143✔
329
  }
330
  return FUNCTION_TYPE_GROUP_KEY == funcMgtBuiltins[funcId].type;
317,979,933✔
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) {
12,341,824✔
341
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
12,341,824✔
342
    return false;
×
343
  }
344
  return FUNCTION_TYPE_GROUP_CONST_VALUE == funcMgtBuiltins[funcId].type;
12,341,824✔
345
}
346

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

354
bool fmIsBlockDistFunc(int32_t funcId) {
539,300,697✔
355
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
539,300,697✔
356
    return false;
180,784✔
357
  }
358
  return FUNCTION_TYPE_BLOCK_DIST == funcMgtBuiltins[funcId].type;
539,120,190✔
359
}
360

361
bool fmIsDBUsageFunc(int32_t funcId) {
539,299,506✔
362
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
539,299,506✔
363
    return false;
169,061✔
364
  }
365
  return FUNCTION_TYPE_DB_USAGE == funcMgtBuiltins[funcId].type;
539,131,535✔
366
}
367

368
bool fmIsProcessByRowFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PROCESS_BY_ROW); }
581,974,000✔
369

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

372
void fmFuncMgtDestroy() {
1,336,601✔
373
  void* m = gFunMgtService.pFuncNameHashTable;
1,336,601✔
374
  if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {
1,336,601✔
375
    taosHashCleanup(m);
1,336,555✔
376
  }
377
}
1,336,601✔
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) {
41,627,879✔
391
  bool res = false;
41,627,879✔
392
  switch (funcMgtBuiltins[funcId].type) {
41,627,879✔
393
    case FUNCTION_TYPE_MAX:
15,623,197✔
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;
15,623,197✔
403
      break;
15,623,197✔
404
    default:
26,004,718✔
405
      break;
26,004,718✔
406
  }
407
  return res;
41,627,915✔
408
}
409

410
bool fmIsConstantResFunc(SFunctionNode* pFunc) {
13,380,677✔
411
  SNode* pNode;
412
  FOREACH(pNode, pFunc->pParameterList) {
13,461,173✔
413
    if (nodeType(pNode) != QUERY_NODE_VALUE) {
213,667✔
414
      return false;
133,171✔
415
    }
416
  }
417
  return true;
13,247,506✔
418
}
419

420
bool fmIsSkipScanCheckFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SKIP_SCAN_CHECK_FUNC); }
39,542,178✔
421

422
bool fmIsPrimaryKeyFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PRIMARY_KEY_FUNC); }
467,749,746✔
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) {
133,426,017✔
433
  char msg[128] = {0};
133,426,017✔
434
  return fmGetFuncInfo(pFunc, msg, sizeof(msg));
133,426,089✔
435
}
436

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

454
static void resetOutputChangedFunc(SFunctionNode *pFunc, const SFunctionNode* pSrcFunc) {
124,448,666✔
455
  if (funcMgtBuiltins[pFunc->funcId].type == FUNCTION_TYPE_LAST_MERGE) {
124,448,666✔
456
    pFunc->node.resType = pSrcFunc->node.resType;
18,716,767✔
457
    return;
18,716,803✔
458
  }
459
}
460

461
int32_t createFunctionWithSrcFunc(const char* pName, const SFunctionNode* pSrcFunc, SNodeList* pParameterList, SFunctionNode** ppFunc) {
124,448,469✔
462
  int32_t code = nodesMakeNode(QUERY_NODE_FUNCTION, (SNode**)ppFunc);
124,448,469✔
463
  if (NULL == *ppFunc) {
124,453,150✔
464
    return code;
×
465
  }
466

467
  (*ppFunc)->hasPk = pSrcFunc->hasPk;
124,453,150✔
468
  (*ppFunc)->pkBytes = pSrcFunc->pkBytes;
124,453,114✔
469
  (*ppFunc)->pSrcFuncRef = pSrcFunc;
124,453,114✔
470

471
  (void)snprintf((*ppFunc)->functionName, sizeof((*ppFunc)->functionName), "%s", pName);
124,453,114✔
472
  (*ppFunc)->pParameterList = pParameterList;
124,453,150✔
473
  code = getFuncInfo((*ppFunc));
124,453,150✔
474
  if (TSDB_CODE_SUCCESS != code) {
124,450,669✔
475
    (*ppFunc)->pParameterList = NULL;
×
476
    nodesDestroyNode((SNode*)*ppFunc);
×
477
    *ppFunc = NULL;
×
478
    return code;
×
479
  }
480
  resetOutputChangedFunc(*ppFunc, pSrcFunc);
124,450,669✔
481
  (*ppFunc)->node.relatedTo = pSrcFunc->node.relatedTo;
124,450,586✔
482
  (*ppFunc)->node.bindExprID = pSrcFunc->node.bindExprID;
124,450,978✔
483
  return code;
124,451,014✔
484
}
485

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

496
bool fmIsDistExecFunc(int32_t funcId) {
327,231,189✔
497
  if (fmIsUserDefinedFunc(funcId)) {
327,231,189✔
498
    return false;
98,295✔
499
  }
500
  if (!fmIsVectorFunc(funcId)) {
327,151,797✔
501
    return true;
216,980✔
502
  }
503
  return (NULL != funcMgtBuiltins[funcId].pPartialFunc && NULL != funcMgtBuiltins[funcId].pMergeFunc);
326,943,356✔
504
}
505

506
static int32_t createPartialFunction(const SFunctionNode* pSrcFunc, SFunctionNode** pPartialFunc) {
41,628,921✔
507
  SNodeList* pParameterList = NULL;
41,628,921✔
508
  int32_t    code = nodesCloneList(pSrcFunc->pParameterList, &pParameterList);
41,628,921✔
509
  if (NULL == pParameterList) {
41,632,962✔
510
    return code;
×
511
  }
512
  code =
513
      createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pPartialFunc, pSrcFunc, pParameterList, pPartialFunc);
41,632,962✔
514
  if (TSDB_CODE_SUCCESS != code) {
41,631,422✔
515
    nodesDestroyList(pParameterList);
2,247✔
516
    return code;
×
517
  }
518
  (*pPartialFunc)->hasOriginalFunc = true;
41,629,175✔
519
  (*pPartialFunc)->originalFuncId = pSrcFunc->hasOriginalFunc ? pSrcFunc->originalFuncId : pSrcFunc->funcId;
41,629,175✔
520
  char name[TSDB_FUNC_NAME_LEN + TSDB_NAME_DELIMITER_LEN + TSDB_POINTER_PRINT_BYTES + 1] = {0};
41,629,175✔
521

522
  int32_t len = tsnprintf(name, sizeof(name), "%s.%p", (*pPartialFunc)->functionName, pSrcFunc);
41,629,211✔
523
  if (taosHashBinary(name, len) < 0) {
41,631,396✔
524
    return TSDB_CODE_FAILED;
×
525
  }
526
  tstrncpy((*pPartialFunc)->node.aliasName, name, TSDB_COL_NAME_LEN);
41,631,396✔
527
  return TSDB_CODE_SUCCESS;
41,631,360✔
528
}
529

530
static int32_t createMergeFuncPara(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,
82,812,862✔
531
                                   SNodeList** pParameterList) {
532
  SNode*  pRes = NULL;
82,812,862✔
533
  int32_t code = createColumnByFunc(pPartialFunc, (SColumnNode**)&pRes);
82,812,934✔
534
  if (TSDB_CODE_SUCCESS != code) {
82,818,174✔
535
    return code;
×
536
  }
537
  if (NULL != funcMgtBuiltins[pSrcFunc->funcId].createMergeParaFuc) {
82,818,174✔
538
    return funcMgtBuiltins[pSrcFunc->funcId].createMergeParaFuc(pSrcFunc->pParameterList, pRes, pParameterList);
657,484✔
539
  } else {
540
    return nodesListMakeStrictAppend(pParameterList, pRes);
82,160,690✔
541
  }
542
}
543

544
static int32_t createMidFunction(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,
41,182,807✔
545
                                 SFunctionNode** pMidFunc) {
546
  SNodeList*     pParameterList = NULL;
41,182,807✔
547
  SFunctionNode* pFunc = NULL;
41,182,843✔
548

549
  int32_t code = createMergeFuncPara(pSrcFunc, pPartialFunc, &pParameterList);
41,182,843✔
550
  if (TSDB_CODE_SUCCESS == code) {
41,188,431✔
551
    if(funcMgtBuiltins[pSrcFunc->funcId].pMiddleFunc != NULL){
41,188,303✔
552
      code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMiddleFunc, pSrcFunc, pParameterList, &pFunc);
4,886,153✔
553
    }else{
554
      code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pSrcFunc, pParameterList, &pFunc);
36,302,150✔
555
    }
556
  }
557
  if (TSDB_CODE_SUCCESS == code) {
41,188,120✔
558
    tstrncpy(pFunc->node.aliasName, pPartialFunc->node.aliasName, TSDB_COL_NAME_LEN);
41,188,221✔
559
  }
560

561
  if (TSDB_CODE_SUCCESS == code) {
41,188,012✔
562
    *pMidFunc = pFunc;
41,188,012✔
563
  } else {
564
    nodesDestroyList(pParameterList);
×
565
  }
566
  return code;
41,187,940✔
567
}
568

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

574
  int32_t code = createMergeFuncPara(pSrcFunc, pPartialFunc, &pParameterList);
41,629,608✔
575
  if (TSDB_CODE_SUCCESS == code) {
41,632,469✔
576
    code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pSrcFunc, pParameterList, &pFunc);
41,632,469✔
577
  }
578
  if (TSDB_CODE_SUCCESS == code) {
41,634,746✔
579
    pFunc->hasOriginalFunc = true;
41,632,744✔
580
    pFunc->originalFuncId = pSrcFunc->hasOriginalFunc ? pSrcFunc->originalFuncId : pSrcFunc->funcId;
41,632,744✔
581
    // overwrite function restype set by translate function
582
    if (fmIsSameInOutType(pSrcFunc->funcId)) {
41,632,708✔
583
      pFunc->node.resType = pSrcFunc->node.resType;
15,623,408✔
584
    }
585
    tstrncpy(pFunc->node.aliasName, pSrcFunc->node.aliasName, TSDB_COL_NAME_LEN);
41,627,991✔
586
  }
587

588
  if (TSDB_CODE_SUCCESS == code) {
41,629,993✔
589
    *pMergeFunc = pFunc;
41,629,993✔
590
  } else {
591
    nodesDestroyList(pParameterList);
×
592
  }
593
  return code;
41,631,063✔
594
}
595

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

602
  int32_t code = createPartialFunction(pFunc, pPartialFunc);
41,632,121✔
603
  if (TSDB_CODE_SUCCESS == code && pMidFunc) {
41,628,533✔
604
    code = createMidFunction(pFunc, *pPartialFunc, pMidFunc);
41,185,257✔
605
  }
606
  if (TSDB_CODE_SUCCESS == code) {
41,631,380✔
607
    code = createMergeFunction(pFunc, *pPartialFunc, pMergeFunc);
41,632,352✔
608
  }
609

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

616
  return code;
41,631,725✔
617
}
618

619
const char* fmGetFuncName(int32_t funcId) {
318,309✔
620
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
318,309✔
621
    return "invalid function";
×
622
  }
623
  return funcMgtBuiltins[funcId].name;
318,309✔
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) {
6,186✔
629
  if (funcMgtBuiltins[pFunc->funcId].pStateFunc) {
6,186✔
630
    SNodeList* pParams = NULL;
6,186✔
631
    int32_t    code = nodesCloneList(pFunc->pParameterList, &pParams);
6,186✔
632
    if (!pParams) return code;
6,186✔
633
    code = createFunction(funcMgtBuiltins[pFunc->funcId].pStateFunc, pParams, pStateFunc);
6,186✔
634
    if (TSDB_CODE_SUCCESS != code) {
6,186✔
635
      nodesDestroyList(pParams);
×
636
      return code;
×
637
    }
638
    tstrncpy((*pStateFunc)->node.aliasName, pFunc->node.aliasName, TSDB_COL_NAME_LEN);
6,186✔
639
    tstrncpy((*pStateFunc)->node.userAlias, pFunc->node.userAlias, TSDB_COL_NAME_LEN);
6,186✔
640
  }
641
  return TSDB_CODE_SUCCESS;
6,186✔
642
}
643

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

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

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

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

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

714
int32_t fmGetFuncId(const char* name) {
1,845,299✔
715
  if (NULL != gFunMgtService.pFuncNameHashTable) {
1,845,299✔
716
    void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, name, strlen(name));
1,845,299✔
717
    if (NULL != pVal) {
1,845,299✔
718
      return *(int32_t*)pVal;
1,845,299✔
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) {
1,906,821✔
731
  const SBuiltinFuncDefinition* pFunc = &funcMgtBuiltins[funcId];
1,906,821✔
732
  const SBuiltinFuncDefinition* pStateFunc = &funcMgtBuiltins[stateFuncId];
1,906,821✔
733
  if (!pFunc->pStateFunc) {
1,906,821✔
734
    return false;
×
735
  }
736
  if (strcmp(pFunc->pStateFunc, pStateFunc->name) == 0) return true;
1,906,821✔
737
  int32_t stateMergeFuncId = fmGetFuncId(pFunc->pStateFunc);
696,787✔
738
  if (stateMergeFuncId == -1) {
696,787✔
739
    return false;
×
740
  }
741
  const SBuiltinFuncDefinition* pStateMergeFunc = &funcMgtBuiltins[stateMergeFuncId];
696,787✔
742
  return strcmp(pStateFunc->name, pStateMergeFunc->pMergeFunc) == 0;
696,787✔
743
}
744

745
bool fmIsCountLikeFunc(int32_t funcId) {
518,662,042✔
746
  return isSpecificClassifyFunc(funcId, FUNC_MGT_COUNT_LIKE_FUNC);
518,662,042✔
747
}
748

749
bool fmIsRowTsOriginFunc(int32_t funcId) {
5,004,153✔
750
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
5,004,153✔
751
    return false;
×
752
  }
753
  return FUNCTION_TYPE_IROWTS_ORIGIN == funcMgtBuiltins[funcId].type;
5,004,153✔
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) {
41,870,545✔
764
  SSTriggerCalcParam *pParams = taosArrayGet(pStreamRuntimeFuncInfo->pStreamPesudoFuncVals, pStreamRuntimeFuncInfo->curIdx);
41,870,545✔
765
  switch (funcMgtBuiltins[funcId].type) {
41,873,520✔
766
    case FUNCTION_TYPE_TPREV_TS:
×
767
      return &pParams->prevTs;
×
768
    case FUNCTION_TYPE_TCURRENT_TS:
2,101,390✔
769
      return &pParams->currentTs;
2,101,390✔
770
    case FUNCTION_TYPE_TNEXT_TS:
×
771
      return &pParams->nextTs;
×
772
    case FUNCTION_TYPE_TWSTART:
20,072,432✔
773
      return &pParams->wstart;
20,072,432✔
774
    case FUNCTION_TYPE_TWEND:
15,458,614✔
775
      return &pParams->wend;
15,458,614✔
776
    case FUNCTION_TYPE_TWDURATION:
953,932✔
777
      return &pParams->wduration;
953,932✔
778
    case FUNCTION_TYPE_TWROWNUM:
95,868✔
779
      return &pParams->wrownum;
95,868✔
780
    case FUNCTION_TYPE_TPREV_LOCALTIME:
88,597✔
781
      return &pParams->prevLocalTime;
88,597✔
782
    case FUNCTION_TYPE_TLOCALTIME:
268,582✔
783
      return &pParams->triggerTime;
268,582✔
784
    case FUNCTION_TYPE_TNEXT_LOCALTIME:
140,915✔
785
      return &pParams->nextLocalTime;
140,915✔
786
    case FUNCTION_TYPE_TGRPID:
113,964✔
787
      return &pStreamRuntimeFuncInfo->groupId;
113,964✔
788
    case FUNCTION_TYPE_PLACEHOLDER_COLUMN:
722,875✔
789
      return pStreamRuntimeFuncInfo->pStreamPartColVals;
722,875✔
790
    case FUNCTION_TYPE_PLACEHOLDER_TBNAME:
1,855,525✔
791
      return pStreamRuntimeFuncInfo->pStreamPartColVals;
1,855,525✔
792
    default:
×
793
      break;
×
794
  }
795
  return NULL;
×
796
}
797

798
bool fmIsStreamPesudoColVal(int32_t funcId) {
450,908✔
799
  return funcMgtBuiltins[funcId].type == FUNCTION_TYPE_PLACEHOLDER_COLUMN
450,908✔
800
         || funcMgtBuiltins[funcId].type == FUNCTION_TYPE_PLACEHOLDER_TBNAME;
450,908✔
801
}
802

803
int32_t fmGetStreamPesudoFuncEnv(int32_t funcId, SNodeList* pParamNodes, SFuncExecEnv *pEnv) {
6,641,971✔
804
  if (NULL == pParamNodes || pParamNodes->length < 1) {
6,641,971✔
805
    uError("invalid stream pesudo func param list %p", pParamNodes);
216✔
806
    return TSDB_CODE_INTERNAL_ERROR;
×
807
  }
808

809
  int32_t firstParamType = nodeType(nodesListGetNode(pParamNodes, 0));
6,641,971✔
810
  if (QUERY_NODE_VALUE != firstParamType) {
6,642,146✔
811
    uError("invalid stream pesudo func first param type %d", firstParamType);
×
812
    return TSDB_CODE_INTERNAL_ERROR;
×
813
  }
814

815
  SValueNode* pResDesc = (SValueNode*)nodesListGetNode(pParamNodes, 0);
6,642,146✔
816
  pEnv->calcMemSize = pResDesc->node.resType.bytes;
6,642,146✔
817

818
  return TSDB_CODE_SUCCESS;
6,641,930✔
819
}
820

821

822
int32_t fmSetStreamPseudoFuncParamVal(int32_t funcId, SNodeList* pParamNodes, const SStreamRuntimeFuncInfo* pStreamRuntimeInfo) {
41,867,101✔
823
  if (!pStreamRuntimeInfo) {
41,867,101✔
824
    uError("internal error, should have pVals for stream pseudo funcs");
×
825
    return TSDB_CODE_INTERNAL_ERROR;
×
826
  }
827
  int32_t code = 0;
41,867,101✔
828
  SArray *pVals1 = NULL;
41,867,101✔
829
  SNode* pFirstParam = nodesListGetNode(pParamNodes, 0);
41,867,101✔
830
  if (nodeType(pFirstParam) != QUERY_NODE_VALUE) {
41,866,700✔
831
    uError("invalid param node type: %d for func: %d", nodeType(pFirstParam), funcId);
×
832
    return TSDB_CODE_INTERNAL_ERROR;
×
833
  }
834

835
  int32_t t = funcMgtBuiltins[funcId].type;
41,865,921✔
836
  uDebug("set stream pseudo func param val, functype: %d, pStreamRuntimeInfo: %p", t, pStreamRuntimeInfo);
41,865,220✔
837
  if (FUNCTION_TYPE_TGRPID == t) {
41,871,362✔
838
    SValue v = {0};
114,180✔
839
    v.type = TSDB_DATA_TYPE_BIGINT;
114,180✔
840
    v.val = *(int64_t*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
114,180✔
841
    
842
    code = nodesSetValueNodeValue((SValueNode*)pFirstParam, VALUE_GET_DATUM(&v, pFirstParam->type));
114,180✔
843
    if (code != 0) {
114,594✔
844
      uError("failed to set value node value: %s", tstrerror(code));
×
845
      return code;
×
846
    }
847
  } else if (FUNCTION_TYPE_PLACEHOLDER_COLUMN == t) {
41,757,182✔
848
    SNode* pSecondParam = nodesListGetNode(pParamNodes, 1);
722,519✔
849
    if (nodeType(pSecondParam) != QUERY_NODE_VALUE) {
722,875✔
850
      uError("invalid param node type: %d for func: %d", nodeType(pSecondParam), funcId);
×
851
      return TSDB_CODE_INTERNAL_ERROR;
×
852
    }
853
    SArray* pVal = (SArray*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
722,875✔
854
    int32_t idx = ((SValueNode*)pSecondParam)->datum.i;
722,875✔
855
    if (idx - 1 < 0 || idx - 1 >= taosArrayGetSize(pVal)) {
722,875✔
856
      uError("invalid idx: %d for func: %d, should be in [1, %d]", idx, funcId, (int32_t)taosArrayGetSize(pVal));
223✔
857
      return TSDB_CODE_INTERNAL_ERROR;
×
858
    }
859
    SStreamGroupValue* pValue = taosArrayGet(pVal, idx - 1);
722,652✔
860
    if (pValue == NULL) {
722,875✔
861
      uError("invalid idx: %d for func: %d, should be in [1, %d]", idx, funcId, (int32_t)taosArrayGetSize(pVal));
×
862
      return TSDB_CODE_INTERNAL_ERROR;
×
863
    }
864
    if (!pValue->isNull){
722,875✔
865
      if (pValue->data.type != ((SValueNode*)pFirstParam)->node.resType.type){
716,855✔
866
        uError("invalid value type: %d for func: %d, should be: %d", pValue->data.type, funcId, ((SValueNode*)pFirstParam)->node.resType.type);
×
867
        return TSDB_CODE_INTERNAL_ERROR;
×
868
      }
869
      if (IS_VAR_DATA_TYPE(((SValueNode*)pFirstParam)->node.resType.type)) {
716,632✔
870
        char* tmp = taosMemoryCalloc(1, pValue->data.nData + VARSTR_HEADER_SIZE); 
512,105✔
871
        if (tmp == NULL) {
511,973✔
872
          return TSDB_CODE_OUT_OF_MEMORY;
×
873
        }
874
        taosMemoryFree(((SValueNode*)pFirstParam)->datum.p);
511,973✔
875
        ((SValueNode*)pFirstParam)->datum.p = tmp;
511,658✔
876
        memcpy(varDataVal(((SValueNode*)pFirstParam)->datum.p), pValue->data.pData, pValue->data.nData);
511,658✔
877
        varDataLen(((SValueNode*)pFirstParam)->datum.p) = pValue->data.nData;
511,658✔
878
      } else {
879
        code = nodesSetValueNodeValue((SValueNode*)pFirstParam, VALUE_GET_DATUM(&pValue->data, pValue->data.type));
204,750✔
880
      }
881
    }
882
    ((SValueNode*)pFirstParam)->isNull = pValue->isNull;
722,875✔
883
  } else if (FUNCTION_TYPE_PLACEHOLDER_TBNAME == t) {
41,034,663✔
884
    SArray* pVal = (SArray*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
1,856,039✔
885
    for (int32_t i = 0; i < taosArrayGetSize(pVal); ++i) {
2,236,454✔
886
      SStreamGroupValue* pValue = taosArrayGet(pVal, i);
2,236,199✔
887
      if (pValue != NULL && pValue->isTbname) {
2,236,454✔
888
        taosMemoryFreeClear(((SValueNode*)pFirstParam)->datum.p);
1,855,780✔
889
        ((SValueNode*)pFirstParam)->datum.p = taosMemoryCalloc(pValue->data.nData + VARSTR_HEADER_SIZE, 1);
1,855,523✔
890
        if (NULL == ((SValueNode*)pFirstParam)->datum.p ) {
1,856,039✔
891
          return terrno;
×
892
        }
893

894
        (void)memcpy(varDataVal(((SValueNode*)pFirstParam)->datum.p), pValue->data.pData, pValue->data.nData);
1,855,782✔
895
        varDataLen(((SValueNode*)pFirstParam)->datum.p) = pValue->data.nData;
1,856,039✔
896
        break;
1,855,782✔
897
      }
898
    }
899
  } else if (LIST_LENGTH(pParamNodes) == 1) {
78,357,422✔
900
    // twstart, twend
901
    const void* pVal = fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
39,178,334✔
902
    if (!pVal) {
39,179,446✔
903
      uError("failed to set stream pseudo func param val, NULL val for funcId: %d", funcId);
×
904
      return TSDB_CODE_INTERNAL_ERROR;
×
905
    }
906
    code = nodesSetValueNodeValue((SValueNode*)pFirstParam, (void*)pVal);
39,179,446✔
907
    if (code != 0) {
39,179,154✔
908
      uError("failed to set value node value: %s", tstrerror(code));
356✔
909
      return code;
×
910
    }
911
  } else {
912
    uError("invalid placeholder function type: %d", t);
480✔
913
    return TSDB_CODE_INTERNAL_ERROR;
×
914
  }
915
  
916
  return code;
41,871,183✔
917
}
918

919

920

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