#4917

Committed 07 Jan 2026 03:52PM UTC coverage: 65.42% (+0.02%) from 65.402%

Build # #4917

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch #34204

Run Details

31 of 34 new or added lines in 2 files covered. (91.18%)

819 existing lines in 129 files now uncovered.

202679 of 309814 relevant lines covered (65.42%)

116724351.99 hits per line

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

40.0

/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;
} 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,
};

typedef enum EStreamType {
  STREAM_NORMAL = 1,
  STREAM_INVERT,
  STREAM_CLEAR,
  STREAM_INVALID,
  STREAM_GET_ALL,
  STREAM_DELETE_RESULT,
  STREAM_DELETE_DATA,
  STREAM_RETRIEVE,
  STREAM_PULL_DATA,
  STREAM_PULL_OVER,
  STREAM_FILL_OVER,
  STREAM_CHECKPOINT,
  STREAM_CREATE_CHILD_TABLE,
  STREAM_TRANS_STATE,
  STREAM_MID_RETRIEVE,
  STREAM_PARTITION_DELETE_DATA,
  STREAM_GET_RESULT,
  STREAM_DROP_CHILD_TABLE,
  STREAM_NOTIFY_EVENT,
  STREAM_RECALCULATE_DATA,
  STREAM_RECALCULATE_DELETE,
  STREAM_RECALCULATE_START,
  STREAM_RECALCULATE_END,
} EStreamType;

#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.
  uint64_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;
} 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
  EStreamType type;       // 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];
} 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 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);
int32_t dumpConfToDataBlock(SSDataBlock* pBlock, int32_t startCol, char* likePattern);

#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);
}

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

taosdata / TDengine / #4917

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