#5059

Committed 17 May 2026 01:15AM UTC coverage: 73.443% (+0.06%) from 73.387%

Build # #5059

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat (TDgpt): Dynamic Model Synchronization Enhancements (#35344)

* refactor: do some internal refactor.

* fix: fix multiprocess sync issue.

* feat: add dynamic anomaly detection and forecasting services

* fix: log error message for undeploying model in exception handling

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix: handle undeploy when model exists only on disk

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/286aafa0-c3ce-4c27-b803-2707571e9dc1

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: guard dynamic registry concurrent access

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: tighten service list locking scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: restore prophet support and update tests per review feedback

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: improve test name and move copy inside lock scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* Potential fix for pull request finding

Co-au... (continued)

Coverage Stats

281870 of 383795 relevant lines covered (73.44%)

135516561.93 hits per line

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

57.14

/include/common/tcommon.h

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

#ifndef _TD_COMMON_DEF_H_
#define _TD_COMMON_DEF_H_

#include "tarray.h"
#include "tmsg.h"
#include "tvariant.h"

#ifdef __cplusplus
extern "C" {
#endif

// clang-format off
#define IS_META_MSG(x) ( \
     x == TDMT_VND_CREATE_STB     \
  || x == TDMT_VND_ALTER_STB      \
  || x == TDMT_VND_DROP_STB       \
  || x == TDMT_VND_CREATE_TABLE   \
  || x == TDMT_VND_ALTER_TABLE    \
  || x == TDMT_VND_DROP_TABLE     \
  || x == TDMT_VND_DELETE         \
)
// clang-format on

typedef bool (*state_key_cmpr_fn)(void* pKey1, void* pKey2);

typedef struct STableKeyInfo {
  uint64_t uid;
  uint64_t groupId;
  uint64_t baseGId;
} STableKeyInfo;

typedef struct SWinKey {
  uint64_t groupId;
  TSKEY    ts;
  int32_t  numInGroup;
} SWinKey;

typedef struct SSessionKey {
  STimeWindow win;
  uint64_t    groupId;
} SSessionKey;

typedef int64_t COUNT_TYPE;

typedef struct SVersionRange {
  int64_t minVer;
  int64_t maxVer;
} SVersionRange;

static inline int winKeyCmprImpl(const void* pKey1, const void* pKey2) {
  SWinKey* pWin1 = (SWinKey*)pKey1;
  SWinKey* pWin2 = (SWinKey*)pKey2;

  if (pWin1->groupId > pWin2->groupId) {
    return 1;
  } else if (pWin1->groupId < pWin2->groupId) {
    return -1;
  }

  if (pWin1->ts > pWin2->ts) {
    return 1;
  } else if (pWin1->ts < pWin2->ts) {
    return -1;
  }

  return 0;
}

static inline int winKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
  return winKeyCmprImpl(pKey1, pKey2);
}

typedef struct {
  uint64_t groupId;
  TSKEY    ts;
  int32_t  exprIdx;
} STupleKey;

static inline int STupleKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
  STupleKey* pTuple1 = (STupleKey*)pKey1;
  STupleKey* pTuple2 = (STupleKey*)pKey2;

  if (pTuple1->groupId > pTuple2->groupId) {
    return 1;
  } else if (pTuple1->groupId < pTuple2->groupId) {
    return -1;
  }

  if (pTuple1->ts > pTuple2->ts) {
    return 1;
  } else if (pTuple1->ts < pTuple2->ts) {
    return -1;
  }

  if (pTuple1->exprIdx > pTuple2->exprIdx) {
    return 1;
  } else if (pTuple1->exprIdx < pTuple2->exprIdx) {
    return -1;
  }

  return 0;
}

enum {
  TMQ_MSG_TYPE__POLL_DATA_RSP = 0,
  TMQ_MSG_TYPE__POLL_META_RSP,
  TMQ_MSG_TYPE__EP_RSP,
  TMQ_MSG_TYPE__POLL_DATA_META_RSP,
  TMQ_MSG_TYPE__WALINFO_RSP,
  TMQ_MSG_TYPE__POLL_BATCH_META_RSP,
  TMQ_MSG_TYPE__POLL_RAW_DATA_RSP,
};

static const char* const tmqMsgTypeStr[] = {
    "data", "meta", "ask ep", "meta data", "wal info", "batch meta", "raw data"
};

enum {
  STREAM_INPUT__DATA_SUBMIT = 1,
  STREAM_INPUT__DATA_BLOCK,
  STREAM_INPUT__MERGED_SUBMIT,
  STREAM_INPUT__RECALCULATE,
  STREAM_INPUT__DATA_RETRIEVE,
  STREAM_INPUT__GET_RES,
  STREAM_INPUT__CHECKPOINT,
  STREAM_INPUT__CHECKPOINT_TRIGGER,
  STREAM_INPUT__TRANS_STATE,
  STREAM_INPUT__REF_DATA_BLOCK,
  STREAM_INPUT__DESTROY,
};

#pragma pack(push, 1)
typedef struct SColumnDataAgg {
  int32_t colId;
  int16_t numOfNull;
  union {
    struct {
      int64_t sum;
      int64_t max;
      int64_t min;
    };
    struct {
      uint64_t decimal128Sum[2];
      uint64_t decimal128Max[2];
      uint64_t decimal128Min[2];
      uint8_t  overflow;
    };
  };
} SColumnDataAgg;
#pragma pack(pop)

#define DECIMAL_AGG_FLAG 0x80000000

#define COL_AGG_GET_SUM_PTR(pAggs, dataType) \
  (!IS_DECIMAL_TYPE(dataType) ? (void*)&pAggs->sum : (void*)pAggs->decimal128Sum)

#define COL_AGG_GET_MAX_PTR(pAggs, dataType) \
  (!IS_DECIMAL_TYPE(dataType) ? (void*)&pAggs->max : (void*)pAggs->decimal128Max)

#define COL_AGG_GET_MIN_PTR(pAggs, dataType) \
  (!IS_DECIMAL_TYPE(dataType) ? (void*)&pAggs->min : (void*)pAggs->decimal128Min)

typedef struct SBlockID {
  // The uid of table, from which current data block comes. And it is always 0, if current block is the
  // result of calculation.
  uint64_t uid;

  // Block id, acquired and assigned from executor, which created according to the hysical planner. Block id is used
  // to mark the stage of exec task.
  int64_t  blockId;

  // Generated by group/partition by [value|tags]. Created and assigned by table-scan operator, group-by operator,
  // and partition by operator.
  uint64_t groupId;

  // Base group id for stream group generated by trigger
  uint64_t baseGId;
} SBlockID;

typedef struct SPkInfo {
  int8_t  type;
  int32_t bytes;
  union {
    int64_t  val;
    uint8_t* pData;
  } skey;
  union {
    int64_t  val;
    uint8_t* pData;
  } ekey;
} SPkInfo;

typedef struct SDataBlockInfo {
  STimeWindow window;
  int32_t     rowSize;
  uint32_t    capacity;
  int64_t     rows;  // todo hide this attribute
  SBlockID    id;
  int16_t     hasVarCol;
  int16_t     dataLoad;  // denote if the data is loaded or not
  uint8_t     scanFlag;
  bool        blankFill;
  SValue      pks[2];

  // TODO: optimize and remove following
  int64_t     version;    // used for stream, and need serialization
  int32_t     childId;    // used for stream, do not serialize
  STimeWindow calWin;     // used for stream, do not serialize
  TSKEY       watermark;  // used for stream

  char parTbName[TSDB_TABLE_NAME_LEN];  // used for stream partition
} SDataBlockInfo;

typedef struct SSDataBlock {
  SColumnDataAgg* pBlockAgg;
  SArray*         pDataBlock;  // SArray<SColumnInfoData>
  SDataBlockInfo  info;
} SSDataBlock;

typedef struct SVarColAttr {
  int32_t* offset;    // start position for each entry in the list
  uint32_t length;    // used buffer size that contain the valid data
  uint32_t allocLen;  // allocated buffer size
} SVarColAttr;

// pBlockAgg->numOfNull == info.rows, all data are null
// pBlockAgg->numOfNull == 0, no data are null.
typedef struct SColumnInfoData {
  char* pData;  // the corresponding block data in memory
  union {
    char*       nullbitmap;  // bitmap, one bit for each item in the list
    SVarColAttr varmeta;
  };
  SColumnInfo info;        // column info
  bool        hasNull;     // if current column data has null value.
  bool        reassigned;  // if current column data is reassigned.
} SColumnInfoData;

typedef struct SQueryTableDataCond {
  uint64_t     suid;
  int32_t      order;  // desc|asc order to iterate the data block
  int32_t      numOfCols;
  SColumnInfo* colList;
  int32_t*     pSlotList;  // the column output destation slot, and it may be null
  int32_t      type;       // data block load type:
  bool         skipRollup;
  STimeWindow  twindows;
  STimeWindow  extTwindows[2];
  int64_t      startVersion;
  int64_t      endVersion;
  bool         notLoadData;  // response the actual data, not only the rows in the attribute of info.row of ssdatablock
  bool         cacheSttStatis;
} SQueryTableDataCond;

int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock);
void*   tDecodeDataBlock(const void* buf, SSDataBlock* pBlock);

void colDataDestroy(SColumnInfoData* pColData);

//======================================================================================================================
// the following structure shared by parser and executor
typedef struct SColumn {
  union {
    uint64_t uid;
    int64_t  dataBlockId;
  };

  int16_t colId;
  int16_t slotId;

  char    name[TSDB_COL_NAME_LEN];
  int16_t colType;  // column type: normal column, tag, or window column
  int16_t type;
  int32_t bytes;
  uint8_t precision;
  uint8_t scale;
} SColumn;

typedef struct STableBlockDistInfo {
  uint32_t rowSize;
  uint16_t numOfFiles;
  uint32_t numOfTables;
  uint32_t numOfBlocks;
  uint64_t totalSize;
  uint64_t totalRows;
  int32_t  maxRows;
  int32_t  minRows;
  int32_t  defMinRows;
  int32_t  defMaxRows;
  int32_t  firstSeekTimeUs;
  uint32_t numOfInmemRows;
  uint32_t numOfSttRows;
  uint32_t numOfVgroups;
  int32_t  blockRowsHisto[20];
  int32_t  blockRowsHistoFixed[8];  // buckets: ≤64, ≤128, ≤256, ≤512, ≤1024, ≤2048, ≤4096, >4096
} STableBlockDistInfo;

int32_t tSerializeBlockDistInfo(void* buf, int32_t bufLen, const STableBlockDistInfo* pInfo);
int32_t tDeserializeBlockDistInfo(void* buf, int32_t bufLen, STableBlockDistInfo* pInfo);

typedef struct SDBBlockUsageInfo {
  uint64_t dataInDiskSize;
  uint64_t walInDiskSize;
  uint64_t rawDataSize;
} SDBBlockUsageInfo;

int32_t tSerializeBlockDbUsage(void* buf, int32_t bufLen, const SDBBlockUsageInfo* pInfo);
int32_t tDeserializeBlockDbUsage(void* buf, int32_t bufLen, SDBBlockUsageInfo* pInfo);

enum {
  FUNC_PARAM_TYPE_VALUE = 0x1,
  FUNC_PARAM_TYPE_COLUMN = 0x2,
};

typedef struct SFunctParam {
  int32_t  type;
  SColumn* pCol;
  SVariant param;
} SFunctParam;

// the structure for sql function in select clause
typedef struct SResSchame {
  int8_t  type;
  int32_t slotId;
  int32_t bytes;
  int32_t precision;
  int32_t scale;
  char    name[TSDB_COL_NAME_LEN];
} SResSchema;

typedef struct SAggSupporter  SAggSupporter;
typedef struct SExprSupp      SExprSupp;
typedef struct SGroupResInfo  SGroupResInfo;
typedef struct SResultRow     SResultRow;
typedef struct SResultRowInfo SResultRowInfo;
typedef struct SExecTaskInfo  SExecTaskInfo;
typedef struct SRollupCtx {
  void*           pTsdb;     // STsdb*
  void*           pTargets;  // SNodeList*
  void*           pBuf;
  SExprSupp*      exprSup;
  SAggSupporter*  aggSup;
  SResultRow*     resultRow;
  SResultRowInfo* resultRowInfo;
  SGroupResInfo*  pGroupResInfo;
  SExecTaskInfo*  pTaskInfo;
  SSDataBlock*    pInputBlock;  // input data block for rollup
  SSDataBlock*    pResBlock;    // result data block for rollup
  SArray*         pColValArr;   // used the generate the aggregate row
  int32_t         rowSize;
  int32_t         maxBufRows;    // max buffer rows for aggregation
  int64_t         winTotalRows;  // number of total rows for current window
  int64_t         winStartTs;    // start timestamp of current window
} SRollupCtx;

typedef struct {
  const char* key;
  size_t      keyLen;
  uint8_t     type;
  union {
    const char* value;
    int64_t     i;
    uint64_t    u;
    double      d;
    float       f;
  };
  size_t length;
  bool   keyEscaped;
  bool   valueEscaped;
} SSmlKv;

#define QUERY_ASC_FORWARD_STEP  1
#define QUERY_DESC_FORWARD_STEP -1

#define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) != TSDB_ORDER_DESC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)

#define SORT_QSORT_T              0x1
#define SORT_SPILLED_MERGE_SORT_T 0x2
typedef struct SSortExecInfo {
  int32_t sortMethod;
  int32_t sortBuffer;
  int32_t loops;       // loop count
  int32_t writeBytes;  // write io bytes
  int32_t readBytes;   // read io bytes
} SSortExecInfo;

typedef struct SExchangeExplainInfo {
  int8_t   mode;          // 0: concurrent, 1: sequential
  int32_t  numSources;
  double   avgFetchTimes;
  uint64_t maxFetchTimes;
  double   avgFetchRows;
  uint64_t maxFetchRows;
  double   avgFetchCost;  // us
  int64_t  maxFetchCost;  // us
} SExchangeExplainInfo;

typedef struct SNonSortExecInfo {
  int32_t blkNums;
} SNonSortExecInfo;

typedef struct STUidTagInfo {
  char*    name;
  uint64_t uid;
  void*    pTagVal;
} STUidTagInfo;

// stream special block column

#define START_TS_COLUMN_INDEX           0
#define END_TS_COLUMN_INDEX             1
#define UID_COLUMN_INDEX                2
#define GROUPID_COLUMN_INDEX            3
#define CALCULATE_START_TS_COLUMN_INDEX 4
#define CALCULATE_END_TS_COLUMN_INDEX   5
#define TABLE_NAME_COLUMN_INDEX         6
#define PRIMARY_KEY_COLUMN_INDEX        7

//steam get result block column
#define DATA_TS_COLUMN_INDEX            0
#define DATA_VERSION_COLUMN_INDEX       1

// stream create table block column
#define UD_TABLE_NAME_COLUMN_INDEX 0
#define UD_GROUPID_COLUMN_INDEX    1
#define UD_TAG_COLUMN_INDEX        2

// stream notify event block column
#define NOTIFY_EVENT_STR_COLUMN_INDEX 0

int32_t taosGenCrashJsonMsg(int signum, char** pMsg, int64_t clusterId, int64_t startTime);

#define SHOW_VAR_PRIV_SYSTEM   0x01
#define SHOW_VAR_PRIV_SECURITY 0x02
#define SHOW_VAR_PRIV_AUDIT    0x04
#define SHOW_VAR_PRIV_DEBUG    0x08
#define SHOW_VAR_PRIV_ALL (SHOW_VAR_PRIV_SYSTEM | SHOW_VAR_PRIV_SECURITY | SHOW_VAR_PRIV_AUDIT | SHOW_VAR_PRIV_DEBUG)

int32_t dumpConfToDataBlock(SSDataBlock* pBlock, int32_t startCol, char* likePattern, uint8_t showPrivMask);

#define TSMA_RES_STB_POSTFIX          "_tsma_res_stb_"
#define MD5_OUTPUT_LEN                32
#define TSMA_RES_STB_EXTRA_COLUMN_NUM 4  // 3 columns: _wstart, _wend, _wduration, 1 tag: tbname

static inline bool isTsmaResSTb(const char* stbName) {
  static bool showTsmaTables = true;
  if (showTsmaTables) return false;
  const char* pos = strstr(stbName, TSMA_RES_STB_POSTFIX);
  if (pos && strlen(stbName) == (pos - stbName) + strlen(TSMA_RES_STB_POSTFIX)) {
    return true;
  }
  return false;
}

static inline STypeMod typeGetTypeModFromColInfo(const SColumnInfo* pCol) {
  return typeGetTypeMod(pCol->type, pCol->precision, pCol->scale, pCol->bytes);
}

static inline STypeMod typeGetTypeModFromCol(const SColumn* pCol) {
  return typeGetTypeMod(pCol->type, pCol->precision, pCol->scale, pCol->bytes);
}

/**
  @brief Calculate the absolute difference between two int64_t values,
         and return the result as uint64_t.
  @note This function utilizes the rule of unsigned integer arithmetic (always
  mod calculation) to avoid the overflow of the difference calculation and
  absolute calculation.
  For example, if a = INT64_MIN(-2^63) and b = 1, the difference b - a is
  1 - (-2^63), which is bigger than INT64_MAX(2^63 - 1) and overflow.
  But if we transfer them to unsigned integer: (uint64_t)a = 2^63,
  (uint64_t)b = 1. The result is 1 - 2^63 which will be calculated as
  1 - 2^63 + 2^64 = 2^64 - 2^63 + 1 = 2^63 + 1, which is not overflow and
  correct.
*/
static inline uint64_t safe_abs_diff_i64(int64_t a, int64_t b) {
  uint64_t ua = (uint64_t)a;
  uint64_t ub = (uint64_t)b;
  return a > b ? ua - ub : ub - ua;
}

#ifdef __cplusplus
}
#endif

#endif /*_TD_COMMON_DEF_H_*/

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	#ifndef _TD_COMMON_DEF_H_
17	#define _TD_COMMON_DEF_H_
18
19	#include "tarray.h"
20	#include "tmsg.h"
21	#include "tvariant.h"
22
23	#ifdef __cplusplus
24	extern "C" {
25	#endif
26
27	// clang-format off
28	#define IS_META_MSG(x) ( \
29	x == TDMT_VND_CREATE_STB \
30	\|\| x == TDMT_VND_ALTER_STB \
31	\|\| x == TDMT_VND_DROP_STB \
32	\|\| x == TDMT_VND_CREATE_TABLE \
33	\|\| x == TDMT_VND_ALTER_TABLE \
34	\|\| x == TDMT_VND_DROP_TABLE \
35	\|\| x == TDMT_VND_DELETE \
36	)
37	// clang-format on
38
39	typedef bool (state_key_cmpr_fn)(void pKey1, void* pKey2);
40
41	typedef struct STableKeyInfo {
42	uint64_t uid;
43	uint64_t groupId;
44	uint64_t baseGId;
45	} STableKeyInfo;
46
47	typedef struct SWinKey {
48	uint64_t groupId;
49	TSKEY ts;
50	int32_t numInGroup;
51	} SWinKey;
52
53	typedef struct SSessionKey {
54	STimeWindow win;
55	uint64_t groupId;
56	} SSessionKey;
57
58	typedef int64_t COUNT_TYPE;
59
60	typedef struct SVersionRange {
61	int64_t minVer;
62	int64_t maxVer;
63	} SVersionRange;
64
65	static inline int winKeyCmprImpl(const void* pKey1, const void* pKey2) {
66	SWinKey* pWin1 = (SWinKey*)pKey1;
67	SWinKey* pWin2 = (SWinKey*)pKey2;
68
69	if (pWin1->groupId > pWin2->groupId) {
70	return 1;
71	} else if (pWin1->groupId < pWin2->groupId) {
72	return -1;
73	}
74
75	if (pWin1->ts > pWin2->ts) {
76	return 1;
77	} else if (pWin1->ts < pWin2->ts) {
78	return -1;
79	}
80
81	return 0;
82	}
83
84	static inline int winKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
85	return winKeyCmprImpl(pKey1, pKey2);
86	}
87
88	typedef struct {
89	uint64_t groupId;
90	TSKEY ts;
91	int32_t exprIdx;
92	} STupleKey;
93
94	static inline int STupleKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
95	STupleKey* pTuple1 = (STupleKey*)pKey1;
96	STupleKey* pTuple2 = (STupleKey*)pKey2;
97
98	if (pTuple1->groupId > pTuple2->groupId) {
99	return 1;
100	} else if (pTuple1->groupId < pTuple2->groupId) {
101	return -1;
102	}
103
104	if (pTuple1->ts > pTuple2->ts) {
105	return 1;
106	} else if (pTuple1->ts < pTuple2->ts) {
107	return -1;
108	}
109
110	if (pTuple1->exprIdx > pTuple2->exprIdx) {
111	return 1;
112	} else if (pTuple1->exprIdx < pTuple2->exprIdx) {
113	return -1;
114	}
115
116	return 0;
117	}
118
119	enum {
120	TMQ_MSG_TYPE__POLL_DATA_RSP = 0,
121	TMQ_MSG_TYPE__POLL_META_RSP,
122	TMQ_MSG_TYPE__EP_RSP,
123	TMQ_MSG_TYPE__POLL_DATA_META_RSP,
124	TMQ_MSG_TYPE__WALINFO_RSP,
125	TMQ_MSG_TYPE__POLL_BATCH_META_RSP,
126	TMQ_MSG_TYPE__POLL_RAW_DATA_RSP,
127	};
128
129	static const char* const tmqMsgTypeStr[] = {
130	"data", "meta", "ask ep", "meta data", "wal info", "batch meta", "raw data"
131	};
132
133	enum {
134	STREAM_INPUT__DATA_SUBMIT = 1,
135	STREAM_INPUT__DATA_BLOCK,
136	STREAM_INPUT__MERGED_SUBMIT,
137	STREAM_INPUT__RECALCULATE,
138	STREAM_INPUT__DATA_RETRIEVE,
139	STREAM_INPUT__GET_RES,
140	STREAM_INPUT__CHECKPOINT,
141	STREAM_INPUT__CHECKPOINT_TRIGGER,
142	STREAM_INPUT__TRANS_STATE,
143	STREAM_INPUT__REF_DATA_BLOCK,
144	STREAM_INPUT__DESTROY,
145	};
146
147	#pragma pack(push, 1)
148	typedef struct SColumnDataAgg {
149	int32_t colId;
150	int16_t numOfNull;
151	union {
152	struct {
153	int64_t sum;
154	int64_t max;
155	int64_t min;
156	};
157	struct {
158	uint64_t decimal128Sum[2];
159	uint64_t decimal128Max[2];
160	uint64_t decimal128Min[2];
161	uint8_t overflow;
162	};
163	};
164	} SColumnDataAgg;
165	#pragma pack(pop)
166
167	#define DECIMAL_AGG_FLAG 0x80000000
168
169	#define COL_AGG_GET_SUM_PTR(pAggs, dataType) \
170	(!IS_DECIMAL_TYPE(dataType) ? (void)&pAggs->sum : (void)pAggs->decimal128Sum)
171
172	#define COL_AGG_GET_MAX_PTR(pAggs, dataType) \
173	(!IS_DECIMAL_TYPE(dataType) ? (void)&pAggs->max : (void)pAggs->decimal128Max)
174
175	#define COL_AGG_GET_MIN_PTR(pAggs, dataType) \
176	(!IS_DECIMAL_TYPE(dataType) ? (void)&pAggs->min : (void)pAggs->decimal128Min)
177
178	typedef struct SBlockID {
179	// The uid of table, from which current data block comes. And it is always 0, if current block is the
180	// result of calculation.
181	uint64_t uid;
182
183	// Block id, acquired and assigned from executor, which created according to the hysical planner. Block id is used
184	// to mark the stage of exec task.
185	int64_t blockId;
186
187	// Generated by group/partition by [value\|tags]. Created and assigned by table-scan operator, group-by operator,
188	// and partition by operator.
189	uint64_t groupId;
190
191	// Base group id for stream group generated by trigger
192	uint64_t baseGId;
193	} SBlockID;
194
195	typedef struct SPkInfo {
196	int8_t type;
197	int32_t bytes;
198	union {
199	int64_t val;
200	uint8_t* pData;
201	} skey;
202	union {
203	int64_t val;
204	uint8_t* pData;
205	} ekey;
206	} SPkInfo;
207
208	typedef struct SDataBlockInfo {
209	STimeWindow window;
210	int32_t rowSize;
211	uint32_t capacity;
212	int64_t rows; // todo hide this attribute
213	SBlockID id;
214	int16_t hasVarCol;
215	int16_t dataLoad; // denote if the data is loaded or not
216	uint8_t scanFlag;
217	bool blankFill;
218	SValue pks[2];
219
220	// TODO: optimize and remove following
221	int64_t version; // used for stream, and need serialization
222	int32_t childId; // used for stream, do not serialize
223	STimeWindow calWin; // used for stream, do not serialize
224	TSKEY watermark; // used for stream
225
226	char parTbName[TSDB_TABLE_NAME_LEN]; // used for stream partition
227	} SDataBlockInfo;
228
229	typedef struct SSDataBlock {
230	SColumnDataAgg* pBlockAgg;
231	SArray* pDataBlock; // SArray<SColumnInfoData>
232	SDataBlockInfo info;
233	} SSDataBlock;
234
235	typedef struct SVarColAttr {
236	int32_t* offset; // start position for each entry in the list
237	uint32_t length; // used buffer size that contain the valid data
238	uint32_t allocLen; // allocated buffer size
239	} SVarColAttr;
240
241	// pBlockAgg->numOfNull == info.rows, all data are null
242	// pBlockAgg->numOfNull == 0, no data are null.
243	typedef struct SColumnInfoData {
244	char* pData; // the corresponding block data in memory
245	union {
246	char* nullbitmap; // bitmap, one bit for each item in the list
247	SVarColAttr varmeta;
248	};
249	SColumnInfo info; // column info
250	bool hasNull; // if current column data has null value.
251	bool reassigned; // if current column data is reassigned.
252	} SColumnInfoData;
253
254	typedef struct SQueryTableDataCond {
255	uint64_t suid;
256	int32_t order; // desc\|asc order to iterate the data block
257	int32_t numOfCols;
258	SColumnInfo* colList;
259	int32_t* pSlotList; // the column output destation slot, and it may be null
260	int32_t type; // data block load type:
261	bool skipRollup;
262	STimeWindow twindows;
263	STimeWindow extTwindows[2];
264	int64_t startVersion;
265	int64_t endVersion;
266	bool notLoadData; // response the actual data, not only the rows in the attribute of info.row of ssdatablock
267	bool cacheSttStatis;
268	} SQueryTableDataCond;
269
270	int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock);
271	void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock);
272
273	void colDataDestroy(SColumnInfoData* pColData);
274
275	//======================================================================================================================
276	// the following structure shared by parser and executor
277	typedef struct SColumn {
278	union {
279	uint64_t uid;
280	int64_t dataBlockId;
281	};
282
283	int16_t colId;
284	int16_t slotId;
285
286	char name[TSDB_COL_NAME_LEN];
287	int16_t colType; // column type: normal column, tag, or window column
288	int16_t type;
289	int32_t bytes;
290	uint8_t precision;
291	uint8_t scale;
292	} SColumn;
293
294	typedef struct STableBlockDistInfo {
295	uint32_t rowSize;
296	uint16_t numOfFiles;
297	uint32_t numOfTables;
298	uint32_t numOfBlocks;
299	uint64_t totalSize;
300	uint64_t totalRows;
301	int32_t maxRows;
302	int32_t minRows;
303	int32_t defMinRows;
304	int32_t defMaxRows;
305	int32_t firstSeekTimeUs;
306	uint32_t numOfInmemRows;
307	uint32_t numOfSttRows;
308	uint32_t numOfVgroups;
309	int32_t blockRowsHisto[20];
310	int32_t blockRowsHistoFixed[8]; // buckets: ≤64, ≤128, ≤256, ≤512, ≤1024, ≤2048, ≤4096, >4096
311	} STableBlockDistInfo;
312
313	int32_t tSerializeBlockDistInfo(void* buf, int32_t bufLen, const STableBlockDistInfo* pInfo);
314	int32_t tDeserializeBlockDistInfo(void* buf, int32_t bufLen, STableBlockDistInfo* pInfo);
315
316	typedef struct SDBBlockUsageInfo {
317	uint64_t dataInDiskSize;
318	uint64_t walInDiskSize;
319	uint64_t rawDataSize;
320	} SDBBlockUsageInfo;
321
322	int32_t tSerializeBlockDbUsage(void* buf, int32_t bufLen, const SDBBlockUsageInfo* pInfo);
323	int32_t tDeserializeBlockDbUsage(void* buf, int32_t bufLen, SDBBlockUsageInfo* pInfo);
324
325	enum {
326	FUNC_PARAM_TYPE_VALUE = 0x1,
327	FUNC_PARAM_TYPE_COLUMN = 0x2,
328	};
329
330	typedef struct SFunctParam {
331	int32_t type;
332	SColumn* pCol;
333	SVariant param;
334	} SFunctParam;
335
336	// the structure for sql function in select clause
337	typedef struct SResSchame {
338	int8_t type;
339	int32_t slotId;
340	int32_t bytes;
341	int32_t precision;
342	int32_t scale;
343	char name[TSDB_COL_NAME_LEN];
344	} SResSchema;
345
346	typedef struct SAggSupporter SAggSupporter;
347	typedef struct SExprSupp SExprSupp;
348	typedef struct SGroupResInfo SGroupResInfo;
349	typedef struct SResultRow SResultRow;
350	typedef struct SResultRowInfo SResultRowInfo;
351	typedef struct SExecTaskInfo SExecTaskInfo;
352	typedef struct SRollupCtx {
353	void* pTsdb; // STsdb*
354	void* pTargets; // SNodeList*
355	void* pBuf;
356	SExprSupp* exprSup;
357	SAggSupporter* aggSup;
358	SResultRow* resultRow;
359	SResultRowInfo* resultRowInfo;
360	SGroupResInfo* pGroupResInfo;
361	SExecTaskInfo* pTaskInfo;
362	SSDataBlock* pInputBlock; // input data block for rollup
363	SSDataBlock* pResBlock; // result data block for rollup
364	SArray* pColValArr; // used the generate the aggregate row
365	int32_t rowSize;
366	int32_t maxBufRows; // max buffer rows for aggregation
367	int64_t winTotalRows; // number of total rows for current window
368	int64_t winStartTs; // start timestamp of current window
369	} SRollupCtx;
370
371	typedef struct {
372	const char* key;
373	size_t keyLen;
374	uint8_t type;
375	union {
376	const char* value;
377	int64_t i;
378	uint64_t u;
379	double d;
380	float f;
381	};
382	size_t length;
383	bool keyEscaped;
384	bool valueEscaped;
385	} SSmlKv;
386
387	#define QUERY_ASC_FORWARD_STEP 1
388	#define QUERY_DESC_FORWARD_STEP -1
389
390	#define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) != TSDB_ORDER_DESC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)
391
392	#define SORT_QSORT_T 0x1
393	#define SORT_SPILLED_MERGE_SORT_T 0x2
394	typedef struct SSortExecInfo {
395	int32_t sortMethod;
396	int32_t sortBuffer;
397	int32_t loops; // loop count
398	int32_t writeBytes; // write io bytes
399	int32_t readBytes; // read io bytes
400	} SSortExecInfo;
401
402	typedef struct SExchangeExplainInfo {
403	int8_t mode; // 0: concurrent, 1: sequential
404	int32_t numSources;
405	double avgFetchTimes;
406	uint64_t maxFetchTimes;
407	double avgFetchRows;
408	uint64_t maxFetchRows;
409	double avgFetchCost; // us
410	int64_t maxFetchCost; // us
411	} SExchangeExplainInfo;
412
413	typedef struct SNonSortExecInfo {
414	int32_t blkNums;
415	} SNonSortExecInfo;
416
417	typedef struct STUidTagInfo {
418	char* name;
419	uint64_t uid;
420	void* pTagVal;
421	} STUidTagInfo;
422
423	// stream special block column
424
425	#define START_TS_COLUMN_INDEX 0
426	#define END_TS_COLUMN_INDEX 1
427	#define UID_COLUMN_INDEX 2
428	#define GROUPID_COLUMN_INDEX 3
429	#define CALCULATE_START_TS_COLUMN_INDEX 4
430	#define CALCULATE_END_TS_COLUMN_INDEX 5
431	#define TABLE_NAME_COLUMN_INDEX 6
432	#define PRIMARY_KEY_COLUMN_INDEX 7
433
434	//steam get result block column
435	#define DATA_TS_COLUMN_INDEX 0
436	#define DATA_VERSION_COLUMN_INDEX 1
437
438	// stream create table block column
439	#define UD_TABLE_NAME_COLUMN_INDEX 0
440	#define UD_GROUPID_COLUMN_INDEX 1
441	#define UD_TAG_COLUMN_INDEX 2
442
443	// stream notify event block column
444	#define NOTIFY_EVENT_STR_COLUMN_INDEX 0
445
446	int32_t taosGenCrashJsonMsg(int signum, char** pMsg, int64_t clusterId, int64_t startTime);
447
448	#define SHOW_VAR_PRIV_SYSTEM 0x01
449	#define SHOW_VAR_PRIV_SECURITY 0x02
450	#define SHOW_VAR_PRIV_AUDIT 0x04
451	#define SHOW_VAR_PRIV_DEBUG 0x08
452	#define SHOW_VAR_PRIV_ALL (SHOW_VAR_PRIV_SYSTEM \| SHOW_VAR_PRIV_SECURITY \| SHOW_VAR_PRIV_AUDIT \| SHOW_VAR_PRIV_DEBUG)
453
454	int32_t dumpConfToDataBlock(SSDataBlock* pBlock, int32_t startCol, char* likePattern, uint8_t showPrivMask);
455
456	#define TSMA_RES_STB_POSTFIX "_tsma_res_stb_"
457	#define MD5_OUTPUT_LEN 32
458	#define TSMA_RES_STB_EXTRA_COLUMN_NUM 4 // 3 columns: _wstart, _wend, _wduration, 1 tag: tbname
459
460	static inline bool isTsmaResSTb(const char* stbName) {	38,456,955✔
461	static bool showTsmaTables = true;
462	if (showTsmaTables) return false;	38,456,955✔
463	const char* pos = strstr(stbName, TSMA_RES_STB_POSTFIX);	×
464	if (pos && strlen(stbName) == (pos - stbName) + strlen(TSMA_RES_STB_POSTFIX)) {	×
465	return true;	×
466	}
467	return false;	×
468	}
469
470	static inline STypeMod typeGetTypeModFromColInfo(const SColumnInfo* pCol) {	2,147,483,647✔
471	return typeGetTypeMod(pCol->type, pCol->precision, pCol->scale, pCol->bytes);	2,147,483,647✔
472	}
473
474	static inline STypeMod typeGetTypeModFromCol(const SColumn* pCol) {	×
475	return typeGetTypeMod(pCol->type, pCol->precision, pCol->scale, pCol->bytes);	×
476	}
477
478	/**
479	@brief Calculate the absolute difference between two int64_t values,
480	and return the result as uint64_t.
481	@note This function utilizes the rule of unsigned integer arithmetic (always
482	mod calculation) to avoid the overflow of the difference calculation and
483	absolute calculation.
484	For example, if a = INT64_MIN(-2^63) and b = 1, the difference b - a is
485	1 - (-2^63), which is bigger than INT64_MAX(2^63 - 1) and overflow.
486	But if we transfer them to unsigned integer: (uint64_t)a = 2^63,
487	(uint64_t)b = 1. The result is 1 - 2^63 which will be calculated as
488	1 - 2^63 + 2^64 = 2^64 - 2^63 + 1 = 2^63 + 1, which is not overflow and
489	correct.
490	*/
491	static inline uint64_t safe_abs_diff_i64(int64_t a, int64_t b) {	545,591,966✔
492	uint64_t ua = (uint64_t)a;	545,591,966✔
493	uint64_t ub = (uint64_t)b;	545,591,966✔
494	return a > b ? ua - ub : ub - ua;	545,591,966✔
495	}
496
497	#ifdef __cplusplus
498	}
499	#endif
500
501	#endif /_TD_COMMON_DEF_H_/

taosdata / TDengine / #5059

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

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