Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 df57768a 编写于 作者: H Haojun Liao
浏览文件
操作

Merge branch '3.0' into fix/3_liaohj

上级 cdffabcd 78bdb043
展开全部 隐藏空白更改
内联 并排
Showing with 17331 addition and 2885 deletion
docs/en/13-operation/10-monitor.md
浏览文件 @ df57768a
......@@ -214,19 +214,6 @@ The data of tdinsight dashboard is stored in `log` database (default. You can ch
|dnode\_ep|NCHAR|TAG|dnode endpoint|
|cluster\_id|NCHAR|TAG|cluster id|
### logs table
`logs` table contains login information records.
|field|type|is\_tag|comment|
|:----|:---|:-----|:------|
|ts|TIMESTAMP||timestamp|
|level|VARCHAR||log level|
|content|NCHAR||log content|
|dnode\_id|INT|TAG|dnode id|
|dnode\_ep|NCHAR|TAG|dnode endpoint|
|cluster\_id|NCHAR|TAG|cluster id|
### log\_summary table
`log_summary` table contains log summary information records.
......
docs/en/14-reference/03-connector/07-python.mdx
浏览文件 @ df57768a
......@@ -1007,13 +1007,12 @@ consumer.close()
### Other sample programs
| Example program links | Example program content |
| ------------------------------------------------------------------------------------------------------------- | ------------------- ---- |
| [bind_multi.py](https://github.com/taosdata/taos-connector-python/blob/main/examples/bind-multi.py) | parameter binding,
bind multiple rows at once |
| [bind_row.py](https://github.com/taosdata/taos-connector-python/blob/main/examples/bind-row.py) | bind_row.py
|-----------------------|-------------------------|
| [bind_multi.py](https://github.com/taosdata/taos-connector-python/blob/main/examples/bind-multi.py) | parameter binding, bind multiple rows at once |
| [bind_row.py](https://github.com/taosdata/taos-connector-python/blob/main/examples/bind-row.py) | parameter binding, bind one row at once |
| [insert_lines.py](https://github.com/taosdata/taos-connector-python/blob/main/examples/insert-lines.py) | InfluxDB line protocol writing |
| [json_tag.py](https://github.com/taosdata/taos-connector-python/blob/main/examples/json-tag.py) | Use JSON type tags |
| [tmq.py](https://github.com/taosdata/taos-connector-python/blob/main/examples/tmq.py) | TMQ subscription |
| [tmq_consumer.py](https://github.com/taosdata/taos-connector-python/blob/main/examples/tmq_consumer.py) | TMQ subscription |
## Other notes
......
docs/zh/17-operation/10-monitor.md
浏览文件 @ df57768a
......@@ -210,19 +210,6 @@ TDinsight dashboard 数据来源于 log 库（存放监控数据的默认db，
|dnode\_ep|NCHAR|TAG|dnode endpoint|
|cluster\_id|NCHAR|TAG|cluster id|
### logs 表
`logs` 表记录登录信息。
|field|type|is\_tag|comment|
|:----|:---|:-----|:------|
|ts|TIMESTAMP||timestamp|
|level|VARCHAR||log level|
|content|NCHAR||log content，长度不超过1024字节|
|dnode\_id|INT|TAG|dnode id|
|dnode\_ep|NCHAR|TAG|dnode endpoint|
|cluster\_id|NCHAR|TAG|cluster id|
### log\_summary 表
`log_summary` 记录日志统计信息。
......
include/common/tglobal.h
浏览文件 @ df57768a
......@@ -85,8 +85,14 @@ extern int64_t tsVndCommitMaxIntervalMs;
extern int64_t tsMndSdbWriteDelta;
extern int64_t tsMndLogRetention;
extern int8_t tsGrant;
extern int32_t tsMndGrantMode;
extern bool tsMndSkipGrant;
// dnode
extern int64_t tsDndStart;
extern int64_t tsDndStartOsUptime;
extern int64_t tsDndUpTime;
// monitor
extern bool tsEnableMonitor;
extern int32_t tsMonitorInterval;
......
include/os/os.h
浏览文件 @ df57768a
......@@ -53,6 +53,7 @@ extern "C" {
#else
#include <argp.h>
#include <sys/prctl.h>
#include <sys/sysinfo.h>
#if defined(_TD_X86_)
#include <cpuid.h>
#endif
......
include/os/osSysinfo.h
浏览文件 @ df57768a
......@@ -35,6 +35,7 @@ typedef struct {
bool taosCheckSystemIsLittleEnd();
void taosGetSystemInfo();
int64_t taosGetOsUptime();
int32_t taosGetEmail(char *email, int32_t maxLen);
int32_t taosGetOsReleaseName(char *releaseName, char* sName, char* ver, int32_t maxLen);
int32_t taosGetCpuInfo(char *cpuModel, int32_t maxLen, float *numOfCores);
......
include/util/tarray.h
浏览文件 @ df57768a
......@@ -22,7 +22,7 @@
extern "C" {
#endif
#define TARRAY_MIN_SIZE 8
#define TARRAY_MIN_SIZE 4
#define TARRAY_GET_ELEM(array, index) ((void*)((char*)((array)->pData) + (index) * (array)->elemSize))
#define TARRAY_ELEM_IDX(array, ele) (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
......@@ -138,7 +138,7 @@ size_t taosArrayGetSize(const SArray* pArray);
* @param index
* @param pData
*/
void* taosArrayInsert(SArray* pArray, size_t index, void* pData);
void* taosArrayInsert(SArray* pArray, size_t index, const void* pData);
/**
* set data in array
......@@ -204,9 +204,9 @@ void taosArrayClearEx(SArray* pArray, void (*fp)(void*));
void* taosArrayDestroy(SArray* pArray);
void taosArrayDestroyP(SArray* pArray, FDelete fp);
void taosArrayDestroyP(SArray* pArray, FDelete fp);
void taosArrayDestroyEx(SArray* pArray, FDelete fp);
void taosArrayDestroyEx(SArray* pArray, FDelete fp);
void taosArraySwap(SArray* a, SArray* b);
......
include/util/tarray2.h 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "talgo.h"
#ifndef _TD_UTIL_TARRAY2_H_
#define _TD_UTIL_TARRAY2_H_
#ifdef __cplusplus
extern "C" {
#endif
// a: a
// e: element
// ep: element pointer
// cmp: compare function
// idx: index
// cb: callback function
#define TARRAY2(TYPE) \
struct { \
int32_t size; \
int32_t capacity; \
TYPE *data; \
}
typedef void (*TArray2Cb)(void *);
#define TARRAY2_SIZE(a) ((a)->size)
#define TARRAY2_CAPACITY(a) ((a)->capacity)
#define TARRAY2_DATA(a) ((a)->data)
#define TARRAY2_GET(a, i) ((a)->data[i])
#define TARRAY2_GET_PTR(a, i) ((a)->data + i)
#define TARRAY2_FIRST(a) ((a)->data[0])
#define TARRAY2_LAST(a) ((a)->data[(a)->size - 1])
#define TARRAY2_DATA_LEN(a) ((a)->size * sizeof(((a)->data[0])))
static FORCE_INLINE int32_t tarray2_make_room(void *arr, int32_t expSize, int32_t eleSize) {
TARRAY2(void) *a = arr;
int32_t capacity = (a->capacity > 0) ? (a->capacity << 1) : 32;
while (capacity < expSize) {
capacity <<= 1;
}
void *p = taosMemoryRealloc(a->data, capacity * eleSize);
if (p == NULL) return TSDB_CODE_OUT_OF_MEMORY;
a->capacity = capacity;
a->data = p;
return 0;
}
static FORCE_INLINE int32_t tarray2InsertBatch(void *arr, int32_t idx, const void *elePtr, int32_t numEle,
int32_t eleSize) {
TARRAY2(uint8_t) *a = arr;
int32_t ret = 0;
if (a->size + numEle > a->capacity) {
ret = tarray2_make_room(a, a->size + numEle, eleSize);
}
if (ret == 0) {
if (idx < a->size) {
memmove(a->data + (idx + numEle) * eleSize, a->data + idx * eleSize, (a->size - idx) * eleSize);
}
memcpy(a->data + idx * eleSize, elePtr, numEle * eleSize);
a->size += numEle;
}
return ret;
}
static FORCE_INLINE void *tarray2Search(void *arr, const void *elePtr, int32_t eleSize, __compar_fn_t compar,
int32_t flag) {
TARRAY2(void) *a = arr;
return taosbsearch(elePtr, a->data, a->size, eleSize, compar, flag);
}
static FORCE_INLINE int32_t tarray2SearchIdx(void *arr, const void *elePtr, int32_t eleSize, __compar_fn_t compar,
int32_t flag) {
TARRAY2(void) *a = arr;
void *p = taosbsearch(elePtr, a->data, a->size, eleSize, compar, flag);
if (p == NULL) {
return -1;
} else {
return (int32_t)(((uint8_t *)p - (uint8_t *)a->data) / eleSize);
}
}
static FORCE_INLINE int32_t tarray2SortInsert(void *arr, const void *elePtr, int32_t eleSize, __compar_fn_t compar) {
TARRAY2(void) *a = arr;
int32_t idx = tarray2SearchIdx(arr, elePtr, eleSize, compar, TD_GT);
return tarray2InsertBatch(arr, idx < 0 ? a->size : idx, elePtr, 1, eleSize);
}
#define TARRAY2_INIT_EX(a, size_, capacity_, data_) \
do { \
(a)->size = (size_); \
(a)->capacity = (capacity_); \
(a)->data = (data_); \
} while (0)
#define TARRAY2_INIT(a) TARRAY2_INIT_EX(a, 0, 0, NULL)
#define TARRAY2_CLEAR(a, cb) \
do { \
if ((cb) && (a)->size > 0) { \
TArray2Cb cb_ = (TArray2Cb)(cb); \
for (int32_t i = 0; i < (a)->size; ++i) { \
cb_((a)->data + i); \
} \
} \
(a)->size = 0; \
} while (0)
#define TARRAY2_DESTROY(a, cb) \
do { \
TARRAY2_CLEAR(a, cb); \
if ((a)->data) { \
taosMemoryFree((a)->data); \
(a)->data = NULL; \
} \
(a)->capacity = 0; \
} while (0)
#define TARRAY2_INSERT_PTR(a, idx, ep) tarray2InsertBatch(a, idx, ep, 1, sizeof((a)->data[0]))
#define TARRAY2_APPEND_PTR(a, ep) tarray2InsertBatch(a, (a)->size, ep, 1, sizeof((a)->data[0]))
#define TARRAY2_APPEND_BATCH(a, ep, n) tarray2InsertBatch(a, (a)->size, ep, n, sizeof((a)->data[0]))
#define TARRAY2_APPEND(a, e) TARRAY2_APPEND_PTR(a, &(e))
// return (TYPE *)
#define TARRAY2_SEARCH(a, ep, cmp, flag) tarray2Search(a, ep, sizeof(((a)->data[0])), (__compar_fn_t)cmp, flag)
#define TARRAY2_SEARCH_IDX(a, ep, cmp, flag) tarray2SearchIdx(a, ep, sizeof(((a)->data[0])), (__compar_fn_t)cmp, flag)
#define TARRAY2_SORT_INSERT(a, e, cmp) tarray2SortInsert(a, &(e), sizeof(((a)->data[0])), (__compar_fn_t)cmp)
#define TARRAY2_SORT_INSERT_P(a, ep, cmp) tarray2SortInsert(a, ep, sizeof(((a)->data[0])), (__compar_fn_t)cmp)
#define TARRAY2_REMOVE(a, idx, cb) \
do { \
if ((idx) < (a)->size) { \
if (cb) { \
TArray2Cb cb_ = (TArray2Cb)(cb); \
cb_((a)->data + (idx)); \
} \
if ((idx) < (a)->size - 1) { \
memmove((a)->data + (idx), (a)->data + (idx) + 1, sizeof((*(a)->data)) * ((a)->size - (idx)-1)); \
} \
(a)->size--; \
} \
} while (0)
#define TARRAY2_FOREACH(a, e) for (int32_t __i = 0; __i < (a)->size && ((e) = (a)->data[__i], 1); __i++)
#define TARRAY2_FOREACH_REVERSE(a, e) for (int32_t __i = (a)->size - 1; __i >= 0 && ((e) = (a)->data[__i], 1); __i--)
#define TARRAY2_FOREACH_PTR(a, ep) for (int32_t __i = 0; __i < (a)->size && ((ep) = &(a)->data[__i], 1); __i++)
#define TARRAY2_FOREACH_PTR_REVERSE(a, ep) \
for (int32_t __i = (a)->size - 1; __i >= 0 && ((ep) = &(a)->data[__i], 1); __i--)
#ifdef __cplusplus
}
#endif
#endif /*_TD_UTIL_TARRAY2_H_*/
include/util/tdef.h
浏览文件 @ df57768a
......@@ -191,16 +191,16 @@ typedef enum ELogicConditionType {
#define TSDB_MAX_COLUMNS 4096
#define TSDB_MIN_COLUMNS 2 // PRIMARY COLUMN(timestamp) + other columns
#define TSDB_NODE_NAME_LEN 64
#define TSDB_TABLE_NAME_LEN 193 // it is a null-terminated string
#define TSDB_TOPIC_NAME_LEN 193 // it is a null-terminated string
#define TSDB_CGROUP_LEN 193 // it is a null-terminated string
#define TSDB_OFFSET_LEN 64 // it is a null-terminated string
#define TSDB_USER_CGROUP_LEN (TSDB_USER_LEN + TSDB_CGROUP_LEN) // it is a null-terminated string
#define TSDB_STREAM_NAME_LEN 193 // it is a null-terminated string
#define TSDB_DB_NAME_LEN 65
#define TSDB_DB_FNAME_LEN (TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN + TSDB_NAME_DELIMITER_LEN)
#define TSDB_PRIVILEDGE_CONDITION_LEN 200
#define TSDB_NODE_NAME_LEN 64
#define TSDB_TABLE_NAME_LEN 193 // it is a null-terminated string
#define TSDB_TOPIC_NAME_LEN 193 // it is a null-terminated string
#define TSDB_CGROUP_LEN 193 // it is a null-terminated string
#define TSDB_OFFSET_LEN 64 // it is a null-terminated string
#define TSDB_USER_CGROUP_LEN (TSDB_USER_LEN + TSDB_CGROUP_LEN) // it is a null-terminated string
#define TSDB_STREAM_NAME_LEN 193 // it is a null-terminated string
#define TSDB_DB_NAME_LEN 65
#define TSDB_DB_FNAME_LEN (TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN + TSDB_NAME_DELIMITER_LEN)
#define TSDB_PRIVILEDGE_CONDITION_LEN 200
#define TSDB_FUNC_NAME_LEN 65
#define TSDB_FUNC_COMMENT_LEN 1024 * 1024
......@@ -249,15 +249,15 @@ typedef enum ELogicConditionType {
#define TSDB_LABEL_LEN 8
#define TSDB_JOB_STATUS_LEN 32
#define TSDB_CLUSTER_ID_LEN 40
#define TSDB_FQDN_LEN 128
#define TSDB_EP_LEN (TSDB_FQDN_LEN + 6)
#define TSDB_IPv4ADDR_LEN 16
#define TSDB_FILENAME_LEN 128
#define TSDB_SHOW_SQL_LEN 2048
#define TSDB_CLUSTER_ID_LEN 40
#define TSDB_FQDN_LEN 128
#define TSDB_EP_LEN (TSDB_FQDN_LEN + 6)
#define TSDB_IPv4ADDR_LEN 16
#define TSDB_FILENAME_LEN 128
#define TSDB_SHOW_SQL_LEN 2048
#define TSDB_SHOW_SCHEMA_JSON_LEN TSDB_MAX_COLUMNS * 256
#define TSDB_SLOW_QUERY_SQL_LEN 512
#define TSDB_SHOW_SUBQUERY_LEN 1000
#define TSDB_SLOW_QUERY_SQL_LEN 512
#define TSDB_SHOW_SUBQUERY_LEN 1000
#define TSDB_TRANS_STAGE_LEN 12
#define TSDB_TRANS_TYPE_LEN 16
......@@ -370,7 +370,7 @@ typedef enum ELogicConditionType {
#define TSDB_DEFAULT_DB_SCHEMALESS TSDB_DB_SCHEMALESS_OFF
#define TSDB_MIN_STT_TRIGGER 1
#define TSDB_MAX_STT_TRIGGER 16
#define TSDB_DEFAULT_SST_TRIGGER 1
#define TSDB_DEFAULT_SST_TRIGGER 2
#define TSDB_MIN_HASH_PREFIX (2 - TSDB_TABLE_NAME_LEN)
#define TSDB_MAX_HASH_PREFIX (TSDB_TABLE_NAME_LEN - 2)
#define TSDB_DEFAULT_HASH_PREFIX 0
......@@ -410,10 +410,10 @@ typedef enum ELogicConditionType {
#define TSDB_EXPLAIN_RESULT_ROW_SIZE (16 * 1024)
#define TSDB_EXPLAIN_RESULT_COLUMN_NAME "QUERY_PLAN"
#define TSDB_MAX_FIELD_LEN 65519 // 16384:65519
#define TSDB_MAX_BINARY_LEN TSDB_MAX_FIELD_LEN // 16384-8:65519
#define TSDB_MAX_NCHAR_LEN TSDB_MAX_FIELD_LEN // 16384-8:65519
#define TSDB_MAX_GEOMETRY_LEN TSDB_MAX_FIELD_LEN // 16384-8:65519
#define TSDB_MAX_FIELD_LEN 65519 // 16384:65519
#define TSDB_MAX_BINARY_LEN TSDB_MAX_FIELD_LEN // 16384-8:65519
#define TSDB_MAX_NCHAR_LEN TSDB_MAX_FIELD_LEN // 16384-8:65519
#define TSDB_MAX_GEOMETRY_LEN TSDB_MAX_FIELD_LEN // 16384-8:65519
#define PRIMARYKEY_TIMESTAMP_COL_ID 1
#define COL_REACH_END(colId, maxColId) ((colId) > (maxColId))
......
include/util/tlist.h
浏览文件 @ df57768a
......@@ -241,6 +241,54 @@ void tdListNodeGetData(SList *list, SListNode *node, void *target);
void tdListInitIter(SList *list, SListIter *pIter, TD_LIST_DIRECTION_T direction);
SListNode *tdListNext(SListIter *pIter);
// macros ====================================================================================
// q: for queue
// n: for node
// m: for member
#define LISTD(TYPE) \
struct { \
TYPE *next, *prev; \
}
#define LISTD_NEXT(n, m) ((n)->m.next)
#define LISTD_PREV(n, m) ((n)->m.prev)
#define LISTD_INIT(q, m) (LISTD_NEXT(q, m) = LISTD_PREV(q, m) = (q))
#define LISTD_HEAD(q, m) (LISTD_NEXT(q, m))
#define LISTD_TAIL(q, m) (LISTD_PREV(q, m))
#define LISTD_PREV_NEXT(n, m) (LISTD_NEXT(LISTD_PREV(n, m), m))
#define LISTD_NEXT_PREV(n, m) (LISTD_PREV(LISTD_NEXT(n, m), m))
#define LISTD_INSERT_HEAD(q, n, m) \
do { \
LISTD_NEXT(n, m) = LISTD_NEXT(q, m); \
LISTD_PREV(n, m) = (q); \
LISTD_NEXT_PREV(n, m) = (n); \
LISTD_NEXT(q, m) = (n); \
} while (0)
#define LISTD_INSERT_TAIL(q, n, m) \
do { \
LISTD_NEXT(n, m) = (q); \
LISTD_PREV(n, m) = LISTD_PREV(q, m); \
LISTD_PREV_NEXT(n, m) = (n); \
LISTD_PREV(q, m) = (n); \
} while (0)
#define LISTD_REMOVE(n, m) \
do { \
LISTD_PREV_NEXT(n, m) = LISTD_NEXT(n, m); \
LISTD_NEXT_PREV(n, m) = LISTD_PREV(n, m); \
} while (0)
#define LISTD_FOREACH(q, n, m) for ((n) = LISTD_HEAD(q, m); (n) != (q); (n) = LISTD_NEXT(n, m))
#define LISTD_FOREACH_REVERSE(q, n, m) for ((n) = LISTD_TAIL(q, m); (n) != (q); (n) = LISTD_PREV(n, m))
#define LISTD_FOREACH_SAFE(q, n, t, m) \
for ((n) = LISTD_HEAD(q, m), (t) = LISTD_NEXT(n, m); (n) != (q); (n) = (t), (t) = LISTD_NEXT(n, m))
#define LISTD_FOREACH_REVERSE_SAFE(q, n, t, m) \
for ((n) = LISTD_TAIL(q, m), (t) = LISTD_PREV(n, m); (n) != (q); (n) = (t), (t) = LISTD_PREV(n, m))
#ifdef __cplusplus
}
#endif
......
include/util/trbtree.h
浏览文件 @ df57768a
......@@ -39,7 +39,7 @@ void tRBTreeDrop(SRBTree *pTree, SRBTreeNode *z);
SRBTreeNode *tRBTreeDropByKey(SRBTree *pTree, void *pKey);
SRBTreeNode *tRBTreeDropMin(SRBTree *pTree);
SRBTreeNode *tRBTreeDropMax(SRBTree *pTree);
SRBTreeNode *tRBTreeGet(SRBTree *pTree, const SRBTreeNode *pKeyNode);
SRBTreeNode *tRBTreeGet(const SRBTree *pTree, const SRBTreeNode *pKeyNode);
// SRBTreeIter =============================================
#define tRBTreeIterCreate(tree, ascend) \
......@@ -67,9 +67,9 @@ struct SRBTree {
};
struct SRBTreeIter {
int8_t asc;
SRBTree *pTree;
SRBTreeNode *pNode;
int8_t asc;
const SRBTree *pTree;
SRBTreeNode *pNode;
};
#ifdef __cplusplus
......
include/util/tutil.h
浏览文件 @ df57768a
......@@ -29,7 +29,7 @@ extern "C" {
int32_t strdequote(char *src);
size_t strtrim(char *src);
char *strnchr(const char *haystack, char needle, int32_t len, bool skipquote);
TdUcs4* wcsnchr(const TdUcs4* haystack, TdUcs4 needle, size_t len);
TdUcs4 *wcsnchr(const TdUcs4 *haystack, TdUcs4 needle, size_t len);
char **strsplit(char *src, const char *delim, int32_t *num);
char *strtolower(char *dst, const char *src);
......@@ -37,11 +37,11 @@ char *strntolower(char *dst, const char *src, int32_t n);
char *strntolower_s(char *dst, const char *src, int32_t n);
int64_t strnatoi(char *num, int32_t len);
size_t tstrncspn(const char *str, size_t ssize, const char *reject, size_t rsize);
size_t twcsncspn(const TdUcs4 *wcs, size_t size, const TdUcs4 *reject, size_t rsize);
size_t tstrncspn(const char *str, size_t ssize, const char *reject, size_t rsize);
size_t twcsncspn(const TdUcs4 *wcs, size_t size, const TdUcs4 *reject, size_t rsize);
char *strbetween(char *string, char *begin, char *end);
char *paGetToken(char *src, char **token, int32_t *tokenLen);
char *strbetween(char *string, char *begin, char *end);
char *paGetToken(char *src, char **token, int32_t *tokenLen);
int32_t taosByteArrayToHexStr(char bytes[], int32_t len, char hexstr[]);
int32_t taosHexStrToByteArray(char hexstr[], char bytes[]);
......@@ -81,12 +81,13 @@ static FORCE_INLINE void taosEncryptPass_c(uint8_t *inBuf, size_t len, char *tar
static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen, int32_t method, int32_t prefix,
int32_t suffix) {
if ((prefix == 0 && suffix == 0) || (tblen <= (prefix + suffix)) || (tblen <= -1 * (prefix + suffix)) || prefix * suffix < 0) {
if ((prefix == 0 && suffix == 0) || (tblen <= (prefix + suffix)) || (tblen <= -1 * (prefix + suffix)) ||
prefix * suffix < 0) {
return MurmurHash3_32(tbname, tblen);
} else if (prefix > 0 || suffix > 0) {
return MurmurHash3_32(tbname + prefix, tblen - prefix - suffix);
} else {
char tbName[TSDB_TABLE_FNAME_LEN];
char tbName[TSDB_TABLE_FNAME_LEN];
int32_t offset = 0;
if (prefix < 0) {
offset = -1 * prefix;
......@@ -94,20 +95,33 @@ static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen,
}
if (suffix < 0) {
strncpy(tbName + offset, tbname + tblen + suffix, -1 * suffix);
offset += -1 *suffix;
offset += -1 * suffix;
}
return MurmurHash3_32(tbName, offset);
}
}
#define TSDB_CHECK_CODE(CODE, LINO, LABEL) \
if (CODE) { \
LINO = __LINE__; \
goto LABEL; \
do { \
if ((CODE)) { \
LINO = __LINE__; \
goto LABEL; \
} \
} while (0)
#define TSDB_CHECK_NULL(ptr, CODE, LINO, LABEL, ERRNO) \
if ((ptr) == NULL) { \
(CODE) = (ERRNO); \
(LINO) = __LINE__; \
goto LABEL; \
}
#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
#define VND_CHECK_CODE(CODE, LINO, LABEL) TSDB_CHECK_CODE(CODE, LINO, LABEL)
#define TCONTAINER_OF(ptr, type, member) ((type *)((char *)(ptr)-offsetof(type, member)))
#ifdef __cplusplus
}
#endif
......
source/common/src/tdataformat.c
浏览文件 @ df57768a
......@@ -2245,15 +2245,18 @@ static int32_t tColDataUpdateValue72(SColData *pColData, uint8_t *pData, uint32_
}
return 0;
}
static FORCE_INLINE int32_t tColDataUpdateNothing(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) {
return 0;
}
static int32_t (*tColDataUpdateValueImpl[8][3])(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) = {
{NULL, NULL, NULL}, // 0
{tColDataUpdateValue10, NULL, tColDataUpdateValue12}, // HAS_NONE
{tColDataUpdateValue20, NULL, NULL}, // HAS_NULL
{tColDataUpdateValue30, NULL, tColDataUpdateValue32}, // HAS_NULL|HAS_NONE
{tColDataUpdateValue40, NULL, tColDataUpdateValue42}, // HAS_VALUE
{tColDataUpdateValue50, NULL, tColDataUpdateValue52}, // HAS_VALUE|HAS_NONE
{tColDataUpdateValue60, NULL, tColDataUpdateValue62}, // HAS_VALUE|HAS_NULL
{tColDataUpdateValue70, NULL, tColDataUpdateValue72}, // HAS_VALUE|HAS_NULL|HAS_NONE
{NULL, NULL, NULL}, // 0
{tColDataUpdateValue10, tColDataUpdateNothing, tColDataUpdateValue12}, // HAS_NONE
{tColDataUpdateValue20, tColDataUpdateNothing, tColDataUpdateNothing}, // HAS_NULL
{tColDataUpdateValue30, tColDataUpdateNothing, tColDataUpdateValue32}, // HAS_NULL|HAS_NONE
{tColDataUpdateValue40, tColDataUpdateNothing, tColDataUpdateValue42}, // HAS_VALUE
{tColDataUpdateValue50, tColDataUpdateNothing, tColDataUpdateValue52}, // HAS_VALUE|HAS_NONE
{tColDataUpdateValue60, tColDataUpdateNothing, tColDataUpdateValue62}, // HAS_VALUE|HAS_NULL
{tColDataUpdateValue70, tColDataUpdateNothing, tColDataUpdateValue72}, // HAS_VALUE|HAS_NULL|HAS_NONE
// VALUE NONE NULL
};
......
source/common/src/tglobal.c
浏览文件 @ df57768a
......@@ -77,8 +77,14 @@ int64_t tsVndCommitMaxIntervalMs = 600 * 1000;
int64_t tsMndSdbWriteDelta = 200;
int64_t tsMndLogRetention = 2000;
int8_t tsGrant = 1;
int32_t tsMndGrantMode = 0;
bool tsMndSkipGrant = false;
// dnode
int64_t tsDndStart = 0;
int64_t tsDndStartOsUptime = 0;
int64_t tsDndUpTime = 0;
// monitor
bool tsEnableMonitor = true;
int32_t tsMonitorInterval = 30;
......@@ -506,6 +512,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddInt64(pCfg, "mndSdbWriteDelta", tsMndSdbWriteDelta, 20, 10000, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddInt64(pCfg, "mndLogRetention", tsMndLogRetention, 500, 10000, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddInt32(pCfg, "grantMode", tsMndGrantMode, 0, 10000, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddBool(pCfg, "skipGrant", tsMndSkipGrant, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddBool(pCfg, "monitor", tsEnableMonitor, CFG_SCOPE_SERVER) != 0) return -1;
......@@ -915,6 +922,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsMndSdbWriteDelta = cfgGetItem(pCfg, "mndSdbWriteDelta")->i64;
tsMndLogRetention = cfgGetItem(pCfg, "mndLogRetention")->i64;
tsMndSkipGrant = cfgGetItem(pCfg, "skipGrant")->bval;
tsMndGrantMode = cfgGetItem(pCfg, "grantMode")->i32;
tsStartUdfd = cfgGetItem(pCfg, "udf")->bval;
tstrncpy(tsUdfdResFuncs, cfgGetItem(pCfg, "udfdResFuncs")->str, sizeof(tsUdfdResFuncs));
......
source/dnode/mgmt/exe/dmMain.c
浏览文件 @ df57768a
......@@ -373,6 +373,8 @@ int mainWindows(int argc, char **argv) {
dInfo("start to init service");
dmSetSignalHandle();
tsDndStart = taosGetTimestampMs();
tsDndStartOsUptime = taosGetOsUptime();
int32_t code = dmRun();
dInfo("shutting down the service");
......
source/dnode/mgmt/mgmt_dnode/src/dmWorker.c
浏览文件 @ df57768a
......@@ -24,12 +24,16 @@ static void *dmStatusThreadFp(void *param) {
const static int16_t TRIM_FREQ = 30;
int32_t trimCount = 0;
int32_t upTimeCount = 0;
int64_t upTime = 0;
while (1) {
taosMsleep(200);
if (pMgmt->pData->dropped || pMgmt->pData->stopped) break;
int64_t curTime = taosGetTimestampMs();
float interval = (curTime - lastTime) / 1000.0f;
if (curTime < lastTime) lastTime = curTime;
float interval = (curTime - lastTime) / 1000.0f;
if (interval >= tsStatusInterval) {
dmSendStatusReq(pMgmt);
lastTime = curTime;
......@@ -38,6 +42,11 @@ static void *dmStatusThreadFp(void *param) {
if (trimCount == 0) {
taosMemoryTrim(0);
}
if ((upTimeCount = ((upTimeCount + 1) & 63)) == 0) {
upTime = taosGetOsUptime() - tsDndStartOsUptime;
tsDndUpTime = TMAX(tsDndUpTime, upTime);
}
}
}
......@@ -54,7 +63,8 @@ static void *dmMonitorThreadFp(void *param) {
if (pMgmt->pData->dropped || pMgmt->pData->stopped) break;
int64_t curTime = taosGetTimestampMs();
float interval = (curTime - lastTime) / 1000.0f;
if (curTime < lastTime) lastTime = curTime;
float interval = (curTime - lastTime) / 1000.0f;
if (interval >= tsMonitorInterval) {
(*pMgmt->sendMonitorReportFp)();
lastTime = curTime;
......
source/dnode/mgmt/node_mgmt/src/dmTransport.c
浏览文件 @ df57768a
......@@ -290,6 +290,7 @@ int32_t dmInitClient(SDnode *pDnode) {
rpcInit.cfp = (RpcCfp)dmProcessRpcMsg;
rpcInit.sessions = 1024;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.user = TSDB_DEFAULT_USER;
rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.parent = pDnode;
rpcInit.rfp = rpcRfp;
......
source/dnode/mnode/impl/inc/mndCluster.h
浏览文件 @ df57768a
......@@ -27,7 +27,7 @@ void mndCleanupCluster(SMnode *pMnode);
int32_t mndGetClusterName(SMnode *pMnode, char *clusterName, int32_t len);
int64_t mndGetClusterId(SMnode *pMnode);
int64_t mndGetClusterCreateTime(SMnode *pMnode);
float mndGetClusterUpTime(SMnode *pMnode);
int64_t mndGetClusterUpTime(SMnode *pMnode);
#ifdef __cplusplus
}
......
source/dnode/mnode/impl/src/mndCluster.c
浏览文件 @ df57768a
......@@ -123,7 +123,7 @@ static int32_t mndGetClusterUpTimeImp(SClusterObj *pCluster) {
#endif
}
float mndGetClusterUpTime(SMnode *pMnode) {
int64_t mndGetClusterUpTime(SMnode *pMnode) {
int64_t upTime = 0;
void *pIter = NULL;
SClusterObj *pCluster = mndAcquireCluster(pMnode, &pIter);
......@@ -132,7 +132,7 @@ float mndGetClusterUpTime(SMnode *pMnode) {
mndReleaseCluster(pMnode, pCluster, pIter);
}
return upTime / 86400.0f;
return upTime;
}
static SSdbRaw *mndClusterActionEncode(SClusterObj *pCluster) {
......
source/dnode/mnode/impl/src/mndDnode.c
浏览文件 @ df57768a
......@@ -655,6 +655,7 @@ static int32_t mndConfigDnode(SMnode *pMnode, SRpcMsg *pReq, SMCfgDnodeReq *pCfg
STrans *pTrans = NULL;
SDnodeObj *pDnode = NULL;
bool cfgAll = pCfgReq->dnodeId == -1;
int32_t iter = 0;
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
......@@ -662,7 +663,8 @@ static int32_t mndConfigDnode(SMnode *pMnode, SRpcMsg *pReq, SMCfgDnodeReq *pCfg
if (cfgAll) {
pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pDnode);
if (pIter == NULL) break;
} else if(!(pDnode = mndAcquireDnode(pMnode, pCfgReq->dnodeId))) {
++iter;
} else if (!(pDnode = mndAcquireDnode(pMnode, pCfgReq->dnodeId))) {
goto _OVER;
}
......@@ -699,7 +701,7 @@ static int32_t mndConfigDnode(SMnode *pMnode, SRpcMsg *pReq, SMCfgDnodeReq *pCfg
}
if (pTrans && mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
tsGrantHBInterval = TMIN(TMAX(5, iter / 2), 30);
terrno = 0;
_OVER:
......@@ -863,7 +865,7 @@ static int32_t mndProcessCreateDnodeReq(SRpcMsg *pReq) {
code = mndCreateDnode(pMnode, pReq, &createReq);
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
tsGrantHBInterval = 5;
_OVER:
if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
mError("dnode:%s:%d, failed to create since %s", createReq.fqdn, createReq.port, terrstr());
......
source/dnode/mnode/impl/src/mndMain.c
浏览文件 @ df57768a
......@@ -804,7 +804,7 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr
if (pObj->id == pMnode->selfDnodeId) {
pClusterInfo->first_ep_dnode_id = pObj->id;
tstrncpy(pClusterInfo->first_ep, pObj->pDnode->ep, sizeof(pClusterInfo->first_ep));
pClusterInfo->master_uptime = mndGetClusterUpTime(pMnode);
pClusterInfo->master_uptime = (float)mndGetClusterUpTime(pMnode) / 86400.0f;
// pClusterInfo->master_uptime = (ms - pObj->stateStartTime) / (86400000.0f);
tstrncpy(desc.role, syncStr(TAOS_SYNC_STATE_LEADER), sizeof(desc.role));
} else {
......
source/dnode/vnode/CMakeLists.txt
浏览文件 @ df57768a
# vnode
add_library(vnode STATIC "")
target_sources(
vnode
PRIVATE
# vnode
"src/vnd/vnodeOpen.c"
"src/vnd/vnodeBufPool.c"
"src/vnd/vnodeCfg.c"
"src/vnd/vnodeCommit.c"
"src/vnd/vnodeQuery.c"
"src/vnd/vnodeModule.c"
"src/vnd/vnodeSvr.c"
"src/vnd/vnodeSync.c"
"src/vnd/vnodeSnapshot.c"
"src/vnd/vnodeRetention.c"
"src/vnd/vnodeInitApi.c"
set(
VNODE_SOURCE_FILES
"src/vnd/vnodeOpen.c"
"src/vnd/vnodeBufPool.c"
"src/vnd/vnodeCfg.c"
"src/vnd/vnodeCommit.c"
"src/vnd/vnodeQuery.c"
"src/vnd/vnodeModule.c"
"src/vnd/vnodeSvr.c"
"src/vnd/vnodeSync.c"
"src/vnd/vnodeSnapshot.c"
"src/vnd/vnodeRetention.c"
"src/vnd/vnodeInitApi.c"
# meta
"src/meta/metaOpen.c"
......@@ -38,23 +35,23 @@ target_sources(
"src/sma/smaSnapshot.c"
"src/sma/smaTimeRange.c"
# tsdb
"src/tsdb/tsdbCommit.c"
"src/tsdb/tsdbFile.c"
"src/tsdb/tsdbFS.c"
"src/tsdb/tsdbOpen.c"
"src/tsdb/tsdbMemTable.c"
"src/tsdb/tsdbRead.c"
"src/tsdb/tsdbCache.c"
"src/tsdb/tsdbWrite.c"
"src/tsdb/tsdbReaderWriter.c"
"src/tsdb/tsdbUtil.c"
"src/tsdb/tsdbSnapshot.c"
"src/tsdb/tsdbCacheRead.c"
"src/tsdb/tsdbRetention.c"
"src/tsdb/tsdbDiskData.c"
"src/tsdb/tsdbMergeTree.c"
"src/tsdb/tsdbDataIter.c"
# # tsdb
# "src/tsdb/tsdbCommit.c"
# "src/tsdb/tsdbFile.c"
# "src/tsdb/tsdbFS.c"
# "src/tsdb/tsdbOpen.c"
# "src/tsdb/tsdbMemTable.c"
# "src/tsdb/tsdbRead.c"
# "src/tsdb/tsdbCache.c"
# "src/tsdb/tsdbWrite.c"
# "src/tsdb/tsdbReaderWriter.c"
# "src/tsdb/tsdbUtil.c"
# "src/tsdb/tsdbSnapshot.c"
# "src/tsdb/tsdbCacheRead.c"
# "src/tsdb/tsdbRetention.c"
# "src/tsdb/tsdbDiskData.c"
# "src/tsdb/tsdbMergeTree.c"
# "src/tsdb/tsdbDataIter.c"
# tq
"src/tq/tq.c"
......@@ -71,6 +68,19 @@ target_sources(
"src/tq/tqOffsetSnapshot.c"
)
aux_source_directory("src/tsdb/" TSDB_SOURCE_FILES)
list(
APPEND
VNODE_SOURCE_FILES
${TSDB_SOURCE_FILES}
)
target_sources(
vnode
PRIVATE
${VNODE_SOURCE_FILES}
)
IF (TD_VNODE_PLUGINS)
target_sources(
vnode
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ df57768a
......@@ -168,6 +168,27 @@ uint64_t tsdbGetReaderMaxVersion(STsdbReader *pReader);
void tsdbReaderSetCloseFlag(STsdbReader *pReader);
int64_t tsdbGetLastTimestamp(SVnode *pVnode, void *pTableList, int32_t numOfTables, const char *pIdStr);
//======================================================================================================================
int32_t tsdbReaderOpen2(void *pVnode, SQueryTableDataCond *pCond, void *pTableList, int32_t numOfTables,
SSDataBlock *pResBlock, void **ppReader, const char *idstr, bool countOnly,
SHashObj **pIgnoreTables);
int32_t tsdbSetTableList2(STsdbReader *pReader, const void *pTableList, int32_t num);
void tsdbReaderSetId2(STsdbReader *pReader, const char *idstr);
void tsdbReaderClose2(STsdbReader *pReader);
int32_t tsdbNextDataBlock2(STsdbReader *pReader, bool *hasNext);
int32_t tsdbRetrieveDatablockSMA2(STsdbReader *pReader, SSDataBlock *pDataBlock, bool *allHave, bool *hasNullSMA);
void tsdbReleaseDataBlock2(STsdbReader *pReader);
SSDataBlock *tsdbRetrieveDataBlock2(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);
int32_t tsdbReaderReset2(STsdbReader *pReader, SQueryTableDataCond *pCond);
int32_t tsdbGetFileBlocksDistInfo2(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
int64_t tsdbGetNumOfRowsInMemTable2(STsdbReader *pHandle);
void *tsdbGetIdx2(SMeta *pMeta);
void *tsdbGetIvtIdx2(SMeta *pMeta);
uint64_t tsdbGetReaderMaxVersion2(STsdbReader *pReader);
void tsdbReaderSetCloseFlag2(STsdbReader *pReader);
int64_t tsdbGetLastTimestamp2(SVnode *pVnode, void *pTableList, int32_t numOfTables, const char *pIdStr);
//======================================================================================================================
int32_t tsdbReuseCacherowsReader(void *pReader, void *pTableIdList, int32_t numOfTables);
int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, void *pTableIdList, int32_t numOfTables, int32_t numOfCols,
SArray *pCidList, int32_t *pSlotIds, uint64_t suid, void **pReader, const char *idstr);
......
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ df57768a
......@@ -16,6 +16,9 @@
#ifndef _TD_VNODE_TSDB_H_
#define _TD_VNODE_TSDB_H_
// #include "../tsdb/tsdbFile2.h"
// #include "../tsdb/tsdbMerge.h"
// #include "../tsdb/tsdbSttFileRW.h"
#include "tsimplehash.h"
#include "vnodeInt.h"
......@@ -75,9 +78,8 @@ typedef struct STsdbFilterInfo STsdbFilterInfo;
#define TSDBROW_ROW_FMT ((int8_t)0x0)
#define TSDBROW_COL_FMT ((int8_t)0x1)
#define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F)
#define TSDB_MAX_SUBBLOCKS 8
#define TSDB_FHDR_SIZE 512
#define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F)
#define TSDB_FHDR_SIZE 512
#define VERSION_MIN 0
#define VERSION_MAX INT64_MAX
......@@ -165,6 +167,7 @@ void tBlockDataDestroy(SBlockData *pBlockData);
int32_t tBlockDataInit(SBlockData *pBlockData, TABLEID *pId, STSchema *pTSchema, int16_t *aCid, int32_t nCid);
void tBlockDataReset(SBlockData *pBlockData);
int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid);
int32_t tBlockDataUpdateRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema);
int32_t tBlockDataTryUpsertRow(SBlockData *pBlockData, TSDBROW *pRow, int64_t uid);
int32_t tBlockDataUpsertRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid);
void tBlockDataClear(SBlockData *pBlockData);
......@@ -198,7 +201,7 @@ int32_t tMapDataToArray(SMapData *pMapData, int32_t itemSize, int32_t (*tGetItem
// other
int32_t tsdbKeyFid(TSKEY key, int32_t minutes, int8_t precision);
void tsdbFidKeyRange(int32_t fid, int32_t minutes, int8_t precision, TSKEY *minKey, TSKEY *maxKey);
int32_t tsdbFidLevel(int32_t fid, STsdbKeepCfg *pKeepCfg, int64_t now);
int32_t tsdbFidLevel(int32_t fid, STsdbKeepCfg *pKeepCfg, int64_t nowSec);
int32_t tsdbBuildDeleteSkyline(SArray *aDelData, int32_t sidx, int32_t eidx, SArray *aSkyline);
int32_t tPutColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg);
int32_t tGetColumnDataAgg(uint8_t *p, SColumnDataAgg *pColAgg);
......@@ -302,8 +305,11 @@ int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx);
// tsdbRead.c ==============================================================================================
int32_t tsdbTakeReadSnap(STsdbReader *pReader, _query_reseek_func_t reseek, STsdbReadSnap **ppSnap);
void tsdbUntakeReadSnap(STsdbReader *pReader, STsdbReadSnap *pSnap, bool proactive);
int32_t tsdbTakeReadSnap2(STsdbReader *pReader, _query_reseek_func_t reseek, STsdbReadSnap **ppSnap);
void tsdbUntakeReadSnap2(STsdbReader *pReader, STsdbReadSnap *pSnap, bool proactive);
// tsdbMerge.c ==============================================================================================
int32_t tsdbMerge(STsdb *pTsdb);
int32_t tsdbMerge(void *arg);
// tsdbDiskData ==============================================================================================
int32_t tDiskDataBuilderCreate(SDiskDataBuilder **ppBuilder);
......@@ -362,19 +368,20 @@ typedef struct {
} SCacheFlushState;
struct STsdb {
char *path;
SVnode *pVnode;
STsdbKeepCfg keepCfg;
TdThreadRwlock rwLock;
SMemTable *mem;
SMemTable *imem;
STsdbFS fs;
SLRUCache *lruCache;
SCacheFlushState flushState;
TdThreadMutex lruMutex;
SLRUCache *biCache;
TdThreadMutex biMutex;
SRocksCache rCache;
char *path;
SVnode *pVnode;
STsdbKeepCfg keepCfg;
TdThreadRwlock rwLock;
SMemTable *mem;
SMemTable *imem;
STsdbFS fs; // old
SLRUCache *lruCache;
SCacheFlushState flushState;
TdThreadMutex lruMutex;
SLRUCache *biCache;
TdThreadMutex biMutex;
struct STFileSystem *pFS; // new
SRocksCache rCache;
};
struct TSDBKEY {
......@@ -410,6 +417,7 @@ struct STbData {
SDelData *pTail;
SMemSkipList sl;
STbData *next;
SRBTreeNode rbtn[1];
};
struct SMemTable {
......@@ -423,11 +431,10 @@ struct SMemTable {
TSKEY maxKey;
int64_t nRow;
int64_t nDel;
struct {
int32_t nTbData;
int32_t nBucket;
STbData **aBucket;
};
int32_t nTbData;
int32_t nBucket;
STbData **aBucket;
SRBTree tbDataTree[1];
};
struct TSDBROW {
......@@ -500,7 +507,7 @@ struct SDataBlk {
int32_t nRow;
int8_t hasDup;
int8_t nSubBlock;
SBlockInfo aSubBlock[TSDB_MAX_SUBBLOCKS];
SBlockInfo aSubBlock[1];
SSmaInfo smaInfo;
};
......@@ -652,12 +659,19 @@ struct SDelFWriter {
uint8_t *aBuf[1];
};
#include "tarray2.h"
//#include "tsdbFS2.h"
// struct STFileSet;
typedef struct STFileSet STFileSet;
typedef TARRAY2(STFileSet *) TFileSetArray;
struct STsdbReadSnap {
SMemTable *pMem;
SQueryNode *pNode;
SMemTable *pIMem;
SQueryNode *pINode;
STsdbFS fs;
SMemTable *pMem;
SQueryNode *pNode;
SMemTable *pIMem;
SQueryNode *pINode;
TFileSetArray *pfSetArray;
STsdbFS fs;
};
struct SDataFWriter {
......@@ -696,6 +710,7 @@ typedef struct {
typedef struct SSttBlockLoadInfo {
SBlockData blockData[2];
void *pSttStatisBlkArray;
SArray *aSttBlk;
int32_t blockIndex[2]; // to denote the loaded block in the corresponding position.
int32_t currentLoadBlockIndex;
......@@ -704,10 +719,9 @@ typedef struct SSttBlockLoadInfo {
STSchema *pSchema;
int16_t *colIds;
int32_t numOfCols;
bool checkRemainingRow;
bool checkRemainingRow; // todo: no assign value?
bool isLast;
bool sttBlockLoaded;
int32_t numOfStt;
// keep the last access position, this position may be used to reduce the binary times for
// starting last block data for a new table
......@@ -766,60 +780,107 @@ struct SDiskDataBuilder {
};
typedef struct SLDataIter {
SRBTreeNode node;
SSttBlk *pSttBlk;
SDataFReader *pReader;
int32_t iStt;
int8_t backward;
int32_t iSttBlk;
int32_t iRow;
SRowInfo rInfo;
uint64_t uid;
STimeWindow timeWindow;
SVersionRange verRange;
SSttBlockLoadInfo *pBlockLoadInfo;
bool ignoreEarlierTs;
SRBTreeNode node;
SSttBlk *pSttBlk;
int32_t iStt; // for debug purpose
int8_t backward;
int32_t iSttBlk;
int32_t iRow;
SRowInfo rInfo;
uint64_t uid;
STimeWindow timeWindow;
SVersionRange verRange;
SSttBlockLoadInfo *pBlockLoadInfo;
bool ignoreEarlierTs;
struct SSttFileReader *pReader;
} SLDataIter;
#define tMergeTreeGetRow(_t) (&((_t)->pIter->rInfo.row))
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
STimeWindow *pTimeWindow, SVersionRange *pVerRange, SSttBlockLoadInfo *pBlockLoadInfo,
bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter *pLDataIter);
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree);
bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree);
void tMergeTreeClose(SMergeTree *pMTree);
struct SSttFileReader;
typedef int32_t (*_load_tomb_fn)(STsdbReader *pReader, struct SSttFileReader *pSttFileReader,
SSttBlockLoadInfo *pLoadInfo);
typedef struct {
int8_t backward;
STsdb *pTsdb;
uint64_t suid;
uint64_t uid;
STimeWindow timewindow;
SVersionRange verRange;
bool strictTimeRange;
SArray *pSttFileBlockIterArray;
void *pCurrentFileset;
STSchema *pSchema;
int16_t *pCols;
int32_t numOfCols;
_load_tomb_fn loadTombFn;
void *pReader;
void *idstr;
} SMergeTreeConf;
int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf);
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree);
bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree);
void tMergeTreeClose(SMergeTree *pMTree);
SSttBlockLoadInfo *tCreateLastBlockLoadInfo(STSchema *pSchema, int16_t *colList, int32_t numOfCols, int32_t numOfStt);
SSttBlockLoadInfo *tCreateOneLastBlockLoadInfo(STSchema *pSchema, int16_t *colList, int32_t numOfCols);
void resetLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
void getLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo, int64_t *blocks, double *el);
void *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
void *destroySttBlockReader(SArray *pLDataIterArray, int64_t *blocks, double *el);
// tsdbCache ==============================================================================================
typedef enum {
READ_MODE_COUNT_ONLY = 0x1,
READ_MODE_ALL,
} EReadMode;
typedef struct STsdbReaderInfo {
uint64_t suid;
STSchema *pSchema;
EReadMode readMode;
uint64_t rowsNum;
STimeWindow window;
SVersionRange verRange;
int16_t order;
} STsdbReaderInfo;
typedef struct {
SArray *pTombData;
} STableLoadInfo;
struct SDataFileReader;
typedef struct SCacheRowsReader {
STsdb *pTsdb;
SVersionRange verRange;
TdThreadMutex readerMutex;
SVnode *pVnode;
STSchema *pSchema;
STSchema *pCurrSchema;
uint64_t uid;
uint64_t suid;
char **transferBuf; // todo remove it soon
int32_t numOfCols;
SArray *pCidList;
int32_t *pSlotIds;
int32_t type;
int32_t tableIndex; // currently returned result tables
STableKeyInfo *pTableList; // table id list
int32_t numOfTables;
SSttBlockLoadInfo *pLoadInfo;
SLDataIter *pDataIter;
STsdbReadSnap *pReadSnap;
SDataFReader *pDataFReader;
SDataFReader *pDataFReaderLast;
const char *idstr;
int64_t lastTs;
STsdb *pTsdb;
STsdbReaderInfo info;
TdThreadMutex readerMutex;
SVnode *pVnode;
STSchema *pSchema;
STSchema *pCurrSchema;
uint64_t uid;
char **transferBuf; // todo remove it soon
int32_t numOfCols;
SArray *pCidList;
int32_t *pSlotIds;
int32_t type;
int32_t tableIndex; // currently returned result tables
STableKeyInfo *pTableList; // table id list
int32_t numOfTables;
uint64_t *uidList;
SSHashObj *pTableMap;
SArray *pLDataIterArray;
struct SDataFileReader *pFileReader;
STFileSet *pCurFileSet;
STsdbReadSnap *pReadSnap;
char *idstr;
int64_t lastTs;
} SCacheRowsReader;
typedef struct {
......
source/dnode/vnode/src/inc/vnd.h
浏览文件 @ df57768a
......@@ -49,7 +49,8 @@ int32_t vnodeEncodeConfig(const void* pObj, SJson* pJson);
int32_t vnodeDecodeConfig(const SJson* pJson, void* pObj);
// vnodeModule.c
int32_t vnodeScheduleTask(int32_t (*execute)(void*), void* arg);
int vnodeScheduleTask(int (*execute)(void*), void* arg);
int vnodeScheduleTaskEx(int tpid, int (*execute)(void*), void* arg);
// vnodeBufPool.c
typedef struct SVBufPoolNode SVBufPoolNode;
......
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ df57768a
......@@ -180,8 +180,8 @@ SArray* metaGetSmaTbUids(SMeta* pMeta);
void* metaGetIdx(SMeta* pMeta);
void* metaGetIvtIdx(SMeta* pMeta);
int64_t metaGetTbNum(SMeta *pMeta);
void metaReaderDoInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags);
int64_t metaGetTbNum(SMeta* pMeta);
void metaReaderDoInit(SMetaReader* pReader, SMeta* pMeta, int32_t flags);
int32_t metaCreateTSma(SMeta* pMeta, int64_t version, SSmaCfg* pCfg);
int32_t metaDropTSma(SMeta* pMeta, int64_t indexUid);
......@@ -198,12 +198,12 @@ int32_t metaGetInfo(SMeta* pMeta, int64_t uid, SMetaInfo* pInfo, SMetaReader* pR
int tsdbOpen(SVnode* pVnode, STsdb** ppTsdb, const char* dir, STsdbKeepCfg* pKeepCfg, int8_t rollback);
int tsdbClose(STsdb** pTsdb);
int32_t tsdbBegin(STsdb* pTsdb);
int32_t tsdbPrepareCommit(STsdb* pTsdb);
int32_t tsdbCommit(STsdb* pTsdb, SCommitInfo* pInfo);
// int32_t tsdbPrepareCommit(STsdb* pTsdb);
// int32_t tsdbCommit(STsdb* pTsdb, SCommitInfo* pInfo);
int32_t tsdbCacheCommit(STsdb* pTsdb);
int32_t tsdbCompact(STsdb* pTsdb, SCompactInfo* pInfo);
int32_t tsdbFinishCommit(STsdb* pTsdb);
int32_t tsdbRollbackCommit(STsdb* pTsdb);
// int32_t tsdbFinishCommit(STsdb* pTsdb);
// int32_t tsdbRollbackCommit(STsdb* pTsdb);
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq2* pMsg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq2* pMsg, SSubmitRsp2* pRsp);
int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitTbData* pSubmitTbData, int32_t* affectedRows);
......
source/dnode/vnode/src/sma/smaCommit.c
浏览文件 @ df57768a
......@@ -103,15 +103,16 @@ _exit:
return code;
}
int32_t smaFinishCommit(SSma *pSma) {
extern int32_t tsdbCommitCommit(STsdb *tsdb);
int32_t smaFinishCommit(SSma *pSma) {
int32_t code = 0;
int32_t lino = 0;
SVnode *pVnode = pSma->pVnode;
if (VND_RSMA1(pVnode) && (code = tsdbFinishCommit(VND_RSMA1(pVnode))) < 0) {
if (VND_RSMA1(pVnode) && (code = tsdbCommitCommit(VND_RSMA1(pVnode))) < 0) {
TSDB_CHECK_CODE(code, lino, _exit);
}
if (VND_RSMA2(pVnode) && (code = tsdbFinishCommit(VND_RSMA2(pVnode))) < 0) {
if (VND_RSMA2(pVnode) && (code = tsdbCommitCommit(VND_RSMA2(pVnode))) < 0) {
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
......@@ -130,6 +131,7 @@ _exit:
* @param isCommit
* @return int32_t
*/
extern int32_t tsdbPreCommit(STsdb *tsdb);
static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
int32_t code = 0;
int32_t lino = 0;
......@@ -186,11 +188,11 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
// all rsma results are written completely
STsdb *pTsdb = NULL;
if ((pTsdb = VND_RSMA1(pSma->pVnode))) {
code = tsdbPrepareCommit(pTsdb);
code = tsdbPreCommit(pTsdb);
TSDB_CHECK_CODE(code, lino, _exit);
}
if ((pTsdb = VND_RSMA2(pSma->pVnode))) {
code = tsdbPrepareCommit(pTsdb);
code = tsdbPreCommit(pTsdb);
TSDB_CHECK_CODE(code, lino, _exit);
}
......@@ -207,6 +209,7 @@ _exit:
* @param pSma
* @return int32_t
*/
extern int32_t tsdbCommitBegin(STsdb *tsdb, SCommitInfo *info);
static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
int32_t code = 0;
int32_t lino = 0;
......@@ -217,10 +220,10 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
goto _exit;
}
code = tsdbCommit(VND_RSMA1(pVnode), pInfo);
code = tsdbCommitBegin(VND_RSMA1(pVnode), pInfo);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbCommit(VND_RSMA2(pVnode), pInfo);
code = tsdbCommitBegin(VND_RSMA2(pVnode), pInfo);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
......
source/dnode/vnode/src/sma/smaRollup.c
浏览文件 @ df57768a
......@@ -264,7 +264,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
return TSDB_CODE_FAILED;
}
SReadHandle handle = { .vnode = pVnode, .initTqReader = 1, .pStateBackend = pStreamState };
SReadHandle handle = {.vnode = pVnode, .initTqReader = 1, .pStateBackend = pStreamState};
initStorageAPI(&handle.api);
pRSmaInfo->taskInfo[idx] = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle, TD_VID(pVnode));
......@@ -572,8 +572,8 @@ int32_t smaDoRetention(SSma *pSma, int64_t now) {
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
if (pSma->pRSmaTsdb[i]) {
code = tsdbDoRetention(pSma->pRSmaTsdb[i], now);
if (code) goto _end;
// code = tsdbDoRetention(pSma->pRSmaTsdb[i], now);
// if (code) goto _end;
}
}
......@@ -612,7 +612,6 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
blockDebugShowDataBlocks(pResList, flag);
#endif
for (int32_t i = 0; i < taosArrayGetSize(pResList); ++i) {
output = taosArrayGetP(pResList, i);
smaDebug("vgId:%d, result block, uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%" PRIi64, SMA_VID(pSma),
output->info.id.uid, output->info.id.groupId, output->info.rows);
......@@ -1114,8 +1113,8 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
}
if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) {
smaWarn("rsma fetch task not start since rsma stat already destroyed, rsetId:%d refId:%" PRIi64 ")",
smaMgmt.rsetId, pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
smaWarn("rsma fetch task not start since rsma stat already destroyed, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
return;
}
......
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
浏览文件 @ df57768a
......@@ -17,6 +17,7 @@
#include "tarray.h"
#include "tcommon.h"
#include "tsdb.h"
#include "tsdbDataFileRW.h"
#define HASTYPE(_type, _t) (((_type) & (_t)) == (_t))
......@@ -124,11 +125,29 @@ int32_t tsdbReuseCacherowsReader(void* reader, void* pTableIdList, int32_t numOf
pReader->numOfTables = numOfTables;
pReader->lastTs = INT64_MIN;
resetLastBlockLoadInfo(pReader->pLoadInfo);
int64_t blocks;
double elapse;
pReader->pLDataIterArray = destroySttBlockReader(pReader->pLDataIterArray, &blocks, &elapse);
pReader->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
return TSDB_CODE_SUCCESS;
}
static int32_t uidComparFunc(const void* p1, const void* p2) {
uint64_t pu1 = *(uint64_t*)p1;
uint64_t pu2 = *(uint64_t*)p2;
if (pu1 == pu2) {
return 0;
} else {
return (pu1 < pu2) ? -1 : 1;
}
}
static void freeTableInfoFunc(void* param) {
void** p = (void**)param;
taosMemoryFreeClear(*p);
}
int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList, int32_t numOfTables, int32_t numOfCols,
SArray* pCidList, int32_t* pSlotIds, uint64_t suid, void** pReader, const char* idstr) {
*pReader = NULL;
......@@ -140,11 +159,11 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
p->type = type;
p->pVnode = pVnode;
p->pTsdb = p->pVnode->pTsdb;
p->verRange = (SVersionRange){.minVer = 0, .maxVer = UINT64_MAX};
p->info.verRange = (SVersionRange){.minVer = 0, .maxVer = UINT64_MAX};
p->info.suid = suid;
p->numOfCols = numOfCols;
p->pCidList = pCidList;
p->pSlotIds = pSlotIds;
p->suid = suid;
if (numOfTables == 0) {
*pReader = p;
......@@ -154,6 +173,27 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
p->pTableList = pTableIdList;
p->numOfTables = numOfTables;
p->pTableMap = tSimpleHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));
if (p->pTableMap == NULL) {
tsdbCacherowsReaderClose(p);
return TSDB_CODE_OUT_OF_MEMORY;
}
p->uidList = taosMemoryMalloc(numOfTables * sizeof(uint64_t));
if (p->uidList == NULL) {
tsdbCacherowsReaderClose(p);
return TSDB_CODE_OUT_OF_MEMORY;
}
for (int32_t i = 0; i < numOfTables; ++i) {
uint64_t uid = p->pTableList[i].uid;
p->uidList[i] = uid;
STableLoadInfo* pInfo = taosMemoryCalloc(1, sizeof(STableLoadInfo));
tSimpleHashPut(p->pTableMap, &uid, sizeof(uint64_t), &pInfo, POINTER_BYTES);
}
tSimpleHashSetFreeFp(p->pTableMap, freeTableInfoFunc);
taosSort(p->uidList, numOfTables, sizeof(uint64_t), uidComparFunc);
int32_t code = setTableSchema(p, suid, idstr);
if (code != TSDB_CODE_SUCCESS) {
tsdbCacherowsReaderClose(p);
......@@ -178,14 +218,8 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
SVnodeCfg* pCfg = &((SVnode*)pVnode)->config;
int32_t numOfStt = pCfg->sttTrigger;
p->pLoadInfo = tCreateLastBlockLoadInfo(p->pSchema, NULL, 0, numOfStt);
if (p->pLoadInfo == NULL) {
tsdbCacherowsReaderClose(p);
return TSDB_CODE_OUT_OF_MEMORY;
}
p->pDataIter = taosMemoryCalloc(pCfg->sttTrigger, sizeof(SLDataIter));
if (p->pDataIter == NULL) {
p->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
if (p->pLDataIterArray == NULL) {
tsdbCacherowsReaderClose(p);
return TSDB_CODE_OUT_OF_MEMORY;
}
......@@ -214,14 +248,34 @@ void* tsdbCacherowsReaderClose(void* pReader) {
taosMemoryFree(p->pSchema);
}
taosMemoryFree(p->pDataIter);
taosMemoryFree(p->pCurrSchema);
destroyLastBlockLoadInfo(p->pLoadInfo);
int64_t loadBlocks = 0;
double elapse = 0;
destroySttBlockReader(p->pLDataIterArray, &loadBlocks, &elapse);
if (p->pFileReader) {
tsdbDataFileReaderClose(&p->pFileReader);
p->pFileReader = NULL;
}
taosMemoryFree((void*)p->idstr);
taosThreadMutexDestroy(&p->readerMutex);
if (p->pTableMap) {
void* pe = NULL;
int32_t iter = 0;
while ((pe = tSimpleHashIterate(p->pTableMap, pe, &iter)) != NULL) {
STableLoadInfo* pInfo = *(STableLoadInfo**)pe;
pInfo->pTombData = taosArrayDestroy(pInfo->pTombData);
}
tSimpleHashCleanup(p->pTableMap);
}
if (p->uidList) {
taosMemoryFree(p->uidList);
}
taosMemoryFree(pReader);
return NULL;
}
......@@ -298,12 +352,10 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
}
taosThreadMutexLock(&pr->readerMutex);
code = tsdbTakeReadSnap((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap);
code = tsdbTakeReadSnap2((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap);
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
pr->pDataFReader = NULL;
pr->pDataFReaderLast = NULL;
int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
......@@ -424,11 +476,13 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
}
_end:
tsdbDataFReaderClose(&pr->pDataFReaderLast);
tsdbDataFReaderClose(&pr->pDataFReader);
tsdbUntakeReadSnap2((STsdbReader*)pr, pr->pReadSnap, true);
int64_t loadBlocks = 0;
double elapse = 0;
pr->pLDataIterArray = destroySttBlockReader(pr->pLDataIterArray, &loadBlocks, &elapse);
pr->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
resetLastBlockLoadInfo(pr->pLoadInfo);
tsdbUntakeReadSnap((STsdbReader*)pr, pr->pReadSnap, true);
taosThreadMutexUnlock(&pr->readerMutex);
if (pRes != NULL) {
......
source/dnode/vnode/src/tsdb/tsdbCommit2.c 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "tsdbCommit2.h"
// extern dependencies
typedef struct {
STsdb *tsdb;
TFileSetArray *fsetArr;
TFileOpArray fopArray[1];
// SSkmInfo skmTb[1];
// SSkmInfo skmRow[1];
int32_t minutes;
int8_t precision;
int32_t minRow;
int32_t maxRow;
int8_t cmprAlg;
int32_t sttTrigger;
int32_t szPage;
int64_t compactVersion;
struct {
int64_t cid;
int64_t now;
TSKEY nextKey;
int32_t fid;
int32_t expLevel;
SDiskID did;
TSKEY minKey;
TSKEY maxKey;
STFileSet *fset;
TABLEID tbid[1];
bool hasTSData;
} ctx[1];
// reader
SSttFileReader *sttReader;
// iter
TTsdbIterArray dataIterArray[1];
SIterMerger *dataIterMerger;
TTsdbIterArray tombIterArray[1];
SIterMerger *tombIterMerger;
// writer
SFSetWriter *writer;
} SCommitter2;
static int32_t tsdbCommitOpenWriter(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
SFSetWriterConfig config = {
.tsdb = committer->tsdb,
.toSttOnly = true,
.compactVersion = committer->compactVersion,
.minRow = committer->minRow,
.maxRow = committer->maxRow,
.szPage = committer->szPage,
.cmprAlg = committer->cmprAlg,
.fid = committer->ctx->fid,
.cid = committer->ctx->cid,
.did = committer->ctx->did,
.level = 0,
};
if (committer->sttTrigger == 1) {
config.toSttOnly = false;
if (committer->ctx->fset) {
for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ftype++) {
if (committer->ctx->fset->farr[ftype] != NULL) {
config.files[ftype].exist = true;
config.files[ftype].file = committer->ctx->fset->farr[ftype]->f[0];
}
}
}
}
code = tsdbFSetWriterOpen(&config, &committer->writer);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
}
return code;
}
static int32_t tsdbCommitCloseWriter(SCommitter2 *committer) {
return tsdbFSetWriterClose(&committer->writer, 0, committer->fopArray);
}
static int32_t tsdbCommitTSData(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
int64_t numOfRow = 0;
SMetaInfo info;
committer->ctx->hasTSData = false;
committer->ctx->tbid->suid = 0;
committer->ctx->tbid->uid = 0;
for (SRowInfo *row; (row = tsdbIterMergerGetData(committer->dataIterMerger)) != NULL;) {
if (row->uid != committer->ctx->tbid->uid) {
committer->ctx->tbid->suid = row->suid;
committer->ctx->tbid->uid = row->uid;
if (metaGetInfo(committer->tsdb->pVnode->pMeta, row->uid, &info, NULL) != 0) {
code = tsdbIterMergerSkipTableData(committer->dataIterMerger, committer->ctx->tbid);
TSDB_CHECK_CODE(code, lino, _exit);
continue;
}
}
int64_t ts = TSDBROW_TS(&row->row);
if (ts > committer->ctx->maxKey) {
committer->ctx->nextKey = TMIN(committer->ctx->nextKey, ts);
code = tsdbIterMergerSkipTableData(committer->dataIterMerger, committer->ctx->tbid);
TSDB_CHECK_CODE(code, lino, _exit);
continue;
}
committer->ctx->hasTSData = true;
numOfRow++;
code = tsdbFSetWriteRow(committer->writer, row);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbIterMergerNext(committer->dataIterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
} else {
tsdbDebug("vgId:%d fid:%d commit %" PRId64 " rows", TD_VID(committer->tsdb->pVnode), committer->ctx->fid, numOfRow);
}
return code;
}
static int32_t tsdbCommitTombData(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
int64_t numRecord = 0;
SMetaInfo info;
if (committer->ctx->fset == NULL && !committer->ctx->hasTSData) {
return 0;
}
committer->ctx->tbid->suid = 0;
committer->ctx->tbid->uid = 0;
for (STombRecord *record; (record = tsdbIterMergerGetTombRecord(committer->tombIterMerger));) {
if (record->uid != committer->ctx->tbid->uid) {
committer->ctx->tbid->suid = record->suid;
committer->ctx->tbid->uid = record->uid;
if (metaGetInfo(committer->tsdb->pVnode->pMeta, record->uid, &info, NULL) != 0) {
code = tsdbIterMergerSkipTableData(committer->dataIterMerger, committer->ctx->tbid);
TSDB_CHECK_CODE(code, lino, _exit);
continue;
}
}
if (record->ekey < committer->ctx->minKey) {
goto _next;
} else if (record->skey > committer->ctx->maxKey) {
committer->ctx->maxKey = TMIN(record->skey, committer->ctx->maxKey);
goto _next;
}
if (record->ekey > committer->ctx->maxKey) {
committer->ctx->maxKey = committer->ctx->maxKey + 1;
}
if (record->ekey > committer->ctx->maxKey) {
committer->ctx->nextKey = record->ekey;
}
record->skey = TMAX(record->skey, committer->ctx->minKey);
record->ekey = TMIN(record->ekey, committer->ctx->maxKey);
numRecord++;
code = tsdbFSetWriteTombRecord(committer->writer, record);
TSDB_CHECK_CODE(code, lino, _exit);
_next:
code = tsdbIterMergerNext(committer->tombIterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
} else {
tsdbDebug("vgId:%d fid:%d commit %" PRId64 " tomb records", TD_VID(committer->tsdb->pVnode), committer->ctx->fid,
numRecord);
}
return code;
}
static int32_t tsdbCommitOpenReader(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
ASSERT(committer->sttReader == NULL);
if (committer->ctx->fset == NULL //
|| committer->sttTrigger > 1 //
|| TARRAY2_SIZE(committer->ctx->fset->lvlArr) == 0 //
) {
return 0;
}
ASSERT(TARRAY2_SIZE(committer->ctx->fset->lvlArr) == 1);
SSttLvl *lvl = TARRAY2_FIRST(committer->ctx->fset->lvlArr);
ASSERT(lvl->level == 0);
if (TARRAY2_SIZE(lvl->fobjArr) == 0) {
return 0;
}
ASSERT(TARRAY2_SIZE(lvl->fobjArr) == 1);
STFileObj *fobj = TARRAY2_FIRST(lvl->fobjArr);
SSttFileReaderConfig config = {
.tsdb = committer->tsdb,
.szPage = committer->szPage,
.file = fobj->f[0],
};
code = tsdbSttFileReaderOpen(fobj->fname, &config, &committer->sttReader);
TSDB_CHECK_CODE(code, lino, _exit);
STFileOp op = {
.optype = TSDB_FOP_REMOVE,
.fid = fobj->f->fid,
.of = fobj->f[0],
};
code = TARRAY2_APPEND(committer->fopArray, op);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
}
return code;
}
static int32_t tsdbCommitCloseReader(SCommitter2 *committer) { return tsdbSttFileReaderClose(&committer->sttReader); }
static int32_t tsdbCommitOpenIter(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
ASSERT(TARRAY2_SIZE(committer->dataIterArray) == 0);
ASSERT(committer->dataIterMerger == NULL);
ASSERT(TARRAY2_SIZE(committer->tombIterArray) == 0);
ASSERT(committer->tombIterMerger == NULL);
STsdbIter *iter;
STsdbIterConfig config = {0};
// mem data iter
config.type = TSDB_ITER_TYPE_MEMT;
config.memt = committer->tsdb->imem;
config.from->ts = committer->ctx->minKey;
config.from->version = VERSION_MIN;
code = tsdbIterOpen(&config, &iter);
TSDB_CHECK_CODE(code, lino, _exit);
code = TARRAY2_APPEND(committer->dataIterArray, iter);
TSDB_CHECK_CODE(code, lino, _exit);
// mem tomb iter
config.type = TSDB_ITER_TYPE_MEMT_TOMB;
config.memt = committer->tsdb->imem;
code = tsdbIterOpen(&config, &iter);
TSDB_CHECK_CODE(code, lino, _exit);
code = TARRAY2_APPEND(committer->tombIterArray, iter);
TSDB_CHECK_CODE(code, lino, _exit);
// STT
if (committer->sttReader) {
// data iter
config.type = TSDB_ITER_TYPE_STT;
config.sttReader = committer->sttReader;
code = tsdbIterOpen(&config, &iter);
TSDB_CHECK_CODE(code, lino, _exit);
code = TARRAY2_APPEND(committer->dataIterArray, iter);
TSDB_CHECK_CODE(code, lino, _exit);
// tomb iter
config.type = TSDB_ITER_TYPE_STT_TOMB;
config.sttReader = committer->sttReader;
code = tsdbIterOpen(&config, &iter);
TSDB_CHECK_CODE(code, lino, _exit);
code = TARRAY2_APPEND(committer->tombIterArray, iter);
TSDB_CHECK_CODE(code, lino, _exit);
}
// open merger
code = tsdbIterMergerOpen(committer->dataIterArray, &committer->dataIterMerger, false);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbIterMergerOpen(committer->tombIterArray, &committer->tombIterMerger, true);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
}
return code;
}
static int32_t tsdbCommitCloseIter(SCommitter2 *committer) {
tsdbIterMergerClose(&committer->tombIterMerger);
tsdbIterMergerClose(&committer->dataIterMerger);
TARRAY2_CLEAR(committer->tombIterArray, tsdbIterClose);
TARRAY2_CLEAR(committer->dataIterArray, tsdbIterClose);
return 0;
}
static int32_t tsdbCommitFileSetBegin(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
STsdb *tsdb = committer->tsdb;
committer->ctx->fid = tsdbKeyFid(committer->ctx->nextKey, committer->minutes, committer->precision);
committer->ctx->expLevel = tsdbFidLevel(committer->ctx->fid, &tsdb->keepCfg, committer->ctx->now);
tsdbFidKeyRange(committer->ctx->fid, committer->minutes, committer->precision, &committer->ctx->minKey,
&committer->ctx->maxKey);
code = tfsAllocDisk(committer->tsdb->pVnode->pTfs, committer->ctx->expLevel, &committer->ctx->did);
TSDB_CHECK_CODE(code, lino, _exit);
tfsMkdirRecurAt(committer->tsdb->pVnode->pTfs, committer->tsdb->path, committer->ctx->did);
STFileSet fset = {.fid = committer->ctx->fid};
committer->ctx->fset = &fset;
STFileSet **fsetPtr = TARRAY2_SEARCH(committer->fsetArr, &committer->ctx->fset, tsdbTFileSetCmprFn, TD_EQ);
committer->ctx->fset = (fsetPtr == NULL) ? NULL : *fsetPtr;
committer->ctx->tbid->suid = 0;
committer->ctx->tbid->uid = 0;
ASSERT(TARRAY2_SIZE(committer->dataIterArray) == 0);
ASSERT(committer->dataIterMerger == NULL);
ASSERT(committer->writer == NULL);
code = tsdbCommitOpenReader(committer);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbCommitOpenIter(committer);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbCommitOpenWriter(committer);
TSDB_CHECK_CODE(code, lino, _exit);
// reset nextKey
committer->ctx->nextKey = TSKEY_MAX;
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
} else {
tsdbDebug("vgId:%d %s done, fid:%d minKey:%" PRId64 " maxKey:%" PRId64 " expLevel:%d", TD_VID(tsdb->pVnode),
__func__, committer->ctx->fid, committer->ctx->minKey, committer->ctx->maxKey, committer->ctx->expLevel);
}
return 0;
}
static int32_t tsdbCommitFileSetEnd(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
code = tsdbCommitCloseWriter(committer);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbCommitCloseIter(committer);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbCommitCloseReader(committer);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
} else {
tsdbDebug("vgId:%d %s done, fid:%d", TD_VID(committer->tsdb->pVnode), __func__, committer->ctx->fid);
}
return code;
}
static int32_t tsdbCommitFileSet(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
// fset commit start
code = tsdbCommitFileSetBegin(committer);
TSDB_CHECK_CODE(code, lino, _exit);
// commit fset
code = tsdbCommitTSData(committer);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbCommitTombData(committer);
TSDB_CHECK_CODE(code, lino, _exit);
// fset commit end
code = tsdbCommitFileSetEnd(committer);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
} else {
tsdbDebug("vgId:%d %s done, fid:%d", TD_VID(committer->tsdb->pVnode), __func__, committer->ctx->fid);
}
return code;
}
static int32_t tsdbOpenCommitter(STsdb *tsdb, SCommitInfo *info, SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
memset(committer, 0, sizeof(committer[0]));
committer->tsdb = tsdb;
code = tsdbFSCreateCopySnapshot(tsdb->pFS, &committer->fsetArr);
TSDB_CHECK_CODE(code, lino, _exit);
committer->minutes = tsdb->keepCfg.days;
committer->precision = tsdb->keepCfg.precision;
committer->minRow = info->info.config.tsdbCfg.minRows;
committer->maxRow = info->info.config.tsdbCfg.maxRows;
committer->cmprAlg = info->info.config.tsdbCfg.compression;
committer->sttTrigger = info->info.config.sttTrigger;
committer->szPage = info->info.config.tsdbPageSize;
committer->compactVersion = INT64_MAX;
committer->ctx->cid = tsdbFSAllocEid(tsdb->pFS);
committer->ctx->now = taosGetTimestampSec();
committer->ctx->nextKey = tsdb->imem->minKey;
if (tsdb->imem->nDel > 0) {
SRBTreeIter iter[1] = {tRBTreeIterCreate(tsdb->imem->tbDataTree, 1)};
for (SRBTreeNode *node = tRBTreeIterNext(iter); node; node = tRBTreeIterNext(iter)) {
STbData *tbData = TCONTAINER_OF(node, STbData, rbtn);
for (SDelData *delData = tbData->pHead; delData; delData = delData->pNext) {
if (delData->sKey < committer->ctx->nextKey) {
committer->ctx->nextKey = delData->sKey;
}
}
}
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
} else {
tsdbDebug("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);
}
return code;
}
static int32_t tsdbCloseCommitter(SCommitter2 *committer, int32_t eno) {
int32_t code = 0;
int32_t lino = 0;
if (eno == 0) {
code = tsdbFSEditBegin(committer->tsdb->pFS, committer->fopArray, TSDB_FEDIT_COMMIT);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
// TODO
ASSERT(0);
}
ASSERT(committer->writer == NULL);
ASSERT(committer->dataIterMerger == NULL);
ASSERT(committer->tombIterMerger == NULL);
TARRAY2_DESTROY(committer->dataIterArray, NULL);
TARRAY2_DESTROY(committer->tombIterArray, NULL);
TARRAY2_DESTROY(committer->fopArray, NULL);
tsdbFSDestroyCopySnapshot(&committer->fsetArr);
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s, eid:%" PRId64, TD_VID(committer->tsdb->pVnode), __func__, lino,
tstrerror(code), committer->ctx->cid);
} else {
tsdbDebug("vgId:%d %s done, eid:%" PRId64, TD_VID(committer->tsdb->pVnode), __func__, committer->ctx->cid);
}
return code;
}
int32_t tsdbPreCommit(STsdb *tsdb) {
taosThreadRwlockWrlock(&tsdb->rwLock);
ASSERT(tsdb->imem == NULL);
tsdb->imem = tsdb->mem;
tsdb->mem = NULL;
taosThreadRwlockUnlock(&tsdb->rwLock);
return 0;
}
int32_t tsdbCommitBegin(STsdb *tsdb, SCommitInfo *info) {
if (!tsdb) return 0;
int32_t code = 0;
int32_t lino = 0;
SMemTable *imem = tsdb->imem;
int64_t nRow = imem->nRow;
int64_t nDel = imem->nDel;
if (nRow == 0 && nDel == 0) {
taosThreadRwlockWrlock(&tsdb->rwLock);
tsdb->imem = NULL;
taosThreadRwlockUnlock(&tsdb->rwLock);
tsdbUnrefMemTable(imem, NULL, true);
} else {
SCommitter2 committer[1];
code = tsdbOpenCommitter(tsdb, info, committer);
TSDB_CHECK_CODE(code, lino, _exit);
while (committer->ctx->nextKey != TSKEY_MAX) {
code = tsdbCommitFileSet(committer);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbCloseCommitter(committer, code);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
} else {
tsdbInfo("vgId:%d %s done, nRow:%" PRId64 " nDel:%" PRId64, TD_VID(tsdb->pVnode), __func__, nRow, nDel);
}
return code;
}
int32_t tsdbCommitCommit(STsdb *tsdb) {
int32_t code = 0;
int32_t lino = 0;
if (tsdb->imem == NULL) goto _exit;
SMemTable *pMemTable = tsdb->imem;
taosThreadRwlockWrlock(&tsdb->rwLock);
code = tsdbFSEditCommit(tsdb->pFS);
if (code) {
taosThreadRwlockUnlock(&tsdb->rwLock);
TSDB_CHECK_CODE(code, lino, _exit);
}
tsdb->imem = NULL;
taosThreadRwlockUnlock(&tsdb->rwLock);
tsdbUnrefMemTable(pMemTable, NULL, true);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
} else {
tsdbInfo("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);
}
return code;
}
int32_t tsdbCommitAbort(STsdb *pTsdb) {
int32_t code = 0;
int32_t lino = 0;
if (pTsdb->imem == NULL) goto _exit;
code = tsdbFSEditAbort(pTsdb->pFS);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
} else {
tsdbInfo("vgId:%d %s done", TD_VID(pTsdb->pVnode), __func__);
}
return code;
}
source/dnode/vnode/src/tsdb/tsdbCommit2.h 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "tsdbDataFileRW.h"
#include "tsdbFS2.h"
#include "tsdbFSetRW.h"
#include "tsdbIter.h"
#include "tsdbSttFileRW.h"
#ifndef _TSDB_COMMIT_H_
#define _TSDB_COMMIT_H_
#ifdef __cplusplus
extern "C" {
#endif
#ifdef __cplusplus
}
#endif
#endif /*_TSDB_COMMIT_H_*/
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbDataFileRW.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbDataFileRW.h 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "tsdbDef.h"
#include "tsdbFSet2.h"
#include "tsdbSttFileRW.h"
#include "tsdbUtil2.h"
#ifndef _TSDB_DATA_FILE_RW_H
#define _TSDB_DATA_FILE_RW_H
#ifdef __cplusplus
extern "C" {
#endif
typedef TARRAY2(SBlockIdx) TBlockIdxArray;
typedef TARRAY2(SDataBlk) TDataBlkArray;
typedef TARRAY2(SColumnDataAgg) TColumnDataAggArray;
typedef struct {
SFDataPtr brinBlkPtr[1];
SFDataPtr rsrvd[2];
} SHeadFooter;
typedef struct {
SFDataPtr tombBlkPtr[1];
SFDataPtr rsrvd[2];
} STombFooter;
// SDataFileReader =============================================
typedef struct SDataFileReader SDataFileReader;
typedef struct SDataFileReaderConfig {
STsdb *tsdb;
int32_t szPage;
struct {
bool exist;
STFile file;
} files[TSDB_FTYPE_MAX];
uint8_t **bufArr;
} SDataFileReaderConfig;
int32_t tsdbDataFileReaderOpen(const char *fname[/* TSDB_FTYPE_MAX */], const SDataFileReaderConfig *config,
SDataFileReader **reader);
int32_t tsdbDataFileReaderClose(SDataFileReader **reader);
// .head
int32_t tsdbDataFileReadBrinBlk(SDataFileReader *reader, const TBrinBlkArray **brinBlkArray);
int32_t tsdbDataFileReadBrinBlock(SDataFileReader *reader, const SBrinBlk *brinBlk, SBrinBlock *brinBlock);
// .data
int32_t tsdbDataFileReadBlockData(SDataFileReader *reader, const SBrinRecord *record, SBlockData *bData);
int32_t tsdbDataFileReadBlockDataByColumn(SDataFileReader *reader, const SBrinRecord *record, SBlockData *bData,
STSchema *pTSchema, int16_t cids[], int32_t ncid);
// .sma
int32_t tsdbDataFileReadBlockSma(SDataFileReader *reader, const SBrinRecord *record,
TColumnDataAggArray *columnDataAggArray);
// .tomb
int32_t tsdbDataFileReadTombBlk(SDataFileReader *reader, const TTombBlkArray **tombBlkArray);
int32_t tsdbDataFileReadTombBlock(SDataFileReader *reader, const STombBlk *tombBlk, STombBlock *tData);
// SDataFileWriter =============================================
typedef struct SDataFileWriter SDataFileWriter;
typedef struct SDataFileWriterConfig {
STsdb *tsdb;
int8_t cmprAlg;
int32_t maxRow;
int32_t szPage;
int32_t fid;
int64_t cid;
SDiskID did;
int64_t compactVersion;
struct {
bool exist;
STFile file;
} files[TSDB_FTYPE_MAX];
SSkmInfo *skmTb;
SSkmInfo *skmRow;
uint8_t **bufArr;
} SDataFileWriterConfig;
int32_t tsdbDataFileWriterOpen(const SDataFileWriterConfig *config, SDataFileWriter **writer);
int32_t tsdbDataFileWriterClose(SDataFileWriter **writer, bool abort, TFileOpArray *opArr);
int32_t tsdbDataFileWriteRow(SDataFileWriter *writer, SRowInfo *row);
int32_t tsdbDataFileWriteBlockData(SDataFileWriter *writer, SBlockData *bData);
int32_t tsdbDataFileFlush(SDataFileWriter *writer);
int32_t tsdbDataFileWriteTombRecord(SDataFileWriter *writer, const STombRecord *record);
#ifdef __cplusplus
}
#endif
#endif /*_TSDB_DATA_FILE_RW_H*/
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbDef.h 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "tarray2.h"
#include "tsdb.h"
#ifndef _TD_TSDB_DEF_H_
#define _TD_TSDB_DEF_H_
#ifdef __cplusplus
extern "C" {
#endif
#define TSDB_ERROR_LOG(vid, lino, code) \
tsdbError("vgId:%d %s failed at line %d since %s", vid, __func__, lino, tstrerror(code))
typedef struct SFDataPtr {
int64_t offset;
int64_t size;
} SFDataPtr;
extern int32_t tsdbOpenFile(const char *path, int32_t szPage, int32_t flag, STsdbFD **ppFD);
extern void tsdbCloseFile(STsdbFD **ppFD);
extern int32_t tsdbWriteFile(STsdbFD *pFD, int64_t offset, const uint8_t *pBuf, int64_t size);
extern int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size);
extern int32_t tsdbFsyncFile(STsdbFD *pFD);
#ifdef __cplusplus
}
#endif
#endif /*_TD_TSDB_DEF_H_*/
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbFS.c
浏览文件 @ df57768a
......@@ -181,10 +181,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
TSDB_CHECK_CODE(code, lino, _exit);
}
if (size != tsdbLogicToFileSize(pTsdb->fs.pDelFile->size, pTsdb->pVnode->config.tsdbPageSize)) {
code = TSDB_CODE_FILE_CORRUPTED;
TSDB_CHECK_CODE(code, lino, _exit);
}
// if (size != tsdbLogicToFileSize(pTsdb->fs.pDelFile->size, pTsdb->pVnode->config.tsdbPageSize)) {
// code = TSDB_CODE_FILE_CORRUPTED;
// TSDB_CHECK_CODE(code, lino, _exit);
// }
}
// SArray<SDFileSet>
......@@ -199,10 +199,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (size != tsdbLogicToFileSize(pSet->pHeadF->size, pTsdb->pVnode->config.tsdbPageSize)) {
code = TSDB_CODE_FILE_CORRUPTED;
TSDB_CHECK_CODE(code, lino, _exit);
}
// if (size != tsdbLogicToFileSize(pSet->pHeadF->size, pTsdb->pVnode->config.tsdbPageSize)) {
// code = TSDB_CODE_FILE_CORRUPTED;
// TSDB_CHECK_CODE(code, lino, _exit);
// }
// data =========
tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
......@@ -210,10 +210,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (size < tsdbLogicToFileSize(pSet->pDataF->size, pTsdb->pVnode->config.tsdbPageSize)) {
code = TSDB_CODE_FILE_CORRUPTED;
TSDB_CHECK_CODE(code, lino, _exit);
}
// if (size < tsdbLogicToFileSize(pSet->pDataF->size, pTsdb->pVnode->config.tsdbPageSize)) {
// code = TSDB_CODE_FILE_CORRUPTED;
// TSDB_CHECK_CODE(code, lino, _exit);
// }
// else if (size > tsdbLogicToFileSize(pSet->pDataF->size, pTsdb->pVnode->config.tsdbPageSize)) {
// code = tsdbDFileRollback(pTsdb, pSet, TSDB_DATA_FILE);
// TSDB_CHECK_CODE(code, lino, _exit);
......@@ -225,10 +225,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (size < tsdbLogicToFileSize(pSet->pSmaF->size, pTsdb->pVnode->config.tsdbPageSize)) {
code = TSDB_CODE_FILE_CORRUPTED;
TSDB_CHECK_CODE(code, lino, _exit);
}
// if (size < tsdbLogicToFileSize(pSet->pSmaF->size, pTsdb->pVnode->config.tsdbPageSize)) {
// code = TSDB_CODE_FILE_CORRUPTED;
// TSDB_CHECK_CODE(code, lino, _exit);
// }
// else if (size > tsdbLogicToFileSize(pSet->pSmaF->size, pTsdb->pVnode->config.tsdbPageSize)) {
// code = tsdbDFileRollback(pTsdb, pSet, TSDB_SMA_FILE);
// TSDB_CHECK_CODE(code, lino, _exit);
......@@ -241,10 +241,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (size != tsdbLogicToFileSize(pSet->aSttF[iStt]->size, pTsdb->pVnode->config.tsdbPageSize)) {
code = TSDB_CODE_FILE_CORRUPTED;
TSDB_CHECK_CODE(code, lino, _exit);
}
// if (size != tsdbLogicToFileSize(pSet->aSttF[iStt]->size, pTsdb->pVnode->config.tsdbPageSize)) {
// code = TSDB_CODE_FILE_CORRUPTED;
// TSDB_CHECK_CODE(code, lino, _exit);
// }
}
}
......@@ -270,7 +270,7 @@ int32_t tDFileSetCmprFn(const void *p1, const void *p2) {
return 0;
}
static void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t) {
void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t) {
SVnode *pVnode = pTsdb->pVnode;
int32_t offset = 0;
......@@ -289,7 +289,7 @@ static void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t) {
}
}
static int32_t tsdbLoadFSFromFile(const char *fname, STsdbFS *pFS) {
static int32_t load_fs(const char *fname, STsdbFS *pFS) {
int32_t code = 0;
int32_t lino = 0;
uint8_t *pData = NULL;
......@@ -666,7 +666,7 @@ static int32_t tsdbFSApplyChange(STsdb *pTsdb, STsdbFS *pFS) {
taosArrayRemove(pTsdb->fs.aDFileSet, iOld);
} else {
code = tsdbNewFileSet(pTsdb, &fSet, pSetNew);
TSDB_CHECK_CODE(code, lino, _exit)
TSDB_CHECK_CODE(code, lino, _exit);
if (taosArrayInsert(pTsdb->fs.aDFileSet, iOld, &fSet) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
......@@ -682,7 +682,7 @@ static int32_t tsdbFSApplyChange(STsdb *pTsdb, STsdbFS *pFS) {
taosArrayRemove(pTsdb->fs.aDFileSet, iOld);
} else {
code = tsdbNewFileSet(pTsdb, &fSet, pSetNew);
TSDB_CHECK_CODE(code, lino, _exit)
TSDB_CHECK_CODE(code, lino, _exit);
if (taosArrayInsert(pTsdb->fs.aDFileSet, iOld, &fSet) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
......@@ -723,7 +723,7 @@ int32_t tsdbFSCommit(STsdb *pTsdb) {
code = tsdbFSCreate(&fs);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbLoadFSFromFile(current, &fs);
code = load_fs(current, &fs);
TSDB_CHECK_CODE(code, lino, _exit);
// apply file change
......@@ -768,7 +768,7 @@ int32_t tsdbFSOpen(STsdb *pTsdb, int8_t rollback) {
tsdbGetCurrentFName(pTsdb, current, current_t);
if (taosCheckExistFile(current)) {
code = tsdbLoadFSFromFile(current, &pTsdb->fs);
code = load_fs(current, &pTsdb->fs);
TSDB_CHECK_CODE(code, lino, _exit);
if (taosCheckExistFile(current_t)) {
......
source/dnode/vnode/src/tsdb/tsdbFS2.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbFS2.h 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "tsdbFSet2.h"
#ifndef _TSDB_FILE_SYSTEM_H
#define _TSDB_FILE_SYSTEM_H
#ifdef __cplusplus
extern "C" {
#endif
/* Exposed Handle */
typedef struct STFileSystem STFileSystem;
typedef struct STFSBgTask STFSBgTask;
// typedef TARRAY2(STFileSet *) TFileSetArray;
typedef enum {
TSDB_FEDIT_COMMIT = 1, //
TSDB_FEDIT_MERGE
} EFEditT;
typedef enum {
TSDB_BG_TASK_MERGER = 1,
TSDB_BG_TASK_RETENTION,
TSDB_BG_TASK_COMPACT,
} EFSBgTaskT;
typedef enum {
TSDB_FCURRENT = 1,
TSDB_FCURRENT_C, // for commit
TSDB_FCURRENT_M, // for merge
} EFCurrentT;
/* Exposed APIs */
// open/close
int32_t tsdbOpenFS(STsdb *pTsdb, STFileSystem **fs, int8_t rollback);
int32_t tsdbCloseFS(STFileSystem **fs);
// snapshot
int32_t tsdbFSCreateCopySnapshot(STFileSystem *fs, TFileSetArray **fsetArr);
int32_t tsdbFSDestroyCopySnapshot(TFileSetArray **fsetArr);
int32_t tsdbFSCreateRefSnapshot(STFileSystem *fs, TFileSetArray **fsetArr);
int32_t tsdbFSDestroyRefSnapshot(TFileSetArray **fsetArr);
// txn
int64_t tsdbFSAllocEid(STFileSystem *fs);
int32_t tsdbFSEditBegin(STFileSystem *fs, const TFileOpArray *opArray, EFEditT etype);
int32_t tsdbFSEditCommit(STFileSystem *fs);
int32_t tsdbFSEditAbort(STFileSystem *fs);
// background task
int32_t tsdbFSScheduleBgTask(STFileSystem *fs, EFSBgTaskT type, int32_t (*run)(void *), void *arg, int64_t *taskid);
int32_t tsdbFSWaitBgTask(STFileSystem *fs, int64_t taskid);
int32_t tsdbFSWaitAllBgTask(STFileSystem *fs);
int32_t tsdbFSDisableBgTask(STFileSystem *fs);
int32_t tsdbFSEnableBgTask(STFileSystem *fs);
// other
int32_t tsdbFSGetFSet(STFileSystem *fs, int32_t fid, STFileSet **fset);
struct STFSBgTask {
EFSBgTaskT type;
int32_t (*run)(void *arg);
void *arg;
TdThreadCond done[1];
int32_t numWait;
int64_t taskid;
int64_t scheduleTime;
int64_t launchTime;
int64_t finishTime;
struct STFSBgTask *prev;
struct STFSBgTask *next;
};
/* Exposed Structs */
struct STFileSystem {
STsdb *tsdb;
tsem_t canEdit;
int32_t state;
int64_t neid;
EFEditT etype;
TFileSetArray fSetArr[1];
TFileSetArray fSetArrTmp[1];
// background task queue
TdThreadMutex mutex[1];
bool stop;
int64_t taskid;
int32_t bgTaskNum;
STFSBgTask bgTaskQueue[1];
STFSBgTask *bgTaskRunning;
};
#ifdef __cplusplus
}
#endif
#endif /*_TSDB_FILE_SYSTEM_H*/
source/dnode/vnode/src/tsdb/tsdbFSet2.c 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "tsdbFSet2.h"
int32_t tsdbSttLvlInit(int32_t level, SSttLvl **lvl) {
if (!(lvl[0] = taosMemoryMalloc(sizeof(SSttLvl)))) return TSDB_CODE_OUT_OF_MEMORY;
lvl[0]->level = level;
TARRAY2_INIT(lvl[0]->fobjArr);
return 0;
}
static void tsdbSttLvlClearFObj(void *data) { tsdbTFileObjUnref(*(STFileObj **)data); }
int32_t tsdbSttLvlClear(SSttLvl **lvl) {
if (lvl[0] != NULL) {
TARRAY2_DESTROY(lvl[0]->fobjArr, tsdbSttLvlClearFObj);
taosMemoryFree(lvl[0]);
lvl[0] = NULL;
}
return 0;
}
static int32_t tsdbSttLvlInitEx(STsdb *pTsdb, const SSttLvl *lvl1, SSttLvl **lvl) {
int32_t code = tsdbSttLvlInit(lvl1->level, lvl);
if (code) return code;
const STFileObj *fobj1;
TARRAY2_FOREACH(lvl1->fobjArr, fobj1) {
STFileObj *fobj;
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fobj);
if (code) {
tsdbSttLvlClear(lvl);
return code;
}
TARRAY2_APPEND(lvl[0]->fobjArr, fobj);
}
return 0;
}
static int32_t tsdbSttLvlInitRef(STsdb *pTsdb, const SSttLvl *lvl1, SSttLvl **lvl) {
int32_t code = tsdbSttLvlInit(lvl1->level, lvl);
if (code) return code;
STFileObj *fobj1;
TARRAY2_FOREACH(lvl1->fobjArr, fobj1) {
tsdbTFileObjRef(fobj1);
code = TARRAY2_APPEND(lvl[0]->fobjArr, fobj1);
if (code) return code;
}
return 0;
}
static void tsdbSttLvlRemoveFObj(void *data) { tsdbTFileObjRemove(*(STFileObj **)data); }
static void tsdbSttLvlRemove(SSttLvl **lvl) {
TARRAY2_DESTROY(lvl[0]->fobjArr, tsdbSttLvlRemoveFObj);
taosMemoryFree(lvl[0]);
lvl[0] = NULL;
}
static int32_t tsdbSttLvlApplyEdit(STsdb *pTsdb, const SSttLvl *lvl1, SSttLvl *lvl2) {
int32_t code = 0;
ASSERT(lvl1->level == lvl2->level);
int32_t i1 = 0, i2 = 0;
while (i1 < TARRAY2_SIZE(lvl1->fobjArr) || i2 < TARRAY2_SIZE(lvl2->fobjArr)) {
STFileObj *fobj1 = i1 < TARRAY2_SIZE(lvl1->fobjArr) ? TARRAY2_GET(lvl1->fobjArr, i1) : NULL;
STFileObj *fobj2 = i2 < TARRAY2_SIZE(lvl2->fobjArr) ? TARRAY2_GET(lvl2->fobjArr, i2) : NULL;
if (fobj1 && fobj2) {
if (fobj1->f->cid < fobj2->f->cid) {
// create a file obj
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fobj2);
if (code) return code;
code = TARRAY2_APPEND(lvl2->fobjArr, fobj2);
if (code) return code;
i1++;
i2++;
} else if (fobj1->f->cid > fobj2->f->cid) {
// remove a file obj
TARRAY2_REMOVE(lvl2->fobjArr, i2, tsdbSttLvlRemoveFObj);
} else {
if (tsdbIsSameTFile(fobj1->f, fobj2->f)) {
if (tsdbIsTFileChanged(fobj1->f, fobj2->f)) {
fobj2->f[0] = fobj1->f[0];
}
} else {
TARRAY2_REMOVE(lvl2->fobjArr, i2, tsdbSttLvlRemoveFObj);
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fobj2);
if (code) return code;
code = TARRAY2_SORT_INSERT(lvl2->fobjArr, fobj2, tsdbTFileObjCmpr);
if (code) return code;
}
i1++;
i2++;
}
} else if (fobj1) {
// create a file obj
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fobj2);
if (code) return code;
code = TARRAY2_APPEND(lvl2->fobjArr, fobj2);
if (code) return code;
i1++;
i2++;
} else {
// remove a file obj
TARRAY2_REMOVE(lvl2->fobjArr, i2, tsdbSttLvlRemoveFObj);
}
}
return 0;
}
static int32_t tsdbSttLvlCmprFn(const SSttLvl **lvl1, const SSttLvl **lvl2) {
if (lvl1[0]->level < lvl2[0]->level) return -1;
if (lvl1[0]->level > lvl2[0]->level) return 1;
return 0;
}
static int32_t tsdbSttLvlToJson(const SSttLvl *lvl, cJSON *json) {
if (cJSON_AddNumberToObject(json, "level", lvl->level) == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
cJSON *ajson = cJSON_AddArrayToObject(json, "files");
if (ajson == NULL) return TSDB_CODE_OUT_OF_MEMORY;
const STFileObj *fobj;
TARRAY2_FOREACH(lvl->fobjArr, fobj) {
cJSON *item = cJSON_CreateObject();
if (item == NULL) return TSDB_CODE_OUT_OF_MEMORY;
cJSON_AddItemToArray(ajson, item);
int32_t code = tsdbTFileToJson(fobj->f, item);
if (code) return code;
}
return 0;
}
static int32_t tsdbJsonToSttLvl(STsdb *pTsdb, const cJSON *json, SSttLvl **lvl) {
const cJSON *item1, *item2;
int32_t level;
item1 = cJSON_GetObjectItem(json, "level");
if (cJSON_IsNumber(item1)) {
level = item1->valuedouble;
} else {
return TSDB_CODE_FILE_CORRUPTED;
}
int32_t code = tsdbSttLvlInit(level, lvl);
if (code) return code;
item1 = cJSON_GetObjectItem(json, "files");
if (!cJSON_IsArray(item1)) {
tsdbSttLvlClear(lvl);
return TSDB_CODE_FILE_CORRUPTED;
}
cJSON_ArrayForEach(item2, item1) {
STFile tf;
code = tsdbJsonToTFile(item2, TSDB_FTYPE_STT, &tf);
if (code) {
tsdbSttLvlClear(lvl);
return code;
}
STFileObj *fobj;
code = tsdbTFileObjInit(pTsdb, &tf, &fobj);
if (code) {
tsdbSttLvlClear(lvl);
return code;
}
TARRAY2_APPEND(lvl[0]->fobjArr, fobj);
}
return 0;
}
int32_t tsdbTFileSetToJson(const STFileSet *fset, cJSON *json) {
int32_t code = 0;
cJSON *item1, *item2;
// fid
if (cJSON_AddNumberToObject(json, "fid", fset->fid) == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset->farr[ftype] == NULL) continue;
code = tsdbTFileToJson(fset->farr[ftype]->f, json);
if (code) return code;
}
// each level
item1 = cJSON_AddArrayToObject(json, "stt lvl");
if (item1 == NULL) return TSDB_CODE_OUT_OF_MEMORY;
const SSttLvl *lvl;
TARRAY2_FOREACH(fset->lvlArr, lvl) {
item2 = cJSON_CreateObject();
if (!item2) return TSDB_CODE_OUT_OF_MEMORY;
cJSON_AddItemToArray(item1, item2);
code = tsdbSttLvlToJson(lvl, item2);
if (code) return code;
}
return 0;
}
int32_t tsdbJsonToTFileSet(STsdb *pTsdb, const cJSON *json, STFileSet **fset) {
int32_t code;
const cJSON *item1, *item2;
int32_t fid;
STFile tf;
// fid
item1 = cJSON_GetObjectItem(json, "fid");
if (cJSON_IsNumber(item1)) {
fid = item1->valuedouble;
} else {
return TSDB_CODE_FILE_CORRUPTED;
}
code = tsdbTFileSetInit(fid, fset);
if (code) return code;
for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
code = tsdbJsonToTFile(json, ftype, &tf);
if (code == TSDB_CODE_NOT_FOUND) {
continue;
} else if (code) {
tsdbTFileSetClear(fset);
return code;
} else {
code = tsdbTFileObjInit(pTsdb, &tf, &(*fset)->farr[ftype]);
if (code) return code;
}
}
// each level
item1 = cJSON_GetObjectItem(json, "stt lvl");
if (cJSON_IsArray(item1)) {
cJSON_ArrayForEach(item2, item1) {
SSttLvl *lvl;
code = tsdbJsonToSttLvl(pTsdb, item2, &lvl);
if (code) {
tsdbTFileSetClear(fset);
return code;
}
TARRAY2_APPEND((*fset)->lvlArr, lvl);
}
} else {
return TSDB_CODE_FILE_CORRUPTED;
}
return 0;
}
// NOTE: the api does not remove file, only do memory operation
int32_t tsdbTFileSetEdit(STsdb *pTsdb, STFileSet *fset, const STFileOp *op) {
int32_t code = 0;
if (op->optype == TSDB_FOP_CREATE) {
// create a new file
STFileObj *fobj;
code = tsdbTFileObjInit(pTsdb, &op->nf, &fobj);
if (code) return code;
if (fobj->f->type == TSDB_FTYPE_STT) {
SSttLvl *lvl = tsdbTFileSetGetSttLvl(fset, fobj->f->stt->level);
if (!lvl) {
code = tsdbSttLvlInit(fobj->f->stt->level, &lvl);
if (code) return code;
code = TARRAY2_SORT_INSERT(fset->lvlArr, lvl, tsdbSttLvlCmprFn);
if (code) return code;
}
code = TARRAY2_SORT_INSERT(lvl->fobjArr, fobj, tsdbTFileObjCmpr);
if (code) return code;
} else {
ASSERT(fset->farr[fobj->f->type] == NULL);
fset->farr[fobj->f->type] = fobj;
}
} else if (op->optype == TSDB_FOP_REMOVE) {
// delete a file
if (op->of.type == TSDB_FTYPE_STT) {
SSttLvl *lvl = tsdbTFileSetGetSttLvl(fset, op->of.stt->level);
ASSERT(lvl);
STFileObj tfobj = {.f[0] = {.cid = op->of.cid}};
STFileObj *tfobjp = &tfobj;
int32_t idx = TARRAY2_SEARCH_IDX(lvl->fobjArr, &tfobjp, tsdbTFileObjCmpr, TD_EQ);
ASSERT(idx >= 0);
TARRAY2_REMOVE(lvl->fobjArr, idx, tsdbSttLvlClearFObj);
if (TARRAY2_SIZE(lvl->fobjArr) == 0) {
// TODO: remove the stt level if no file exists anymore
// TARRAY2_REMOVE(&fset->lvlArr, lvl - fset->lvlArr.data, tsdbSttLvlClear);
}
} else {
ASSERT(tsdbIsSameTFile(&op->of, fset->farr[op->of.type]->f));
tsdbTFileObjUnref(fset->farr[op->of.type]);
fset->farr[op->of.type] = NULL;
}
} else {
if (op->nf.type == TSDB_FTYPE_STT) {
SSttLvl *lvl = tsdbTFileSetGetSttLvl(fset, op->of.stt->level);
ASSERT(lvl);
STFileObj tfobj = {.f[0] = {.cid = op->of.cid}}, *tfobjp = &tfobj;
STFileObj **fobjPtr = TARRAY2_SEARCH(lvl->fobjArr, &tfobjp, tsdbTFileObjCmpr, TD_EQ);
tfobjp = (fobjPtr ? *fobjPtr : NULL);
ASSERT(tfobjp);
tfobjp->f[0] = op->nf;
} else {
fset->farr[op->nf.type]->f[0] = op->nf;
}
}
return 0;
}
int32_t tsdbTFileSetApplyEdit(STsdb *pTsdb, const STFileSet *fset1, STFileSet *fset2) {
int32_t code = 0;
ASSERT(fset1->fid == fset2->fid);
for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (!fset1->farr[ftype] && !fset2->farr[ftype]) continue;
STFileObj *fobj1 = fset1->farr[ftype];
STFileObj *fobj2 = fset2->farr[ftype];
if (fobj1 && fobj2) {
if (tsdbIsSameTFile(fobj1->f, fobj2->f)) {
if (tsdbIsTFileChanged(fobj1->f, fobj2->f)) {
fobj2->f[0] = fobj1->f[0];
}
} else {
tsdbTFileObjRemove(fobj2);
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fset2->farr[ftype]);
if (code) return code;
}
} else if (fobj1) {
// create a new file
code = tsdbTFileObjInit(pTsdb, fobj1->f, &fset2->farr[ftype]);
if (code) return code;
} else {
// remove the file
tsdbTFileObjRemove(fobj2);
fset2->farr[ftype] = NULL;
}
}
// stt part
int32_t i1 = 0, i2 = 0;
while (i1 < TARRAY2_SIZE(fset1->lvlArr) || i2 < TARRAY2_SIZE(fset2->lvlArr)) {
SSttLvl *lvl1 = i1 < TARRAY2_SIZE(fset1->lvlArr) ? TARRAY2_GET(fset1->lvlArr, i1) : NULL;
SSttLvl *lvl2 = i2 < TARRAY2_SIZE(fset2->lvlArr) ? TARRAY2_GET(fset2->lvlArr, i2) : NULL;
if (lvl1 && lvl2) {
if (lvl1->level < lvl2->level) {
// add a new stt level
code = tsdbSttLvlInitEx(pTsdb, lvl1, &lvl2);
if (code) return code;
code = TARRAY2_SORT_INSERT(fset2->lvlArr, lvl2, tsdbSttLvlCmprFn);
if (code) return code;
i1++;
i2++;
} else if (lvl1->level > lvl2->level) {
// remove the stt level
TARRAY2_REMOVE(fset2->lvlArr, i2, tsdbSttLvlRemove);
} else {
// apply edit on stt level
code = tsdbSttLvlApplyEdit(pTsdb, lvl1, lvl2);
if (code) return code;
i1++;
i2++;
}
} else if (lvl1) {
// add a new stt level
code = tsdbSttLvlInitEx(pTsdb, lvl1, &lvl2);
if (code) return code;
code = TARRAY2_SORT_INSERT(fset2->lvlArr, lvl2, tsdbSttLvlCmprFn);
if (code) return code;
i1++;
i2++;
} else {
// remove the stt level
TARRAY2_REMOVE(fset2->lvlArr, i2, tsdbSttLvlRemove);
}
}
return 0;
}
int32_t tsdbTFileSetInit(int32_t fid, STFileSet **fset) {
fset[0] = taosMemoryCalloc(1, sizeof(STFileSet));
if (fset[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;
fset[0]->fid = fid;
TARRAY2_INIT(fset[0]->lvlArr);
return 0;
}
int32_t tsdbTFileSetInitDup(STsdb *pTsdb, const STFileSet *fset1, STFileSet **fset) {
int32_t code = tsdbTFileSetInit(fset1->fid, fset);
if (code) return code;
for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset1->farr[ftype] == NULL) continue;
code = tsdbTFileObjInit(pTsdb, fset1->farr[ftype]->f, &fset[0]->farr[ftype]);
if (code) {
tsdbTFileSetClear(fset);
return code;
}
}
const SSttLvl *lvl1;
TARRAY2_FOREACH(fset1->lvlArr, lvl1) {
SSttLvl *lvl;
code = tsdbSttLvlInitEx(pTsdb, lvl1, &lvl);
if (code) {
tsdbTFileSetClear(fset);
return code;
}
code = TARRAY2_APPEND(fset[0]->lvlArr, lvl);
if (code) return code;
}
return 0;
}
int32_t tsdbTFileSetInitRef(STsdb *pTsdb, const STFileSet *fset1, STFileSet **fset) {
int32_t code = tsdbTFileSetInit(fset1->fid, fset);
if (code) return code;
for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset1->farr[ftype] == NULL) continue;
tsdbTFileObjRef(fset1->farr[ftype]);
fset[0]->farr[ftype] = fset1->farr[ftype];
}
const SSttLvl *lvl1;
TARRAY2_FOREACH(fset1->lvlArr, lvl1) {
SSttLvl *lvl;
code = tsdbSttLvlInitRef(pTsdb, lvl1, &lvl);
if (code) {
tsdbTFileSetClear(fset);
return code;
}
code = TARRAY2_APPEND(fset[0]->lvlArr, lvl);
if (code) return code;
}
return 0;
}
int32_t tsdbTFileSetClear(STFileSet **fset) {
if (!fset[0]) return 0;
for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset[0]->farr[ftype] == NULL) continue;
tsdbTFileObjUnref(fset[0]->farr[ftype]);
}
TARRAY2_DESTROY(fset[0]->lvlArr, tsdbSttLvlClear);
taosMemoryFree(fset[0]);
fset[0] = NULL;
return 0;
}
int32_t tsdbTFileSetRemove(STFileSet **fset) {
for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset[0]->farr[ftype] == NULL) continue;
tsdbTFileObjRemove(fset[0]->farr[ftype]);
}
TARRAY2_DESTROY(fset[0]->lvlArr, tsdbSttLvlRemove);
taosMemoryFree(fset[0]);
fset[0] = NULL;
return 0;
}
SSttLvl *tsdbTFileSetGetSttLvl(STFileSet *fset, int32_t level) {
SSttLvl sttLvl = {.level = level};
SSttLvl *lvl = &sttLvl;
SSttLvl **lvlPtr = TARRAY2_SEARCH(fset->lvlArr, &lvl, tsdbSttLvlCmprFn, TD_EQ);
return lvlPtr ? lvlPtr[0] : NULL;
}
int32_t tsdbTFileSetCmprFn(const STFileSet **fset1, const STFileSet **fset2) {
if (fset1[0]->fid < fset2[0]->fid) return -1;
if (fset1[0]->fid > fset2[0]->fid) return 1;
return 0;
}
int64_t tsdbTFileSetMaxCid(const STFileSet *fset) {
int64_t maxCid = 0;
for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset->farr[ftype] == NULL) continue;
maxCid = TMAX(maxCid, fset->farr[ftype]->f->cid);
}
const SSttLvl *lvl;
const STFileObj *fobj;
TARRAY2_FOREACH(fset->lvlArr, lvl) {
TARRAY2_FOREACH(lvl->fobjArr, fobj) { maxCid = TMAX(maxCid, fobj->f->cid); }
}
return maxCid;
}
bool tsdbTFileSetIsEmpty(const STFileSet *fset) {
for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset->farr[ftype] != NULL) return false;
}
return TARRAY2_SIZE(fset->lvlArr) == 0;
}
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbFSet2.h 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "tsdbFile2.h"
#ifndef _TSDB_FILE_SET2_H
#define _TSDB_FILE_SET2_H
#ifdef __cplusplus
extern "C" {
#endif
typedef struct STFileSet STFileSet;
typedef struct STFileOp STFileOp;
typedef struct SSttLvl SSttLvl;
typedef TARRAY2(STFileObj *) TFileObjArray;
typedef TARRAY2(SSttLvl *) TSttLvlArray;
typedef TARRAY2(STFileOp) TFileOpArray;
typedef enum {
TSDB_FOP_NONE = 0,
TSDB_FOP_CREATE,
TSDB_FOP_REMOVE,
TSDB_FOP_MODIFY,
} tsdb_fop_t;
#define TFILE_SET(fid_) \
(STFileSet) { .fid = (fid_) }
// init/clear
int32_t tsdbTFileSetInit(int32_t fid, STFileSet **fset);
int32_t tsdbTFileSetInitDup(STsdb *pTsdb, const STFileSet *fset1, STFileSet **fset);
int32_t tsdbTFileSetInitRef(STsdb *pTsdb, const STFileSet *fset1, STFileSet **fset);
int32_t tsdbTFileSetClear(STFileSet **fset);
int32_t tsdbTFileSetRemove(STFileSet **fset);
// to/from json
int32_t tsdbTFileSetToJson(const STFileSet *fset, cJSON *json);
int32_t tsdbJsonToTFileSet(STsdb *pTsdb, const cJSON *json, STFileSet **fset);
// cmpr
int32_t tsdbTFileSetCmprFn(const STFileSet **fset1, const STFileSet **fset2);
// edit
int32_t tsdbTFileSetEdit(STsdb *pTsdb, STFileSet *fset, const STFileOp *op);
int32_t tsdbTFileSetApplyEdit(STsdb *pTsdb, const STFileSet *fset1, STFileSet *fset);
// max commit id
int64_t tsdbTFileSetMaxCid(const STFileSet *fset);
// get
SSttLvl *tsdbTFileSetGetSttLvl(STFileSet *fset, int32_t level);
// is empty
bool tsdbTFileSetIsEmpty(const STFileSet *fset);
struct STFileOp {
tsdb_fop_t optype;
int32_t fid;
STFile of; // old file state
STFile nf; // new file state
};
struct SSttLvl {
int32_t level;
TFileObjArray fobjArr[1];
};
struct STFileSet {
int32_t fid;
STFileObj *farr[TSDB_FTYPE_MAX]; // file array
TSttLvlArray lvlArr[1]; // level array
};
#ifdef __cplusplus
}
#endif
#endif /*_TSDB_FILE_SET2_H*/
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbFSetRW.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbFSetRW.h 0 → 100644
浏览文件 @ df57768a
/*
* 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/>.
*/
#include "tsdbDataFileRW.h"
#include "tsdbSttFileRW.h"
#ifndef _TSDB_FSET_RW_H
#define _TSDB_FSET_RW_H
#ifdef __cplusplus
extern "C" {
#endif
//
typedef struct SFSetWriter SFSetWriter;
typedef struct {
STsdb *tsdb;
bool toSttOnly;
int64_t compactVersion;
int32_t minRow;
int32_t maxRow;
int32_t szPage;
int8_t cmprAlg;
int32_t fid;
int64_t cid;
SDiskID did;
int32_t level;
struct {
bool exist;
STFile file;
} files[TSDB_FTYPE_MAX];
} SFSetWriterConfig;
int32_t tsdbFSetWriterOpen(SFSetWriterConfig *config, SFSetWriter **writer);
int32_t tsdbFSetWriterClose(SFSetWriter **writer, bool abort, TFileOpArray *fopArr);
int32_t tsdbFSetWriteRow(SFSetWriter *writer, SRowInfo *row);
int32_t tsdbFSetWriteTombRecord(SFSetWriter *writer, const STombRecord *tombRecord);
#ifdef __cplusplus
}
#endif
#endif /*_TSDB_FSET_RW_H*/
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbFile2.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbFile2.h 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbIter.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbIter.h 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbMerge.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbMerge.h 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbRead2.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbReadUtil.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbReadUtil.h 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbSttFileRW.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbSttFileRW.h 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbUpgrade.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbUpgrade.h 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbUtil2.c 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/dnode/vnode/src/tsdb/tsdbUtil2.h 0 → 100644
浏览文件 @ df57768a
此差异已折叠。
source/libs/catalog/inc/catalogInt.h
浏览文件 @ df57768a
......@@ -938,7 +938,7 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const
void* param);
int32_t ctgLaunchJob(SCtgJob* pJob);
int32_t ctgMakeAsyncRes(SCtgJob* pJob);
int32_t ctgLaunchSubTask(SCtgTask* pTask, CTG_TASK_TYPE type, ctgSubTaskCbFp fp, void* param);
int32_t ctgLaunchSubTask(SCtgTask** ppTask, CTG_TASK_TYPE type, ctgSubTaskCbFp fp, void* param);
int32_t ctgGetTbCfgCb(SCtgTask* pTask);
void ctgFreeHandle(SCatalog* pCatalog);
......
source/libs/catalog/src/ctgAsync.c
浏览文件 @ df57768a
此差异已折叠。
source/libs/executor/src/tfill.c
浏览文件 @ df57768a
此差异已折叠。
source/libs/executor/src/tsort.c
浏览文件 @ df57768a
此差异已折叠。
source/libs/monitor/src/monMain.c
浏览文件 @ df57768a
此差异已折叠。
source/libs/parser/src/parUtil.c
浏览文件 @ df57768a
此差异已折叠。
source/os/src/osSysinfo.c
浏览文件 @ df57768a
此差异已折叠。
source/util/src/tarray.c
浏览文件 @ df57768a
此差异已折叠。
source/util/src/trbtree.c
浏览文件 @ df57768a
此差异已折叠。
source/util/src/tutil.c
浏览文件 @ df57768a
此差异已折叠。
tests/script/tsim/parser/fill.sim
浏览文件 @ df57768a
此差异已折叠。
tests/system-test/2-query/interp.py
浏览文件 @ df57768a
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或