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

taosdata / TDengine / #5007

29 Mar 2026 04:32AM UTC coverage: 72.25% (-0.02%) from 72.274%
#5007

push

travis-ci

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

253624 of 351039 relevant lines covered (72.25%)

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

44
  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {
408,228,300✔
45
    if (TSDB_CODE_SUCCESS != taosHashPut(gFunMgtService.pFuncNameHashTable, funcMgtBuiltins[i].name,
406,437,825✔
46
                                         strlen(funcMgtBuiltins[i].name), &i, sizeof(int32_t))) {
406,437,825✔
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,412,032✔
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,790,475✔
66
  (void)taosThreadOnce(&functionHashTableInit, doInitFunctionTable);
1,790,475✔
67
  return initFunctionCode;
1,790,475✔
68
}
69

70
int32_t fmGetFuncInfo(SFunctionNode* pFunc, char* pMsg, int32_t msgLen) {
806,470,685✔
71
  if (NULL != gFunMgtService.pFuncNameHashTable) {
806,470,685✔
72
    void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc->functionName, strlen(pFunc->functionName));
806,312,193✔
73
    if (NULL != pVal) {
806,340,390✔
74
      pFunc->funcId = *(int32_t*)pVal;
806,139,419✔
75
      pFunc->funcType = funcMgtBuiltins[pFunc->funcId].type;
806,138,938✔
76
      return funcMgtBuiltins[pFunc->funcId].translateFunc(pFunc, pMsg, msgLen);
806,140,155✔
77
    }
78
    return TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION;
200,971✔
79
  }
80
  for (int32_t i = 0; i < funcMgtBuiltinsNum; ++i) {
3,179,468✔
81
    if (0 == strcmp(funcMgtBuiltins[i].name, pFunc->functionName)) {
3,180,184✔
82
      pFunc->funcId = i;
170,912✔
83
      pFunc->funcType = funcMgtBuiltins[pFunc->funcId].type;
170,912✔
84
      return funcMgtBuiltins[pFunc->funcId].translateFunc(pFunc, pMsg, msgLen);
170,912✔
85
    }
86
  }
87
  return TSDB_CODE_FUNC_NOT_BUILTIN_FUNTION;
×
88
}
89

90
EFuncReturnRows fmGetFuncReturnRows(SFunctionNode* pFunc) {
17,135,221✔
91
  if (NULL != funcMgtBuiltins[pFunc->funcId].estimateReturnRowsFunc) {
17,135,221✔
92
    return funcMgtBuiltins[pFunc->funcId].estimateReturnRowsFunc(pFunc);
12,092,943✔
93
  }
94
  return (fmIsIndefiniteRowsFunc(pFunc->funcId) || fmIsMultiRowsFunc(pFunc->funcId)) ? FUNC_RETURN_ROWS_INDEFINITE
5,042,278✔
95
                                                                                     : FUNC_RETURN_ROWS_NORMAL;
5,042,278✔
96
}
97

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

102
EFunctionType fmGetFuncType(const char* pFunc) {
596,719,692✔
103
  void* pVal = taosHashGet(gFunMgtService.pFuncNameHashTable, pFunc, strlen(pFunc));
596,719,692✔
104
  if (NULL != pVal) {
596,741,243✔
105
    return funcMgtBuiltins[*(int32_t*)pVal].type;
596,518,504✔
106
  }
107
  return FUNCTION_TYPE_UDF;
222,739✔
108
}
109

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

117
EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {
39,508,870✔
118
  if (fmIsUserDefinedFunc(pFunc->funcId) || pFunc->funcId < 0 || pFunc->funcId >= funcMgtBuiltinsNum) {
39,508,870✔
119
    return FUNC_DATA_REQUIRED_DATA_LOAD;
1,005✔
120
  }
121
  if (NULL == funcMgtBuiltins[pFunc->funcId].dataRequiredFunc) {
39,514,253✔
122
    return FUNC_DATA_REQUIRED_DATA_LOAD;
×
123
  }
124
  return funcMgtBuiltins[pFunc->funcId].dataRequiredFunc(pFunc, pTimeWindow);
39,514,255✔
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;
41✔
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,056,741✔
135
    ;
136
  }
137

138
  if (funcMgtBuiltins[funcId].dynDataRequiredFunc == NULL) {
2,147,483,647✔
139
    return FUNC_DATA_REQUIRED_DATA_LOAD;
190,222,651✔
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) {
347,073,010✔
146
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
347,073,010✔
147
    return TSDB_CODE_FAILED;
×
148
  }
149
  pFpSet->getEnv = funcMgtBuiltins[funcId].getEnvFunc;
347,085,798✔
150
  pFpSet->init = funcMgtBuiltins[funcId].initFunc;
347,019,296✔
151
  pFpSet->process = funcMgtBuiltins[funcId].processFunc;
347,041,204✔
152
  pFpSet->finalize = funcMgtBuiltins[funcId].finalizeFunc;
347,027,511✔
153
  pFpSet->combine = funcMgtBuiltins[funcId].combineFunc;
347,040,704✔
154
  pFpSet->processFuncByRow = funcMgtBuiltins[funcId].processFuncByRow;
347,030,191✔
155
  pFpSet->cleanup = funcMgtBuiltins[funcId].cleanupFunc;
347,026,822✔
156
  return TSDB_CODE_SUCCESS;
347,057,357✔
157
}
158

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

174
int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet) {
1,049,809,476✔
175
  if (fmIsUserDefinedFunc(funcId) || funcId < 0 || funcId >= funcMgtBuiltinsNum) {
1,049,809,476✔
176
    return TSDB_CODE_FAILED;
36,158✔
177
  }
178
  pFpSet->process = funcMgtBuiltins[funcId].sprocessFunc;
1,050,025,644✔
179
  pFpSet->getEnv = funcMgtBuiltins[funcId].getEnvFunc;
1,049,981,640✔
180
  return TSDB_CODE_SUCCESS;
1,049,969,194✔
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,252,775,989✔
188

189
bool fmIsSelectColsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_COLS_FUNC); }
145,517,249✔
190

191
bool fmIsSelectFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SELECT_FUNC); }
1,941,514,781✔
192

193
bool fmIsTimelineFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_TIMELINE_FUNC); }
1,145,861,577✔
194

195
bool fmIsDateTimeFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_DATETIME_FUNC); }
579,528,868✔
196

197
bool fmIsPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PSEUDO_COLUMN_FUNC); }
2,144,480,391✔
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,457,526,712✔
202

203
bool fmIsWindowClauseFunc(int32_t funcId) { return fmIsAggFunc(funcId) || fmIsWindowPseudoColumnFunc(funcId); }
59,030,953✔
204

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

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

209
bool fmIsIndefiniteRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INDEFINITE_ROWS_FUNC); }
1,538,266,418✔
210

211
bool fmIsSpecialDataRequiredFunc(int32_t funcId) {
89,904,697✔
212
  return isSpecificClassifyFunc(funcId, FUNC_MGT_SPECIAL_DATA_REQUIRED);
89,904,697✔
213
}
214

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

219
bool fmIsMultiResFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_RES_FUNC); }
1,175,238,378✔
220

221
bool fmIsRepeatScanFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_REPEAT_SCAN_FUNC); }
1,157,932,725✔
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); }
579,777,550✔
226

227
bool fmIsIntervalInterpoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INTERVAL_INTERPO_FUNC); }
59,712,425✔
228

229
bool fmIsSystemInfoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SYSTEM_INFO_FUNC); }
582,692,253✔
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); }
582,430,704✔
234

235
bool fmIsMultiRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_ROWS_FUNC); }
1,155,111,220✔
236

237
static bool fmIsKeepOrderFuncId(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_KEEP_ORDER_FUNC); }
695,116,604✔
238

239
bool fmIsKeepOrderFunc(SFunctionNode* pFunc) {
701,245,287✔
240
  if (pFunc->funcType == FUNCTION_TYPE_TOP || pFunc->funcType == FUNCTION_TYPE_BOTTOM ||
701,245,287✔
241
      pFunc->funcType == FUNCTION_TYPE_SAMPLE) {
697,391,532✔
242
    if (pFunc->pParameterList != NULL && pFunc->pParameterList->length >= 2) {
6,132,966✔
243
      SNode* pParam = nodesListGetNode(pFunc->pParameterList, 1);
6,147,388✔
244
      if (pParam != NULL && nodeType(pParam) == QUERY_NODE_VALUE) {
6,147,388✔
245
        SValueNode* pConst = (SValueNode*)pParam;
6,147,388✔
246
        if (pConst->datum.i == 1) {
6,147,388✔
247
          return true;
725,862✔
248
        }
249
      }
250
    }
251
    return false;
5,413,612✔
252
  }
253
  return fmIsKeepOrderFuncId(pFunc->funcId);
695,123,885✔
254
}
255

256
bool fmIsCumulativeFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_CUMULATIVE_FUNC); }
147,805,560✔
257

258
bool fmIsForbidSysTableFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_SYSTABLE_FUNC); }
579,709,136✔
259

260
bool fmIsInterpFunc(int32_t funcId) {
1,187,673,697✔
261
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
1,187,673,697✔
262
    return false;
367,502✔
263
  }
264
  return FUNCTION_TYPE_INTERP == funcMgtBuiltins[funcId].type;
1,187,312,675✔
265
}
266

267
bool fmIsInterpPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_INTERP_PC_FUNC); }
1,168,889,057✔
268

269
bool fmIsForecastFunc(int32_t funcId) {
1,140,220,025✔
270
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
1,140,220,025✔
271
    return false;
377,637✔
272
  }
273
  return FUNCTION_TYPE_FORECAST == funcMgtBuiltins[funcId].type;
1,139,845,818✔
274
}
275

276
bool fmIsAnalysisPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_ANALYTICS_PC_FUNC); }
1,171,750,604✔
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) {
66,449,440✔
289
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
66,449,440✔
290
    return false;
×
291
  }
292
  return FUNCTION_TYPE_LAST_ROW == funcMgtBuiltins[funcId].type;
66,450,014✔
293
}
294

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

302
bool fmIsNotNullOutputFunc(int32_t funcId) {
510,199,505✔
303
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
510,199,505✔
304
    return false;
81,315✔
305
  }
306
  return FUNCTION_TYPE_LAST == funcMgtBuiltins[funcId].type ||
510,125,617✔
307
         FUNCTION_TYPE_CACHE_LAST == funcMgtBuiltins[funcId].type ||
491,001,500✔
308
         FUNCTION_TYPE_LAST_PARTIAL == funcMgtBuiltins[funcId].type ||
489,791,077✔
309
         FUNCTION_TYPE_LAST_MERGE == funcMgtBuiltins[funcId].type ||
474,429,152✔
310
         FUNCTION_TYPE_FIRST == funcMgtBuiltins[funcId].type ||
469,803,265✔
311
         FUNCTION_TYPE_FIRST_PARTIAL == funcMgtBuiltins[funcId].type ||
456,361,570✔
312
         FUNCTION_TYPE_FIRST_MERGE == funcMgtBuiltins[funcId].type ||
447,960,491✔
313
         FUNCTION_TYPE_COUNT == funcMgtBuiltins[funcId].type ||
445,473,731✔
314
         FUNCTION_TYPE_HYPERLOGLOG == funcMgtBuiltins[funcId].type ||
392,699,094✔
315
         FUNCTION_TYPE_HYPERLOGLOG_PARTIAL == funcMgtBuiltins[funcId].type ||
1,390,168,822✔
316
         FUNCTION_TYPE_HYPERLOGLOG_MERGE == funcMgtBuiltins[funcId].type;
388,935,703✔
317
}
318

319
bool fmIsSelectValueFunc(int32_t funcId) {
329,203,201✔
320
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
329,203,201✔
321
    return false;
×
322
  }
323
  return FUNCTION_TYPE_SELECT_VALUE == funcMgtBuiltins[funcId].type;
329,204,789✔
324
}
325

326
bool fmIsGroupKeyFunc(int32_t funcId) {
251,638,960✔
327
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
251,638,960✔
328
    return false;
199✔
329
  }
330
  return FUNCTION_TYPE_GROUP_KEY == funcMgtBuiltins[funcId].type;
251,638,761✔
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) {
13,453,562✔
341
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
13,453,562✔
342
    return false;
×
343
  }
344
  return FUNCTION_TYPE_GROUP_CONST_VALUE == funcMgtBuiltins[funcId].type;
13,453,562✔
345
}
346

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

354
bool fmIsBlockDistFunc(int32_t funcId) {
579,516,620✔
355
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
579,516,620✔
356
    return false;
171,787✔
357
  }
358
  return FUNCTION_TYPE_BLOCK_DIST == funcMgtBuiltins[funcId].type;
579,351,645✔
359
}
360

361
bool fmIsDBUsageFunc(int32_t funcId) {
579,518,829✔
362
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
579,518,829✔
363
    return false;
180,579✔
364
  }
365
  return FUNCTION_TYPE_DB_USAGE == funcMgtBuiltins[funcId].type;
579,342,026✔
366
}
367

368
bool fmIsProcessByRowFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PROCESS_BY_ROW); }
623,211,874✔
369

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

372
void fmFuncMgtDestroy() {
1,790,526✔
373
  void* m = gFunMgtService.pFuncNameHashTable;
1,790,526✔
374
  if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {
1,790,526✔
375
    taosHashCleanup(m);
1,790,475✔
376
  }
377
}
1,790,526✔
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) {
44,120,867✔
391
  bool res = false;
44,120,867✔
392
  switch (funcMgtBuiltins[funcId].type) {
44,120,867✔
393
    case FUNCTION_TYPE_MAX:
16,509,613✔
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;
16,509,613✔
403
      break;
16,509,613✔
404
    default:
27,611,254✔
405
      break;
27,611,254✔
406
  }
407
  return res;
44,120,867✔
408
}
409

410
bool fmIsConstantResFunc(SFunctionNode* pFunc) {
14,608,792✔
411
  SNode* pNode;
412
  FOREACH(pNode, pFunc->pParameterList) {
14,692,876✔
413
    if (nodeType(pNode) != QUERY_NODE_VALUE) {
218,995✔
414
      return false;
134,911✔
415
    }
416
  }
417
  return true;
14,473,881✔
418
}
419

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

422
bool fmIsPrimaryKeyFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_PRIMARY_KEY_FUNC); }
508,359,023✔
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) {
141,340,728✔
433
  char msg[128] = {0};
141,340,728✔
434
  return fmGetFuncInfo(pFunc, msg, sizeof(msg));
141,340,728✔
435
}
436

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

454
static void resetOutputChangedFunc(SFunctionNode *pFunc, const SFunctionNode* pSrcFunc) {
131,828,623✔
455
  if (funcMgtBuiltins[pFunc->funcId].type == FUNCTION_TYPE_LAST_MERGE) {
131,828,623✔
456
    pFunc->node.resType = pSrcFunc->node.resType;
19,789,277✔
457
    return;
19,789,277✔
458
  }
459
}
460

461
int32_t createFunctionWithSrcFunc(const char* pName, const SFunctionNode* pSrcFunc, SNodeList* pParameterList, SFunctionNode** ppFunc) {
131,828,865✔
462
  int32_t code = nodesMakeNode(QUERY_NODE_FUNCTION, (SNode**)ppFunc);
131,828,865✔
463
  if (NULL == *ppFunc) {
131,832,840✔
464
    return code;
×
465
  }
466

467
  (*ppFunc)->hasPk = pSrcFunc->hasPk;
131,832,840✔
468
  (*ppFunc)->pkBytes = pSrcFunc->pkBytes;
131,832,785✔
469
  (*ppFunc)->pSrcFuncRef = pSrcFunc;
131,832,840✔
470

471
  (void)snprintf((*ppFunc)->functionName, sizeof((*ppFunc)->functionName), "%s", pName);
131,832,785✔
472
  (*ppFunc)->pParameterList = pParameterList;
131,832,785✔
473
  code = getFuncInfo((*ppFunc));
131,832,793✔
474
  if (TSDB_CODE_SUCCESS != code) {
131,831,097✔
475
    (*ppFunc)->pParameterList = NULL;
×
476
    nodesDestroyNode((SNode*)*ppFunc);
×
477
    *ppFunc = NULL;
×
478
    return code;
×
479
  }
480
  resetOutputChangedFunc(*ppFunc, pSrcFunc);
131,831,097✔
481
  (*ppFunc)->node.relatedTo = pSrcFunc->node.relatedTo;
131,829,904✔
482
  (*ppFunc)->node.bindExprID = pSrcFunc->node.bindExprID;
131,829,904✔
483
  return code;
131,829,959✔
484
}
485

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

496
bool fmIsDistExecFunc(int32_t funcId) {
346,736,776✔
497
  if (fmIsUserDefinedFunc(funcId)) {
346,736,776✔
498
    return false;
103,306✔
499
  }
500
  if (!fmIsVectorFunc(funcId)) {
346,660,994✔
501
    return true;
188,897✔
502
  }
503
  return (NULL != funcMgtBuiltins[funcId].pPartialFunc && NULL != funcMgtBuiltins[funcId].pMergeFunc);
346,480,752✔
504
}
505

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

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

530
static int32_t createMergeFuncPara(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,
87,696,635✔
531
                                   SNodeList** pParameterList) {
532
  SNode*  pRes = NULL;
87,696,635✔
533
  int32_t code = createColumnByFunc(pPartialFunc, (SColumnNode**)&pRes);
87,696,635✔
534
  if (TSDB_CODE_SUCCESS != code) {
87,704,800✔
535
    return code;
×
536
  }
537
  if (NULL != funcMgtBuiltins[pSrcFunc->funcId].createMergeParaFuc) {
87,704,800✔
538
    return funcMgtBuiltins[pSrcFunc->funcId].createMergeParaFuc(pSrcFunc->pParameterList, pRes, pParameterList);
667,834✔
539
  } else {
540
    return nodesListMakeStrictAppend(pParameterList, pRes);
87,036,958✔
541
  }
542
}
543

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

549
  int32_t code = createMergeFuncPara(pSrcFunc, pPartialFunc, &pParameterList);
43,574,268✔
550
  if (TSDB_CODE_SUCCESS == code) {
43,580,265✔
551
    if(funcMgtBuiltins[pSrcFunc->funcId].pMiddleFunc != NULL){
43,580,466✔
552
      code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMiddleFunc, pSrcFunc, pParameterList, &pFunc);
5,324,423✔
553
    }else{
554
      code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pSrcFunc, pParameterList, &pFunc);
38,256,043✔
555
    }
556
  }
557
  if (TSDB_CODE_SUCCESS == code) {
43,578,673✔
558
    tstrncpy(pFunc->node.aliasName, pPartialFunc->node.aliasName, TSDB_COL_NAME_LEN);
43,578,874✔
559
  }
560

561
  if (TSDB_CODE_SUCCESS == code) {
43,578,673✔
562
    *pMidFunc = pFunc;
43,578,673✔
563
  } else {
564
    nodesDestroyList(pParameterList);
×
565
  }
566
  return code;
43,579,182✔
567
}
568

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

574
  int32_t code = createMergeFuncPara(pSrcFunc, pPartialFunc, &pParameterList);
44,122,940✔
575
  if (TSDB_CODE_SUCCESS == code) {
44,126,663✔
576
    code = createFunctionWithSrcFunc(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pSrcFunc, pParameterList, &pFunc);
44,126,663✔
577
  }
578
  if (TSDB_CODE_SUCCESS == code) {
44,130,598✔
579
    pFunc->hasOriginalFunc = true;
44,127,024✔
580
    pFunc->originalFuncId = pSrcFunc->hasOriginalFunc ? pSrcFunc->originalFuncId : pSrcFunc->funcId;
44,127,024✔
581
    // overwrite function restype set by translate function
582
    if (fmIsSameInOutType(pSrcFunc->funcId)) {
44,127,024✔
583
      pFunc->node.resType = pSrcFunc->node.resType;
16,510,082✔
584
    }
585
    tstrncpy(pFunc->node.aliasName, pSrcFunc->node.aliasName, TSDB_COL_NAME_LEN);
44,122,137✔
586
  }
587

588
  if (TSDB_CODE_SUCCESS == code) {
44,125,766✔
589
    *pMergeFunc = pFunc;
44,125,766✔
590
  } else {
591
    nodesDestroyList(pParameterList);
×
592
  }
593
  return code;
44,125,539✔
594
}
595

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

602
  int32_t code = createPartialFunction(pFunc, pPartialFunc);
44,126,328✔
603
  if (TSDB_CODE_SUCCESS == code && pMidFunc) {
44,124,323✔
604
    code = createMidFunction(pFunc, *pPartialFunc, pMidFunc);
43,579,611✔
605
  }
606
  if (TSDB_CODE_SUCCESS == code) {
44,124,267✔
607
    code = createMergeFunction(pFunc, *pPartialFunc, pMergeFunc);
44,126,005✔
608
  }
609

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

616
  return code;
44,123,537✔
617
}
618

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

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

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

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

674
static int32_t fmCreateStateMergeFunc(SFunctionNode* pFunc, SFunctionNode** pStateMergeFunc) {
1,557✔
675
  if (funcMgtBuiltins[pFunc->funcId].pMergeFunc) {
1,557✔
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,557✔
688
}
689

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

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

745
bool fmIsCountLikeFunc(int32_t funcId) {
543,844,722✔
746
  return isSpecificClassifyFunc(funcId, FUNC_MGT_COUNT_LIKE_FUNC);
543,844,722✔
747
}
748

749
bool fmIsRowTsOriginFunc(int32_t funcId) {
5,207,052✔
750
  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
5,207,052✔
751
    return false;
×
752
  }
753
  return FUNCTION_TYPE_IROWTS_ORIGIN == funcMgtBuiltins[funcId].type;
5,207,052✔
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) {
297,593,210✔
764
  SSTriggerCalcParam *pParams = taosArrayGet(pStreamRuntimeFuncInfo->pStreamPesudoFuncVals, pStreamRuntimeFuncInfo->curIdx);
297,593,210✔
765
  switch (funcMgtBuiltins[funcId].type) {
297,634,804✔
766
    case FUNCTION_TYPE_TPREV_TS:
×
767
      return &pParams->prevTs;
×
768
    case FUNCTION_TYPE_TCURRENT_TS:
2,586,685✔
769
      return &pParams->currentTs;
2,586,685✔
770
    case FUNCTION_TYPE_TNEXT_TS:
×
771
      return &pParams->nextTs;
×
772
    case FUNCTION_TYPE_TWSTART:
275,244,478✔
773
      return &pParams->wstart;
275,244,478✔
774
    case FUNCTION_TYPE_TWEND:
15,708,535✔
775
      return &pParams->wend;
15,708,535✔
776
    case FUNCTION_TYPE_TWDURATION:
309,835✔
777
      return &pParams->wduration;
309,835✔
778
    case FUNCTION_TYPE_TWROWNUM:
98,923✔
779
      return &pParams->wrownum;
98,923✔
780
    case FUNCTION_TYPE_TPREV_LOCALTIME:
151,540✔
781
      return &pParams->prevLocalTime;
151,540✔
782
    case FUNCTION_TYPE_TLOCALTIME:
428,793✔
783
      return &pParams->triggerTime;
428,793✔
784
    case FUNCTION_TYPE_TNEXT_LOCALTIME:
234,484✔
785
      return &pParams->nextLocalTime;
234,484✔
786
    case FUNCTION_TYPE_TGRPID:
126,610✔
787
      return &pStreamRuntimeFuncInfo->groupId;
126,610✔
788
    case FUNCTION_TYPE_PLACEHOLDER_COLUMN:
710,199✔
789
      return pStreamRuntimeFuncInfo->pStreamPartColVals;
710,199✔
790
    case FUNCTION_TYPE_PLACEHOLDER_TBNAME:
2,035,326✔
791
      return pStreamRuntimeFuncInfo->pStreamPartColVals;
2,035,326✔
792
    case FUNCTION_TYPE_TIDLESTART:
5,778✔
793
      return &pParams->idlestart;
5,778✔
794
    case FUNCTION_TYPE_TIDLEEND:
5,778✔
795
      return &pParams->idleend;
5,778✔
796
    default:
×
797
      break;
×
798
  }
799
  return NULL;
×
800
}
801

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

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

813
  int32_t firstParamType = nodeType(nodesListGetNode(pParamNodes, 0));
6,782,218✔
814
  if (QUERY_NODE_VALUE != firstParamType) {
6,782,752✔
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,782,752✔
820
  pEnv->calcMemSize = pResDesc->node.resType.bytes;
6,783,220✔
821

822
  return TSDB_CODE_SUCCESS;
6,782,452✔
823
}
824

825

826
int32_t fmSetStreamPseudoFuncParamVal(int32_t funcId, SNodeList* pParamNodes, const SStreamRuntimeFuncInfo* pStreamRuntimeInfo) {
297,552,532✔
827
  if (!pStreamRuntimeInfo) {
297,552,532✔
828
    uError("internal error, should have pVals for stream pseudo funcs");
×
829
    return TSDB_CODE_INTERNAL_ERROR;
×
830
  }
831
  int32_t code = 0;
297,552,532✔
832
  SArray *pVals1 = NULL;
297,552,532✔
833
  SNode* pFirstParam = nodesListGetNode(pParamNodes, 0);
297,552,532✔
834
  if (nodeType(pFirstParam) != QUERY_NODE_VALUE) {
297,631,725✔
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;
297,636,782✔
840
  uDebug("set stream pseudo func param val, functype: %d, pStreamRuntimeInfo: %p", t, pStreamRuntimeInfo);
297,635,363✔
841
  if (FUNCTION_TYPE_TGRPID == t) {
297,650,570✔
842
    SValue v = {0};
126,610✔
843
    v.type = TSDB_DATA_TYPE_BIGINT;
126,610✔
844
    v.val = *(int64_t*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
126,610✔
845
    
846
    code = nodesSetValueNodeValue((SValueNode*)pFirstParam, VALUE_GET_DATUM(&v, pFirstParam->type));
126,610✔
847
    if (code != 0) {
126,610✔
848
      uError("failed to set value node value: %s", tstrerror(code));
×
849
      return code;
×
850
    }
851
  } else if (FUNCTION_TYPE_PLACEHOLDER_COLUMN == t) {
297,523,960✔
852
    SNode* pSecondParam = nodesListGetNode(pParamNodes, 1);
710,199✔
853
    if (nodeType(pSecondParam) != QUERY_NODE_VALUE) {
710,199✔
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);
709,956✔
858
    int32_t idx = ((SValueNode*)pSecondParam)->datum.i;
710,199✔
859
    if (idx - 1 < 0 || idx - 1 >= taosArrayGetSize(pVal)) {
710,199✔
860
      uError("invalid idx: %d for func: %d, should be in [1, %d]", idx, funcId, (int32_t)taosArrayGetSize(pVal));
243✔
861
      return TSDB_CODE_INTERNAL_ERROR;
×
862
    }
863
    SStreamGroupValue* pValue = taosArrayGet(pVal, idx - 1);
709,956✔
864
    if (pValue == NULL) {
709,956✔
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){
709,956✔
869
      if (pValue->data.type != ((SValueNode*)pFirstParam)->node.resType.type){
705,024✔
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)) {
705,024✔
874
        char* tmp = taosMemoryCalloc(1, pValue->data.nData + VARSTR_HEADER_SIZE); 
512,974✔
875
        if (tmp == NULL) {
512,974✔
876
          return TSDB_CODE_OUT_OF_MEMORY;
×
877
        }
878
        taosMemoryFree(((SValueNode*)pFirstParam)->datum.p);
512,974✔
879
        ((SValueNode*)pFirstParam)->datum.p = tmp;
512,731✔
880
        memcpy(varDataVal(((SValueNode*)pFirstParam)->datum.p), pValue->data.pData, pValue->data.nData);
512,974✔
881
        varDataLen(((SValueNode*)pFirstParam)->datum.p) = pValue->data.nData;
512,974✔
882
      } else {
883
        code = nodesSetValueNodeValue((SValueNode*)pFirstParam, VALUE_GET_DATUM(&pValue->data, pValue->data.type));
192,050✔
884
      }
885
    }
886
    ((SValueNode*)pFirstParam)->isNull = pValue->isNull;
710,199✔
887
  } else if (FUNCTION_TYPE_PLACEHOLDER_TBNAME == t) {
296,813,761✔
888
    SArray* pVal = (SArray*)fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
2,033,957✔
889
    for (int32_t i = 0; i < taosArrayGetSize(pVal); ++i) {
2,147,271✔
890
      SStreamGroupValue* pValue = taosArrayGet(pVal, i);
2,145,884✔
891
      if (pValue != NULL && pValue->isTbname) {
2,147,271✔
892
        taosMemoryFreeClear(((SValueNode*)pFirstParam)->datum.p);
2,038,112✔
893
        ((SValueNode*)pFirstParam)->datum.p = taosMemoryCalloc(pValue->data.nData + VARSTR_HEADER_SIZE, 1);
2,036,710✔
894
        if (NULL == ((SValueNode*)pFirstParam)->datum.p ) {
2,035,882✔
895
          return terrno;
×
896
        }
897

898
        (void)memcpy(varDataVal(((SValueNode*)pFirstParam)->datum.p), pValue->data.pData, pValue->data.nData);
2,036,151✔
899
        varDataLen(((SValueNode*)pFirstParam)->datum.p) = pValue->data.nData;
2,036,991✔
900
        break;
2,036,710✔
901
      }
902
    }
903
  } else if (LIST_LENGTH(pParamNodes) == 1) {
589,537,923✔
904
    // twstart, twend
905
    const void* pVal = fmGetStreamPesudoFuncVal(funcId, pStreamRuntimeInfo);
294,758,016✔
906
    if (!pVal) {
294,770,837✔
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);
294,770,837✔
911
    if (code != 0) {
294,775,288✔
912
      uError("failed to set value node value: %s", tstrerror(code));
17,169✔
913
      return code;
×
914
    }
915
  } else {
916
    uError("invalid placeholder function type: %d", t);
47,399✔
917
    return TSDB_CODE_INTERNAL_ERROR;
×
918
  }
919
  
920
  return code;
297,638,277✔
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