提交 4b538a15 编写于 作者: H Hongze Cheng

Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/tdb

......@@ -28,7 +28,7 @@ int32_t init_env() {
return -1;
}
TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 1");
TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2");
if (taos_errno(pRes) != 0) {
printf("error in create db, reason:%s\n", taos_errstr(pRes));
return -1;
......@@ -42,25 +42,33 @@ int32_t init_env() {
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int) tags(a int)");
pRes =
taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(10)) tags(t1 int)");
if (taos_errno(pRes) != 0) {
printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create table if not exists tu1 using st1 tags(1)");
pRes = taos_query(pConn, "create table if not exists ct0 using st1 tags(1000)");
if (taos_errno(pRes) != 0) {
printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create table if not exists tu2 using st1 tags(2)");
pRes = taos_query(pConn, "create table if not exists ct1 using st1 tags(2000)");
if (taos_errno(pRes) != 0) {
printf("failed to create child table tu2, reason:%s\n", taos_errstr(pRes));
return -1;
}
pRes = taos_query(pConn, "create table if not exists ct3 using st1 tags(3000)");
if (taos_errno(pRes) != 0) {
printf("failed to create child table tu3, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
return 0;
}
......@@ -82,12 +90,40 @@ int32_t create_topic() {
/*const char* sql = "select * from tu1";*/
/*pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));*/
pRes = taos_query(pConn, "create topic test_stb_topic_1 as select * from tu1");
pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1 from ct1");
if (taos_errno(pRes) != 0) {
printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
#if 0
pRes = taos_query(pConn, "insert into tu1 values(now, 1, 1.0, 'bi1')");
if (taos_errno(pRes) != 0) {
printf("failed to create topic test_stb_topic_1, reason:%s\n", taos_errstr(pRes));
printf("failed to insert, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "insert into tu1 values(now+1d, 1, 1.0, 'bi1')");
if (taos_errno(pRes) != 0) {
printf("failed to insert, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "insert into tu2 values(now, 2, 2.0, 'bi2')");
if (taos_errno(pRes) != 0) {
printf("failed to insert, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "insert into tu2 values(now+1d, 2, 2.0, 'bi2')");
if (taos_errno(pRes) != 0) {
printf("failed to insert, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
#endif
taos_close(pConn);
return 0;
}
......@@ -115,7 +151,7 @@ tmq_t* build_consumer() {
tmq_list_t* build_topic_list() {
tmq_list_t* topic_list = tmq_list_new();
tmq_list_append(topic_list, "test_stb_topic_1");
tmq_list_append(topic_list, "topic_ctb_column");
return topic_list;
}
......@@ -215,8 +251,8 @@ int main(int argc, char* argv[]) {
if (argc > 1) {
printf("env init\n");
code = init_env();
create_topic();
}
create_topic();
tmq_t* tmq = build_consumer();
tmq_list_t* topic_list = build_topic_list();
/*perf_loop(tmq, topic_list);*/
......
......@@ -20,7 +20,7 @@
#include "taos.h"
int32_t init_env() {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 7010);
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
if (pConn == NULL) {
return -1;
}
......@@ -65,7 +65,7 @@ int32_t init_env() {
int32_t create_stream() {
printf("create stream\n");
TAOS_RES* pRes;
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 7010);
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
if (pConn == NULL) {
return -1;
}
......
......@@ -37,6 +37,14 @@ enum {
TMQ_MSG_TYPE__EP_RSP,
};
enum {
STREAM_TRIGGER__AT_ONCE = 1,
STREAM_TRIGGER__WINDOW_CLOSE,
STREAM_TRIGGER__BY_COUNT,
STREAM_TRIGGER__BY_BATCH_COUNT,
STREAM_TRIGGER__BY_EVENT_TIME,
};
typedef struct {
uint32_t numOfTables;
SArray* pGroupList;
......@@ -54,13 +62,16 @@ typedef struct SColumnDataAgg {
} SColumnDataAgg;
typedef struct SDataBlockInfo {
STimeWindow window;
int32_t rows;
int32_t rowSize;
int16_t numOfCols;
int16_t hasVarCol;
union {int64_t uid; int64_t blockId;};
int64_t groupId; // no need to serialize
STimeWindow window;
int32_t rows;
int32_t rowSize;
int16_t numOfCols;
int16_t hasVarCol;
union {
int64_t uid;
int64_t blockId;
};
int64_t groupId; // no need to serialize
} SDataBlockInfo;
typedef struct SSDataBlock {
......@@ -93,7 +104,7 @@ void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock);
int32_t tEncodeDataBlocks(void** buf, const SArray* blocks);
void* tDecodeDataBlocks(const void* buf, SArray** blocks);
void colDataDestroy(SColumnInfoData* pColData) ;
void colDataDestroy(SColumnInfoData* pColData);
static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) {
// WARNING: do not use info.numOfCols,
......@@ -198,8 +209,8 @@ typedef struct SGroupbyExpr {
} SGroupbyExpr;
enum {
FUNC_PARAM_TYPE_VALUE = 0,
FUNC_PARAM_TYPE_COLUMN,
FUNC_PARAM_TYPE_VALUE = 0x1,
FUNC_PARAM_TYPE_COLUMN= 0x2,
};
typedef struct SFunctParam {
......@@ -239,7 +250,7 @@ typedef struct SSessionWindow {
SColumn col;
} SSessionWindow;
#define QUERY_ASC_FORWARD_STEP 1
#define QUERY_ASC_FORWARD_STEP 1
#define QUERY_DESC_FORWARD_STEP -1
#define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)
......
......@@ -524,6 +524,7 @@ typedef struct {
int8_t walLevel;
int8_t quorum;
int8_t cacheLastRow;
int8_t replications;
} SAlterDbReq;
int32_t tSerializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq);
......
......@@ -47,21 +47,21 @@ extern "C" {
#define TD_VTYPE_NONE 0x0U // none or unknown/undefined
#define TD_VTYPE_NULL 0x01U // null val
#define TD_VTYPE_NORM 0x02U // normal val: not none, not null
#define TD_VTYPE_MAX 0x03U //
#define TD_VTYPE_MAX 0x03U //
#define TD_VTYPE_NONE_BYTE 0x0U
#define TD_VTYPE_NULL_BYTE 0x55U
#define TD_VTYPE_NORM_BYTE 0xAAU
#define TD_ROWS_ALL_NORM 0x01U
#define TD_ROWS_ALL_NORM 0x01U
#define TD_ROWS_NULL_NORM 0x0U
#define TD_COL_ROWS_NORM(c) ((c)->bitmap == TD_ROWS_ALL_NORM) // all rows of SDataCol/SBlockCol is NORM
#define TD_COL_ROWS_NORM(c) ((c)->bitmap == TD_ROWS_ALL_NORM) // all rows of SDataCol/SBlockCol is NORM
#define TD_SET_COL_ROWS_BTIMAP(c, v) ((c)->bitmap = (v))
#define TD_SET_COL_ROWS_NORM(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_ALL_NORM)
#define TD_SET_COL_ROWS_MISC(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_NULL_NORM)
#define TD_SET_COL_ROWS_NORM(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_ALL_NORM)
#define TD_SET_COL_ROWS_MISC(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_NULL_NORM)
#define KvConvertRatio (0.9f)
#define KvConvertRatio (0.9f)
#define isSelectKVRow(klen, tlen) ((klen) < ((tlen)*KvConvertRatio))
#ifdef TD_SUPPORT_BITMAP
......@@ -98,7 +98,7 @@ typedef void *SRow;
typedef struct {
TDRowValT valType;
void * val;
void *val;
} SCellVal;
typedef struct {
......@@ -158,43 +158,43 @@ typedef struct {
int16_t nBitmaps;
int16_t nBoundBitmaps;
int32_t offset;
void * pBitmap;
void * pOffset;
void *pBitmap;
void *pOffset;
int32_t extendedRowSize;
} SRowBuilder;
#define TD_ROW_HEAD_LEN (sizeof(STSRow))
#define TD_ROW_HEAD_LEN (sizeof(STSRow))
#define TD_ROW_NCOLS_LEN (sizeof(col_id_t))
#define TD_ROW_INFO(r) ((r)->info)
#define TD_ROW_TYPE(r) ((r)->type)
#define TD_ROW_DELETE(r) ((r)->del)
#define TD_ROW_ENDIAN(r) ((r)->endian)
#define TD_ROW_SVER(r) ((r)->sver)
#define TD_ROW_NCOLS(r) ((r)->data) // only valid for SKvRow
#define TD_ROW_DATA(r) ((r)->data)
#define TD_ROW_LEN(r) ((r)->len)
#define TD_ROW_KEY(r) ((r)->ts)
#define TD_ROW_INFO(r) ((r)->info)
#define TD_ROW_TYPE(r) ((r)->type)
#define TD_ROW_DELETE(r) ((r)->del)
#define TD_ROW_ENDIAN(r) ((r)->endian)
#define TD_ROW_SVER(r) ((r)->sver)
#define TD_ROW_NCOLS(r) ((r)->data) // only valid for SKvRow
#define TD_ROW_DATA(r) ((r)->data)
#define TD_ROW_LEN(r) ((r)->len)
#define TD_ROW_KEY(r) ((r)->ts)
#define TD_ROW_KEY_ADDR(r) (r)
// N.B. If without STSchema, getExtendedRowSize() is used to get the rowMaxBytes and
// (int32_t)ceil((double)nCols/TD_VTYPE_PARTS) should be added if TD_SUPPORT_BITMAP defined.
#define TD_ROW_MAX_BYTES_FROM_SCHEMA(s) (schemaTLen(s) + TD_ROW_HEAD_LEN)
#define TD_ROW_SET_INFO(r, i) (TD_ROW_INFO(r) = (i))
#define TD_ROW_SET_TYPE(r, t) (TD_ROW_TYPE(r) = (t))
#define TD_ROW_SET_DELETE(r) (TD_ROW_DELETE(r) = 1)
#define TD_ROW_SET_SVER(r, v) (TD_ROW_SVER(r) = (v))
#define TD_ROW_SET_LEN(r, l) (TD_ROW_LEN(r) = (l))
#define TD_ROW_SET_INFO(r, i) (TD_ROW_INFO(r) = (i))
#define TD_ROW_SET_TYPE(r, t) (TD_ROW_TYPE(r) = (t))
#define TD_ROW_SET_DELETE(r) (TD_ROW_DELETE(r) = 1)
#define TD_ROW_SET_SVER(r, v) (TD_ROW_SVER(r) = (v))
#define TD_ROW_SET_LEN(r, l) (TD_ROW_LEN(r) = (l))
#define TD_ROW_SET_NCOLS(r, n) (*(col_id_t *)TD_ROW_NCOLS(r) = (n))
#define TD_ROW_IS_DELETED(r) (TD_ROW_DELETE(r) == 1)
#define TD_IS_TP_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_TP)
#define TD_IS_KV_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_KV)
#define TD_IS_TP_ROW_T(t) ((t) == TD_ROW_TP)
#define TD_IS_KV_ROW_T(t) ((t) == TD_ROW_KV)
#define TD_IS_TP_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_TP)
#define TD_IS_KV_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_KV)
#define TD_IS_TP_ROW_T(t) ((t) == TD_ROW_TP)
#define TD_IS_KV_ROW_T(t) ((t) == TD_ROW_KV)
#define TD_BOOL_STR(b) ((b) ? "true" : "false")
#define TD_BOOL_STR(b) ((b) ? "true" : "false")
#define isUtilizeKVRow(k, d) ((k) < ((d)*KVRatioConvert))
#define TD_ROW_COL_IDX(r) POINTER_SHIFT(TD_ROW_DATA(r), sizeof(col_id_t))
......@@ -275,7 +275,7 @@ static FORCE_INLINE int32_t tdSetBitmapValType(void *pBitmap, int16_t colIdx, TD
}
int16_t nBytes = colIdx / TD_VTYPE_PARTS;
int16_t nOffset = colIdx & TD_VTYPE_OPTR;
char * pDestByte = (char *)POINTER_SHIFT(pBitmap, nBytes);
char *pDestByte = (char *)POINTER_SHIFT(pBitmap, nBytes);
switch (nOffset) {
case 0:
*pDestByte = ((*pDestByte) & 0x3F) | (valType << 6);
......@@ -313,7 +313,7 @@ static FORCE_INLINE int32_t tdGetBitmapValType(void *pBitmap, int16_t colIdx, TD
}
int16_t nBytes = colIdx / TD_VTYPE_PARTS;
int16_t nOffset = colIdx & TD_VTYPE_OPTR;
char * pDestByte = (char *)POINTER_SHIFT(pBitmap, nBytes);
char *pDestByte = (char *)POINTER_SHIFT(pBitmap, nBytes);
switch (nOffset) {
case 0:
*pValType = (((*pDestByte) & 0xC0) >> 6);
......@@ -620,7 +620,7 @@ static FORCE_INLINE int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowVa
if (tdValIsNorm(valType, val, colType)) {
// ts key stored in STSRow.ts
SKvRowIdx *pColIdx = (SKvRowIdx *)POINTER_SHIFT(TD_ROW_COL_IDX(row), offset);
char * ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row));
char *ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row));
pColIdx->colId = colId;
pColIdx->offset = TD_ROW_LEN(row); // the offset include the TD_ROW_HEAD_LEN
......@@ -638,7 +638,7 @@ static FORCE_INLINE int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowVa
// NULL/None value
else {
SKvRowIdx *pColIdx = (SKvRowIdx *)POINTER_SHIFT(TD_ROW_COL_IDX(row), offset);
char * ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row));
char *ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row));
pColIdx->colId = colId;
pColIdx->offset = TD_ROW_LEN(row); // the offset include the TD_ROW_HEAD_LEN
const void *nullVal = getNullValue(colType);
......@@ -775,8 +775,8 @@ static FORCE_INLINE int32_t tdGetKvRowValOfCol(SCellVal *output, STSRow *pRow, v
typedef struct {
STSchema *pSchema;
STSRow * pRow;
void * pBitmap;
STSRow *pRow;
void *pBitmap;
uint32_t offset;
col_id_t maxColId;
col_id_t colIdx; // [PRIMARYKEY_TIMESTAMP_COL_ID, nSchemaCols], PRIMARYKEY_TIMESTAMP_COL_ID equals 1
......@@ -881,7 +881,7 @@ static FORCE_INLINE bool tdGetTpRowDataOfCol(STSRowIter *pIter, col_type_t colTy
// internal
static FORCE_INLINE bool tdGetKvRowValOfColEx(STSRowIter *pIter, col_id_t colId, col_type_t colType, col_id_t *nIdx,
SCellVal *pVal) {
STSRow * pRow = pIter->pRow;
STSRow *pRow = pIter->pRow;
SKvRowIdx *pKvIdx = NULL;
bool colFound = false;
col_id_t kvNCols = tdRowGetNCols(pRow);
......@@ -937,9 +937,8 @@ static FORCE_INLINE bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col
STColumn *pCol = NULL;
STSchema *pSchema = pIter->pSchema;
while (pIter->colIdx <= pSchema->numOfCols) {
pCol = &pSchema->columns[pIter->colIdx];
pCol = &pSchema->columns[pIter->colIdx]; // 1st column of schema is primary TS key
if (colId == pCol->colId) {
++pIter->colIdx;
break;
} else if (colId < pCol->colId) {
++pIter->colIdx;
......@@ -948,7 +947,8 @@ static FORCE_INLINE bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col
return false;
}
}
return tdGetTpRowDataOfCol(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal);
tdGetTpRowDataOfCol(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal);
++pIter->colIdx;
} else if (TD_IS_KV_ROW(pIter->pRow)) {
return tdGetKvRowValOfColEx(pIter, colId, colType, &pIter->kvIdx, pVal);
} else {
......@@ -1076,7 +1076,7 @@ typedef struct {
typedef struct {
STSchema *pSchema;
STSRow * pRow;
STSRow *pRow;
} STSRowReader;
typedef struct {
......
......@@ -60,8 +60,10 @@ int32_t parseNatualDuration(const char* token, int32_t tokenLen, int64_t* durati
int32_t taosParseTime(const char* timestr, int64_t* time, int32_t len, int32_t timePrec, int8_t dayligth);
void deltaToUtcInitOnce();
char getPrecisionUnit(int32_t precision);
int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision);
int64_t convertTimeFromPrecisionToUnit(int64_t time, int32_t fromPrecision, char toUnit);
void taosFormatUtcTime(char *buf, int32_t bufLen, int64_t time, int32_t precision);
......
......@@ -82,21 +82,21 @@
#define TK_SINGLE_STABLE 64
#define TK_STREAM_MODE 65
#define TK_RETENTIONS 66
#define TK_TABLE 67
#define TK_NK_LP 68
#define TK_NK_RP 69
#define TK_STABLE 70
#define TK_ADD 71
#define TK_COLUMN 72
#define TK_MODIFY 73
#define TK_RENAME 74
#define TK_TAG 75
#define TK_SET 76
#define TK_NK_EQ 77
#define TK_USING 78
#define TK_TAGS 79
#define TK_NK_DOT 80
#define TK_NK_COMMA 81
#define TK_NK_COMMA 67
#define TK_TABLE 68
#define TK_NK_LP 69
#define TK_NK_RP 70
#define TK_STABLE 71
#define TK_ADD 72
#define TK_COLUMN 73
#define TK_MODIFY 74
#define TK_RENAME 75
#define TK_TAG 76
#define TK_SET 77
#define TK_NK_EQ 78
#define TK_USING 79
#define TK_TAGS 80
#define TK_NK_DOT 81
#define TK_COMMENT 82
#define TK_BOOL 83
#define TK_TINYINT 84
......@@ -138,59 +138,62 @@
#define TK_INTERVAL 120
#define TK_TOPIC 121
#define TK_AS 122
#define TK_EXPLAIN 123
#define TK_ANALYZE 124
#define TK_VERBOSE 125
#define TK_NK_BOOL 126
#define TK_RATIO 127
#define TK_NULL 128
#define TK_NK_VARIABLE 129
#define TK_NK_UNDERLINE 130
#define TK_ROWTS 131
#define TK_TBNAME 132
#define TK_QSTARTTS 133
#define TK_QENDTS 134
#define TK_WSTARTTS 135
#define TK_WENDTS 136
#define TK_WDURATION 137
#define TK_BETWEEN 138
#define TK_IS 139
#define TK_NK_LT 140
#define TK_NK_GT 141
#define TK_NK_LE 142
#define TK_NK_GE 143
#define TK_NK_NE 144
#define TK_MATCH 145
#define TK_NMATCH 146
#define TK_IN 147
#define TK_JOIN 148
#define TK_INNER 149
#define TK_SELECT 150
#define TK_DISTINCT 151
#define TK_WHERE 152
#define TK_PARTITION 153
#define TK_BY 154
#define TK_SESSION 155
#define TK_STATE_WINDOW 156
#define TK_SLIDING 157
#define TK_FILL 158
#define TK_VALUE 159
#define TK_NONE 160
#define TK_PREV 161
#define TK_LINEAR 162
#define TK_NEXT 163
#define TK_GROUP 164
#define TK_HAVING 165
#define TK_ORDER 166
#define TK_SLIMIT 167
#define TK_SOFFSET 168
#define TK_LIMIT 169
#define TK_OFFSET 170
#define TK_ASC 171
#define TK_DESC 172
#define TK_NULLS 173
#define TK_FIRST 174
#define TK_LAST 175
#define TK_DESC 123
#define TK_DESCRIBE 124
#define TK_RESET 125
#define TK_QUERY 126
#define TK_EXPLAIN 127
#define TK_ANALYZE 128
#define TK_VERBOSE 129
#define TK_NK_BOOL 130
#define TK_RATIO 131
#define TK_NULL 132
#define TK_NK_VARIABLE 133
#define TK_NK_UNDERLINE 134
#define TK_ROWTS 135
#define TK_TBNAME 136
#define TK_QSTARTTS 137
#define TK_QENDTS 138
#define TK_WSTARTTS 139
#define TK_WENDTS 140
#define TK_WDURATION 141
#define TK_BETWEEN 142
#define TK_IS 143
#define TK_NK_LT 144
#define TK_NK_GT 145
#define TK_NK_LE 146
#define TK_NK_GE 147
#define TK_NK_NE 148
#define TK_MATCH 149
#define TK_NMATCH 150
#define TK_IN 151
#define TK_JOIN 152
#define TK_INNER 153
#define TK_SELECT 154
#define TK_DISTINCT 155
#define TK_WHERE 156
#define TK_PARTITION 157
#define TK_BY 158
#define TK_SESSION 159
#define TK_STATE_WINDOW 160
#define TK_SLIDING 161
#define TK_FILL 162
#define TK_VALUE 163
#define TK_NONE 164
#define TK_PREV 165
#define TK_LINEAR 166
#define TK_NEXT 167
#define TK_GROUP 168
#define TK_HAVING 169
#define TK_ORDER 170
#define TK_SLIMIT 171
#define TK_SOFFSET 172
#define TK_LIMIT 173
#define TK_OFFSET 174
#define TK_ASC 175
#define TK_NULLS 176
#define TK_FIRST 177
#define TK_LAST 178
#define TK_NK_SPACE 300
#define TK_NK_COMMENT 301
......
/*
* 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 "cmdnodes.h"
#include "tmsg.h"
#include "plannodes.h"
int32_t qExecCommand(SNode* pStmt, SRetrieveTableRsp** pRsp);
int32_t qExecStaticExplain(SQueryPlan *pDag, SRetrieveTableRsp **pRsp);
......@@ -20,8 +20,14 @@
extern "C" {
#endif
#include "query.h"
#include "querynodes.h"
#define DESCRIBE_RESULT_COLS 4
#define DESCRIBE_RESULT_FIELD_LEN (TSDB_COL_NAME_LEN - 1 + VARSTR_HEADER_SIZE)
#define DESCRIBE_RESULT_TYPE_LEN (20 + VARSTR_HEADER_SIZE)
#define DESCRIBE_RESULT_NOTE_LEN (8 + VARSTR_HEADER_SIZE)
typedef struct SDatabaseOptions {
ENodeType type;
int32_t numOfBlocks;
......@@ -32,7 +38,9 @@ typedef struct SDatabaseOptions {
int32_t fsyncPeriod;
int32_t maxRowsPerBlock;
int32_t minRowsPerBlock;
int32_t keep;
int32_t keep0;
int32_t keep1;
int32_t keep2;
int32_t precision;
int32_t quorum;
int32_t replica;
......@@ -70,7 +78,9 @@ typedef struct SAlterDatabaseStmt {
typedef struct STableOptions {
ENodeType type;
int32_t keep;
int32_t keep0;
int32_t keep1;
int32_t keep2;
int32_t ttl;
char comments[TSDB_STB_COMMENT_LEN];
SNodeList* pSma;
......@@ -247,6 +257,13 @@ typedef struct SAlterLocalStmt {
char value[TSDB_DNODE_VALUE_LEN];
} SAlterLocalStmt;
typedef struct SDescribeStmt {
ENodeType type;
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
STableMeta* pMeta;
} SDescribeStmt;
#ifdef __cplusplus
}
#endif
......
......@@ -101,6 +101,8 @@ typedef enum ENodeType {
QUERY_NODE_DROP_TOPIC_STMT,
QUERY_NODE_ALTER_LOCAL_STMT,
QUERY_NODE_EXPLAIN_STMT,
QUERY_NODE_DESCRIBE_STMT,
QUERY_NODE_RESET_QUERY_CACHE_STMT,
QUERY_NODE_SHOW_DATABASES_STMT,
QUERY_NODE_SHOW_TABLES_STMT,
QUERY_NODE_SHOW_STABLES_STMT,
......@@ -213,6 +215,8 @@ int32_t nodesStringToNode(const char* pStr, SNode** pNode);
int32_t nodesListToString(const SNodeList* pList, bool format, char** pStr, int32_t* pLen);
int32_t nodesStringToList(const char* pStr, SNodeList** pList);
int32_t nodesNodeToSQL(SNode *pNode, char *buf, int32_t bufSize, int32_t *len);
#ifdef __cplusplus
}
#endif
......
......@@ -253,7 +253,7 @@ typedef struct SIntervalPhysiNode {
int64_t sliding;
int8_t intervalUnit;
int8_t slidingUnit;
uint8_t precision;
uint8_t precision;
SFillNode* pFill;
} SIntervalPhysiNode;
......
......@@ -314,6 +314,7 @@ bool nodesIsTimeorderQuery(const SNode* pQuery);
bool nodesIsTimelineQuery(const SNode* pQuery);
void* nodesGetValueFromNode(SValueNode *pNode);
char* nodesGetStrValueFromNode(SValueNode *pNode);
#ifdef __cplusplus
}
......
......@@ -55,6 +55,7 @@ typedef struct SQuery {
SArray* pDbList;
SArray* pTableList;
bool showRewrite;
bool localCmd;
} SQuery;
int32_t qParseQuerySql(SParseContext* pCxt, SQuery** pQuery);
......
......@@ -53,6 +53,7 @@ typedef struct SIndexMeta {
} SIndexMeta;
/*
* ASSERT(sizeof(SCTableMeta) == 24)
* ASSERT(tableType == TSDB_CHILD_TABLE)
......@@ -235,6 +236,11 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
} \
} while (0)
#define QRY_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define QRY_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define QRY_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
#ifdef __cplusplus
}
#endif
......
......@@ -61,16 +61,16 @@ extern "C" {
} \
}
#define WAL_HEAD_VER 0
#define WAL_HEAD_VER 0
#define WAL_NOSUFFIX_LEN 20
#define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN + 1)
#define WAL_LOG_SUFFIX "log"
#define WAL_SUFFIX_AT (WAL_NOSUFFIX_LEN + 1)
#define WAL_LOG_SUFFIX "log"
#define WAL_INDEX_SUFFIX "idx"
#define WAL_REFRESH_MS 1000
#define WAL_MAX_SIZE (TSDB_MAX_WAL_SIZE + sizeof(SWalHead))
#define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12)
#define WAL_FILE_LEN (WAL_PATH_LEN + 32)
#define WAL_MAGIC 0xFAFBFCFDULL
#define WAL_REFRESH_MS 1000
#define WAL_MAX_SIZE (TSDB_MAX_WAL_SIZE + sizeof(SWalHead))
#define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12)
#define WAL_FILE_LEN (WAL_PATH_LEN + 32)
#define WAL_MAGIC 0xFAFBFCFDULL
#define WAL_CUR_FAILED 1
......@@ -150,14 +150,15 @@ typedef struct SWal {
} SWal; // WAL HANDLE
typedef struct SWalReadHandle {
SWal *pWal;
TdFilePtr pReadLogTFile;
TdFilePtr pReadIdxTFile;
int64_t curFileFirstVer;
int64_t curVersion;
int64_t capacity;
int64_t status; // if cursor valid
SWalHead *pHead;
SWal *pWal;
TdFilePtr pReadLogTFile;
TdFilePtr pReadIdxTFile;
int64_t curFileFirstVer;
int64_t curVersion;
int64_t capacity;
int64_t status; // if cursor valid
TdThreadMutex mutex;
SWalHead *pHead;
} SWalReadHandle;
#pragma pack(pop)
......@@ -191,6 +192,7 @@ int32_t walEndSnapshot(SWal *);
SWalReadHandle *walOpenReadHandle(SWal *);
void walCloseReadHandle(SWalReadHandle *);
int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver);
int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **ppHead);
// deprecated
#if 0
......
......@@ -82,6 +82,7 @@ extern "C" {
#include "osLz4.h"
#include "osMath.h"
#include "osMemory.h"
#include "osProc.h"
#include "osRand.h"
#include "osThread.h"
#include "osSemaphore.h"
......
......@@ -32,6 +32,7 @@ extern "C" {
void *taosMemoryMalloc(int32_t size);
void *taosMemoryCalloc(int32_t num, int32_t size);
void *taosMemoryRealloc(void *ptr, int32_t size);
void *taosMemoryStrDup(void *ptr);
void taosMemoryFree(const void *ptr);
int32_t taosMemorySize(void *ptr);
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_OS_PROC_H_
#define _TD_OS_PROC_H_
#ifdef __cplusplus
extern "C" {
#endif
int32_t taosNewProc(char **args);
void taosWaitProc(int32_t pid);
void taosKillProc(int32_t pid);
bool taosProcExist(int32_t pid);
void taosSetProcName(int32_t argc, char **argv, const char *name);
void taosSetProcPath(int32_t argc, char **argv);
#ifdef __cplusplus
}
#endif
#endif /*_TD_OS_PROC_H_*/
......@@ -29,7 +29,6 @@ typedef struct {
int32_t taosCreateShm(SShm *pShm, int32_t shmsize) ;
void taosDropShm(SShm *pShm);
int32_t taosAttachShm(SShm *pShm);
void taosDetachShm(SShm *pShm);
#ifdef __cplusplus
}
......
......@@ -49,6 +49,8 @@ void taosSetSignal(int32_t signum, FSignalHandler sigfp);
void taosIgnSignal(int32_t signum);
void taosDflSignal(int32_t signum);
void taosKillChildOnParentStopped();
#ifdef __cplusplus
}
#endif
......
......@@ -42,103 +42,103 @@ typedef pthread_barrierattr_t TdThreadBarrierAttr;
// If the error is in a third-party library, place this header file under the third-party library header file.
// When you want to use this feature, you should find or add the same function in the following section.
#ifndef ALLOW_FORBID_FUNC
// #define pthread_t PTHREAD_T_TYPE_TAOS_FORBID
// #define pthread_spinlock_t PTHREAD_SPINLOCK_T_TYPE_TAOS_FORBID
// #define pthread_mutex_t PTHREAD_MUTEX_T_TYPE_TAOS_FORBID
// #define pthread_mutexattr_t PTHREAD_MUTEXATTR_T_TYPE_TAOS_FORBID
// #define pthread_rwlock_t PTHREAD_RWLOCK_T_TYPE_TAOS_FORBID
// #define pthread_attr_t PTHREAD_ATTR_T_TYPE_TAOS_FORBID
// #define pthread_once_t PTHREAD_ONCE_T_TYPE_TAOS_FORBID
// #define pthread_rwlockattr_t PTHREAD_RWLOCKATTR_T_TYPE_TAOS_FORBID
// #define pthread_cond_t PTHREAD_COND_T_TYPE_TAOS_FORBID
// #define pthread_condattr_t PTHREAD_CONDATTR_T_TYPE_TAOS_FORBID
// #define pthread_key_t PTHREAD_KEY_T_TYPE_TAOS_FORBID
// #define pthread_barrier_t PTHREAD_BARRIER_T_TYPE_TAOS_FORBID
// #define pthread_barrierattr_t PTHREAD_BARRIERATTR_T_TYPE_TAOS_FORBID
// #define pthread_create PTHREAD_CREATE_FUNC_TAOS_FORBID
// #define pthread_attr_destroy PTHREAD_ATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_attr_getdetachstate PTHREAD_ATTR_GETDETACHSTATE_FUNC_TAOS_FORBID
// #define pthread_attr_getinheritsched PTHREAD_ATTR_GETINHERITSCHED_FUNC_TAOS_FORBID
// #define pthread_attr_getschedparam PTHREAD_ATTR_GETSCHEDPARAM_FUNC_TAOS_FORBID
// #define pthread_attr_getschedpolicy PTHREAD_ATTR_GETSCHEDPOLICY_FUNC_TAOS_FORBID
// #define pthread_attr_getscope PTHREAD_ATTR_GETSCOPE_FUNC_TAOS_FORBID
// #define pthread_attr_getstacksize PTHREAD_ATTR_GETSTACKSIZE_FUNC_TAOS_FORBID
// #define pthread_attr_init PTHREAD_ATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_attr_setdetachstate PTHREAD_ATTR_SETDETACHSTATE_FUNC_TAOS_FORBID
// #define pthread_attr_setinheritsched PTHREAD_ATTR_SETINHERITSCHED_FUNC_TAOS_FORBID
// #define pthread_attr_setschedparam PTHREAD_ATTR_SETSCHEDPARAM_FUNC_TAOS_FORBID
// #define pthread_attr_setschedpolicy PTHREAD_ATTR_SETSCHEDPOLICY_FUNC_TAOS_FORBID
// #define pthread_attr_setscope PTHREAD_ATTR_SETSCOPE_FUNC_TAOS_FORBID
// #define pthread_attr_setstacksize PTHREAD_ATTR_SETSTACKSIZE_FUNC_TAOS_FORBID
// #define pthread_barrier_destroy PTHREAD_BARRIER_DESTROY_FUNC_TAOS_FORBID
// #define pthread_barrier_init PTHREAD_BARRIER_INIT_FUNC_TAOS_FORBID
// #define pthread_barrier_wait PTHREAD_BARRIER_WAIT_FUNC_TAOS_FORBID
// #define pthread_barrierattr_destroy PTHREAD_BARRIERATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_barrierattr_getpshared PTHREAD_BARRIERATTR_GETPSHARED_FUNC_TAOS_FORBID
// #define pthread_barrierattr_init PTHREAD_BARRIERATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_barrierattr_setpshared PTHREAD_BARRIERATTR_SETPSHARED_FUNC_TAOS_FORBID
// #define pthread_cancel PTHREAD_CANCEL_FUNC_TAOS_FORBID
// #define pthread_cond_destroy PTHREAD_COND_DESTROY_FUNC_TAOS_FORBID
// #define pthread_cond_init PTHREAD_COND_INIT_FUNC_TAOS_FORBID
// #define pthread_cond_signal PTHREAD_COND_SIGNAL_FUNC_TAOS_FORBID
// #define pthread_cond_broadcast PTHREAD_COND_BROADCAST_FUNC_TAOS_FORBID
// #define pthread_cond_wait PTHREAD_COND_WAIT_FUNC_TAOS_FORBID
// #define pthread_cond_timedwait PTHREAD_COND_TIMEDWAIT_FUNC_TAOS_FORBID
// #define pthread_condattr_destroy PTHREAD_CONDATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_condattr_getpshared PTHREAD_CONDATTR_GETPSHARED_FUNC_TAOS_FORBID
// #define pthread_condattr_init PTHREAD_CONDATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_condattr_setpshared PTHREAD_CONDATTR_SETPSHARED_FUNC_TAOS_FORBID
// #define pthread_detach PTHREAD_DETACH_FUNC_TAOS_FORBID
// #define pthread_equal PTHREAD_EQUAL_FUNC_TAOS_FORBID
// #define pthread_exit PTHREAD_EXIT_FUNC_TAOS_FORBID
// #define pthread_getschedparam PTHREAD_GETSCHEDPARAM_FUNC_TAOS_FORBID
// #define pthread_getspecific PTHREAD_GETSPECIFIC_FUNC_TAOS_FORBID
// #define pthread_join PTHREAD_JOIN_FUNC_TAOS_FORBID
// #define pthread_key_create PTHREAD_KEY_CREATE_FUNC_TAOS_FORBID
// #define pthread_key_delete PTHREAD_KEY_DELETE_FUNC_TAOS_FORBID
// #define pthread_kill PTHREAD_KILL_FUNC_TAOS_FORBID
// #define pthread_mutex_consistent PTHREAD_MUTEX_CONSISTENT_FUNC_TAOS_FORBID
// #define pthread_mutex_destroy PTHREAD_MUTEX_DESTROY_FUNC_TAOS_FORBID
// #define pthread_mutex_init PTHREAD_MUTEX_INIT_FUNC_TAOS_FORBID
// #define pthread_mutex_lock PTHREAD_MUTEX_LOCK_FUNC_TAOS_FORBID
// #define pthread_mutex_timedlock PTHREAD_MUTEX_TIMEDLOCK_FUNC_TAOS_FORBID
// #define pthread_mutex_trylock PTHREAD_MUTEX_TRYLOCK_FUNC_TAOS_FORBID
// #define pthread_mutex_unlock PTHREAD_MUTEX_UNLOCK_FUNC_TAOS_FORBID
// #define pthread_mutexattr_destroy PTHREAD_MUTEXATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_mutexattr_getpshared PTHREAD_MUTEXATTR_GETPSHARED_FUNC_TAOS_FORBID
// #define pthread_mutexattr_getrobust PTHREAD_MUTEXATTR_GETROBUST_FUNC_TAOS_FORBID
// #define pthread_mutexattr_gettype PTHREAD_MUTEXATTR_GETTYPE_FUNC_TAOS_FORBID
// #define pthread_mutexattr_init PTHREAD_MUTEXATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_mutexattr_setpshared PTHREAD_MUTEXATTR_SETPSHARED_FUNC_TAOS_FORBID
// #define pthread_mutexattr_setrobust PTHREAD_MUTEXATTR_SETROBUST_FUNC_TAOS_FORBID
// #define pthread_mutexattr_settype PTHREAD_MUTEXATTR_SETTYPE_FUNC_TAOS_FORBID
// #define pthread_once PTHREAD_ONCE_FUNC_TAOS_FORBID
// #define pthread_rwlock_destroy PTHREAD_RWLOCK_DESTROY_FUNC_TAOS_FORBID
// #define pthread_rwlock_init PTHREAD_RWLOCK_INIT_FUNC_TAOS_FORBID
// #define pthread_rwlock_rdlock PTHREAD_RWLOCK_RDLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_timedrdlock PTHREAD_RWLOCK_TIMEDRDLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_timedwrlock PTHREAD_RWLOCK_TIMEDWRLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_tryrdlock PTHREAD_RWLOCK_TRYRDLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_trywrlock PTHREAD_RWLOCK_TRYWRLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_unlock PTHREAD_RWLOCK_UNLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_wrlock PTHREAD_RWLOCK_WRLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlockattr_destroy PTHREAD_RWLOCKATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_rwlockattr_getpshared PTHREAD_RWLOCKATTR_GETPSHARED_FUNC_TAOS_FORBID
// #define pthread_rwlockattr_init PTHREAD_RWLOCKATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_rwlockattr_setpshared PTHREAD_RWLOCKATTR_SETPSHARED_FUNC_TAOS_FORBID
// #define pthread_self PTHREAD_SELF_FUNC_TAOS_FORBID
// #define pthread_setcancelstate PTHREAD_SETCANCELSTATE_FUNC_TAOS_FORBID
// #define pthread_setcanceltype PTHREAD_SETCANCELTYPE_FUNC_TAOS_FORBID
// #define pthread_setschedparam PTHREAD_SETSCHEDPARAM_FUNC_TAOS_FORBID
// #define pthread_setspecific PTHREAD_SETSPECIFIC_FUNC_TAOS_FORBID
// #define pthread_spin_destroy PTHREAD_SPIN_DESTROY_FUNC_TAOS_FORBID
// #define pthread_spin_init PTHREAD_SPIN_INIT_FUNC_TAOS_FORBID
// #define pthread_spin_lock PTHREAD_SPIN_LOCK_FUNC_TAOS_FORBID
// #define pthread_spin_trylock PTHREAD_SPIN_TRYLOCK_FUNC_TAOS_FORBID
// #define pthread_spin_unlock PTHREAD_SPIN_UNLOCK_FUNC_TAOS_FORBID
// #define pthread_testcancel PTHREAD_TESTCANCEL_FUNC_TAOS_FORBID
// #define pthread_sigmask PTHREAD_SIGMASK_FUNC_TAOS_FORBID
// #define sigwait SIGWAIT_FUNC_TAOS_FORBID
#define pthread_t PTHREAD_T_TYPE_TAOS_FORBID
#define pthread_spinlock_t PTHREAD_SPINLOCK_T_TYPE_TAOS_FORBID
#define pthread_mutex_t PTHREAD_MUTEX_T_TYPE_TAOS_FORBID
#define pthread_mutexattr_t PTHREAD_MUTEXATTR_T_TYPE_TAOS_FORBID
#define pthread_rwlock_t PTHREAD_RWLOCK_T_TYPE_TAOS_FORBID
#define pthread_attr_t PTHREAD_ATTR_T_TYPE_TAOS_FORBID
#define pthread_once_t PTHREAD_ONCE_T_TYPE_TAOS_FORBID
#define pthread_rwlockattr_t PTHREAD_RWLOCKATTR_T_TYPE_TAOS_FORBID
#define pthread_cond_t PTHREAD_COND_T_TYPE_TAOS_FORBID
#define pthread_condattr_t PTHREAD_CONDATTR_T_TYPE_TAOS_FORBID
#define pthread_key_t PTHREAD_KEY_T_TYPE_TAOS_FORBID
#define pthread_barrier_t PTHREAD_BARRIER_T_TYPE_TAOS_FORBID
#define pthread_barrierattr_t PTHREAD_BARRIERATTR_T_TYPE_TAOS_FORBID
#define pthread_create PTHREAD_CREATE_FUNC_TAOS_FORBID
#define pthread_attr_destroy PTHREAD_ATTR_DESTROY_FUNC_TAOS_FORBID
#define pthread_attr_getdetachstate PTHREAD_ATTR_GETDETACHSTATE_FUNC_TAOS_FORBID
#define pthread_attr_getinheritsched PTHREAD_ATTR_GETINHERITSCHED_FUNC_TAOS_FORBID
#define pthread_attr_getschedparam PTHREAD_ATTR_GETSCHEDPARAM_FUNC_TAOS_FORBID
#define pthread_attr_getschedpolicy PTHREAD_ATTR_GETSCHEDPOLICY_FUNC_TAOS_FORBID
#define pthread_attr_getscope PTHREAD_ATTR_GETSCOPE_FUNC_TAOS_FORBID
#define pthread_attr_getstacksize PTHREAD_ATTR_GETSTACKSIZE_FUNC_TAOS_FORBID
#define pthread_attr_init PTHREAD_ATTR_INIT_FUNC_TAOS_FORBID
#define pthread_attr_setdetachstate PTHREAD_ATTR_SETDETACHSTATE_FUNC_TAOS_FORBID
#define pthread_attr_setinheritsched PTHREAD_ATTR_SETINHERITSCHED_FUNC_TAOS_FORBID
#define pthread_attr_setschedparam PTHREAD_ATTR_SETSCHEDPARAM_FUNC_TAOS_FORBID
#define pthread_attr_setschedpolicy PTHREAD_ATTR_SETSCHEDPOLICY_FUNC_TAOS_FORBID
#define pthread_attr_setscope PTHREAD_ATTR_SETSCOPE_FUNC_TAOS_FORBID
#define pthread_attr_setstacksize PTHREAD_ATTR_SETSTACKSIZE_FUNC_TAOS_FORBID
#define pthread_barrier_destroy PTHREAD_BARRIER_DESTROY_FUNC_TAOS_FORBID
#define pthread_barrier_init PTHREAD_BARRIER_INIT_FUNC_TAOS_FORBID
#define pthread_barrier_wait PTHREAD_BARRIER_WAIT_FUNC_TAOS_FORBID
#define pthread_barrierattr_destroy PTHREAD_BARRIERATTR_DESTROY_FUNC_TAOS_FORBID
#define pthread_barrierattr_getpshared PTHREAD_BARRIERATTR_GETPSHARED_FUNC_TAOS_FORBID
#define pthread_barrierattr_init PTHREAD_BARRIERATTR_INIT_FUNC_TAOS_FORBID
#define pthread_barrierattr_setpshared PTHREAD_BARRIERATTR_SETPSHARED_FUNC_TAOS_FORBID
#define pthread_cancel PTHREAD_CANCEL_FUNC_TAOS_FORBID
#define pthread_cond_destroy PTHREAD_COND_DESTROY_FUNC_TAOS_FORBID
#define pthread_cond_init PTHREAD_COND_INIT_FUNC_TAOS_FORBID
#define pthread_cond_signal PTHREAD_COND_SIGNAL_FUNC_TAOS_FORBID
#define pthread_cond_broadcast PTHREAD_COND_BROADCAST_FUNC_TAOS_FORBID
#define pthread_cond_wait PTHREAD_COND_WAIT_FUNC_TAOS_FORBID
#define pthread_cond_timedwait PTHREAD_COND_TIMEDWAIT_FUNC_TAOS_FORBID
#define pthread_condattr_destroy PTHREAD_CONDATTR_DESTROY_FUNC_TAOS_FORBID
#define pthread_condattr_getpshared PTHREAD_CONDATTR_GETPSHARED_FUNC_TAOS_FORBID
#define pthread_condattr_init PTHREAD_CONDATTR_INIT_FUNC_TAOS_FORBID
#define pthread_condattr_setpshared PTHREAD_CONDATTR_SETPSHARED_FUNC_TAOS_FORBID
#define pthread_detach PTHREAD_DETACH_FUNC_TAOS_FORBID
#define pthread_equal PTHREAD_EQUAL_FUNC_TAOS_FORBID
#define pthread_exit PTHREAD_EXIT_FUNC_TAOS_FORBID
#define pthread_getschedparam PTHREAD_GETSCHEDPARAM_FUNC_TAOS_FORBID
#define pthread_getspecific PTHREAD_GETSPECIFIC_FUNC_TAOS_FORBID
#define pthread_join PTHREAD_JOIN_FUNC_TAOS_FORBID
#define pthread_key_create PTHREAD_KEY_CREATE_FUNC_TAOS_FORBID
#define pthread_key_delete PTHREAD_KEY_DELETE_FUNC_TAOS_FORBID
#define pthread_kill PTHREAD_KILL_FUNC_TAOS_FORBID
#define pthread_mutex_consistent PTHREAD_MUTEX_CONSISTENT_FUNC_TAOS_FORBID
#define pthread_mutex_destroy PTHREAD_MUTEX_DESTROY_FUNC_TAOS_FORBID
#define pthread_mutex_init PTHREAD_MUTEX_INIT_FUNC_TAOS_FORBID
#define pthread_mutex_lock PTHREAD_MUTEX_LOCK_FUNC_TAOS_FORBID
#define pthread_mutex_timedlock PTHREAD_MUTEX_TIMEDLOCK_FUNC_TAOS_FORBID
#define pthread_mutex_trylock PTHREAD_MUTEX_TRYLOCK_FUNC_TAOS_FORBID
#define pthread_mutex_unlock PTHREAD_MUTEX_UNLOCK_FUNC_TAOS_FORBID
#define pthread_mutexattr_destroy PTHREAD_MUTEXATTR_DESTROY_FUNC_TAOS_FORBID
#define pthread_mutexattr_getpshared PTHREAD_MUTEXATTR_GETPSHARED_FUNC_TAOS_FORBID
#define pthread_mutexattr_getrobust PTHREAD_MUTEXATTR_GETROBUST_FUNC_TAOS_FORBID
#define pthread_mutexattr_gettype PTHREAD_MUTEXATTR_GETTYPE_FUNC_TAOS_FORBID
#define pthread_mutexattr_init PTHREAD_MUTEXATTR_INIT_FUNC_TAOS_FORBID
#define pthread_mutexattr_setpshared PTHREAD_MUTEXATTR_SETPSHARED_FUNC_TAOS_FORBID
#define pthread_mutexattr_setrobust PTHREAD_MUTEXATTR_SETROBUST_FUNC_TAOS_FORBID
#define pthread_mutexattr_settype PTHREAD_MUTEXATTR_SETTYPE_FUNC_TAOS_FORBID
#define pthread_once PTHREAD_ONCE_FUNC_TAOS_FORBID
#define pthread_rwlock_destroy PTHREAD_RWLOCK_DESTROY_FUNC_TAOS_FORBID
#define pthread_rwlock_init PTHREAD_RWLOCK_INIT_FUNC_TAOS_FORBID
#define pthread_rwlock_rdlock PTHREAD_RWLOCK_RDLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_timedrdlock PTHREAD_RWLOCK_TIMEDRDLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_timedwrlock PTHREAD_RWLOCK_TIMEDWRLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_tryrdlock PTHREAD_RWLOCK_TRYRDLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_trywrlock PTHREAD_RWLOCK_TRYWRLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_unlock PTHREAD_RWLOCK_UNLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_wrlock PTHREAD_RWLOCK_WRLOCK_FUNC_TAOS_FORBID
#define pthread_rwlockattr_destroy PTHREAD_RWLOCKATTR_DESTROY_FUNC_TAOS_FORBID
#define pthread_rwlockattr_getpshared PTHREAD_RWLOCKATTR_GETPSHARED_FUNC_TAOS_FORBID
#define pthread_rwlockattr_init PTHREAD_RWLOCKATTR_INIT_FUNC_TAOS_FORBID
#define pthread_rwlockattr_setpshared PTHREAD_RWLOCKATTR_SETPSHARED_FUNC_TAOS_FORBID
#define pthread_self PTHREAD_SELF_FUNC_TAOS_FORBID
#define pthread_setcancelstate PTHREAD_SETCANCELSTATE_FUNC_TAOS_FORBID
#define pthread_setcanceltype PTHREAD_SETCANCELTYPE_FUNC_TAOS_FORBID
#define pthread_setschedparam PTHREAD_SETSCHEDPARAM_FUNC_TAOS_FORBID
#define pthread_setspecific PTHREAD_SETSPECIFIC_FUNC_TAOS_FORBID
#define pthread_spin_destroy PTHREAD_SPIN_DESTROY_FUNC_TAOS_FORBID
#define pthread_spin_init PTHREAD_SPIN_INIT_FUNC_TAOS_FORBID
#define pthread_spin_lock PTHREAD_SPIN_LOCK_FUNC_TAOS_FORBID
#define pthread_spin_trylock PTHREAD_SPIN_TRYLOCK_FUNC_TAOS_FORBID
#define pthread_spin_unlock PTHREAD_SPIN_UNLOCK_FUNC_TAOS_FORBID
#define pthread_testcancel PTHREAD_TESTCANCEL_FUNC_TAOS_FORBID
#define pthread_sigmask PTHREAD_SIGMASK_FUNC_TAOS_FORBID
#define sigwait SIGWAIT_FUNC_TAOS_FORBID
#endif
int32_t taosThreadCreate(TdThread * tid, const TdThreadAttr * attr, void *(*start)(void *), void *arg);
......
......@@ -46,6 +46,14 @@ extern "C" {
#define MILLISECOND_PER_DAY (MILLISECOND_PER_HOUR * 24)
#define MILLISECOND_PER_WEEK (MILLISECOND_PER_DAY * 7)
#define NANOSECOND_PER_USEC (1000L)
#define NANOSECOND_PER_MSEC (1000000L)
#define NANOSECOND_PER_SEC (1000000000L)
#define NANOSECOND_PER_MINUTE (NANOSECOND_PER_SEC * 60)
#define NANOSECOND_PER_HOUR (NANOSECOND_PER_MINUTE * 60)
#define NANOSECOND_PER_DAY (NANOSECOND_PER_HOUR * 24)
#define NANOSECOND_PER_WEEK (NANOSECOND_PER_DAY * 7)
int32_t taosGetTimeOfDay(struct timeval *tv);
//@return timestamp in second
......
......@@ -78,6 +78,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_CFG_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x010C)
#define TSDB_CODE_INVALID_CFG TAOS_DEF_ERROR_CODE(0, 0x010D)
#define TSDB_CODE_OUT_OF_SHM_MEM TAOS_DEF_ERROR_CODE(0, 0x010E)
#define TSDB_CODE_INVALID_SHM_ID TAOS_DEF_ERROR_CODE(0, 0x010F)
#define TSDB_CODE_REF_NO_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0110)
#define TSDB_CODE_REF_FULL TAOS_DEF_ERROR_CODE(0, 0x0111)
#define TSDB_CODE_REF_ID_REMOVED TAOS_DEF_ERROR_CODE(0, 0x0112)
......@@ -478,6 +479,9 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_INVALID_ENDPOINT TAOS_DEF_ERROR_CODE(0, 0x2613)
#define TSDB_CODE_PAR_EXPRIE_STATEMENT TAOS_DEF_ERROR_CODE(0, 0x2614)
#define TSDB_CODE_PAR_INTERVAL_VALUE_TOO_SMALL TAOS_DEF_ERROR_CODE(0, 0x2615)
#define TSDB_CODE_PAR_DB_NOT_SPECIFIED TAOS_DEF_ERROR_CODE(0, 0x2616)
#define TSDB_CODE_PAR_INVALID_IDENTIFIER_NAME TAOS_DEF_ERROR_CODE(0, 0x2617)
#define TSDB_CODE_PAR_CORRESPONDING_STABLE_ERR TAOS_DEF_ERROR_CODE(0, 0x2618)
#ifdef __cplusplus
}
......
......@@ -224,6 +224,7 @@ typedef enum ELogicConditionType {
#define TSDB_APP_NAME_LEN TSDB_UNI_LEN
#define TSDB_STB_COMMENT_LEN 1024
/**
* In some scenarios uint16_t (0~65535) is used to store the row len.
* - Firstly, we use 65531(65535 - 4), as the SDataRow/SKVRow contains 4 bits header.
......@@ -303,13 +304,13 @@ typedef enum ELogicConditionType {
#define TSDB_MAX_TOTAL_BLOCKS 10000
#define TSDB_DEFAULT_TOTAL_BLOCKS 6
#define TSDB_MIN_DAYS_PER_FILE 1
#define TSDB_MAX_DAYS_PER_FILE 3650
#define TSDB_DEFAULT_DAYS_PER_FILE 10
#define TSDB_MIN_DAYS_PER_FILE (1 * 1440) // unit minute
#define TSDB_MAX_DAYS_PER_FILE (3650 * 1440)
#define TSDB_DEFAULT_DAYS_PER_FILE (10 * 1440)
#define TSDB_MIN_KEEP 1 // data in db to be reserved.
#define TSDB_MAX_KEEP 365000 // data in db to be reserved.
#define TSDB_DEFAULT_KEEP 3650 // ten years
#define TSDB_MIN_KEEP (1 * 1440) // data in db to be reserved. unit minute
#define TSDB_MAX_KEEP (365000 * 1440) // data in db to be reserved.
#define TSDB_DEFAULT_KEEP (3650 * 1440) // ten years
#define TSDB_MIN_MIN_ROW_FBLOCK 10
#define TSDB_MAX_MIN_ROW_FBLOCK 1000
......@@ -327,7 +328,7 @@ typedef enum ELogicConditionType {
#define TSDB_MAX_FSYNC_PERIOD 180000 // millisecond
#define TSDB_DEFAULT_FSYNC_PERIOD 3000 // three second
#define TSDB_MIN_WAL_LEVEL 0
#define TSDB_MIN_WAL_LEVEL 1
#define TSDB_MAX_WAL_LEVEL 2
#define TSDB_DEFAULT_WAL_LEVEL 1
......@@ -388,6 +389,7 @@ typedef enum ELogicConditionType {
#define TSDB_DEFAULT_EXPLAIN_RATIO 0.001
#define TSDB_EXPLAIN_RESULT_ROW_SIZE 1024
#define TSDB_EXPLAIN_RESULT_COLUMN_NAME "QUERY PLAN"
#define TSDB_MAX_JOIN_TABLE_NUM 10
#define TSDB_MAX_UNION_CLAUSE 5
......@@ -479,6 +481,9 @@ enum {
#define QND_VGID 1
#define VND_VGID 0
#define MAX_NUM_STR_SIZE 40
#ifdef __cplusplus
}
#endif
......
......@@ -32,29 +32,25 @@ typedef void *(*ProcConsumeFp)(void *pParent, void *pHead, int16_t headLen, void
ProcFuncType ftype);
typedef struct {
int32_t childQueueSize;
ProcConsumeFp childConsumeFp;
ProcMallocFp childMallocHeadFp;
ProcFreeFp childFreeHeadFp;
ProcMallocFp childMallocBodyFp;
ProcFreeFp childFreeBodyFp;
int32_t parentQueueSize;
ProcConsumeFp parentConsumeFp;
ProcMallocFp parentdMallocHeadFp;
ProcMallocFp parentMallocHeadFp;
ProcFreeFp parentFreeHeadFp;
ProcMallocFp parentMallocBodyFp;
ProcFreeFp parentFreeBodyFp;
bool testFlag;
SShm shm;
void *pParent;
const char *name;
bool isChild;
} SProcCfg;
SProcObj *taosProcInit(const SProcCfg *pCfg);
void taosProcCleanup(SProcObj *pProc);
int32_t taosProcRun(SProcObj *pProc);
void taosProcStop(SProcObj *pProc);
bool taosProcIsChild(SProcObj *pProc);
int32_t taosProcChildId(SProcObj *pProc);
int32_t taosProcPutToChildQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
ProcFuncType ftype);
int32_t taosProcPutToParentQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
......
......@@ -8,7 +8,7 @@ target_include_directories(
target_link_libraries(
taos
INTERFACE api
PRIVATE os util common transport nodes parser planner catalog scheduler function qcom
PRIVATE os util common transport nodes parser command planner catalog scheduler function qcom
)
if(${BUILD_TEST})
......
#include "clientInt.h"
#include "clientLog.h"
#include "command.h"
#include "scheduler.h"
#include "tdatablock.h"
#include "tdef.h"
......@@ -170,7 +171,21 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery) {
return code;
}
int32_t execLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
SRetrieveTableRsp* pRsp = NULL;
int32_t code = qExecCommand(pQuery->pRoot, &pRsp);
if (TSDB_CODE_SUCCESS == code && NULL != pRsp) {
code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp);
}
return code;
}
int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
// drop table if exists not_exists_table
if (NULL == pQuery->pCmdMsg) {
return TSDB_CODE_SUCCESS;
}
SCmdMsgInfo* pMsgInfo = pQuery->pCmdMsg;
pRequest->type = pMsgInfo->msgType;
pRequest->body.requestMsg = (SDataBuf){.pData = pMsgInfo->pMsg, .len = pMsgInfo->msgLen, .handle = NULL};
......@@ -259,7 +274,9 @@ SRequestObj* execQueryImpl(STscObj* pTscObj, const char* sql, int sqlLen) {
CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
CHECK_CODE_GOTO(parseSql(pRequest, false, &pQuery), _return);
if (pQuery->directRpc) {
if (pQuery->localCmd) {
CHECK_CODE_GOTO(execLocalCmd(pRequest, pQuery), _return);
} else if (pQuery->directRpc) {
CHECK_CODE_GOTO(execDdlQuery(pRequest, pQuery), _return);
} else {
CHECK_CODE_GOTO(getPlan(pRequest, pQuery, &pRequest->body.pDag, pNodeList), _return);
......@@ -464,9 +481,11 @@ static void destroySendMsgInfo(SMsgSendInfo* pMsgBody) {
taosMemoryFreeClear(pMsgBody->msgInfo.pData);
taosMemoryFreeClear(pMsgBody);
}
bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType) {
return msgType == TDMT_VND_QUERY_RSP || msgType == TDMT_VND_FETCH_RSP || msgType == TDMT_VND_RES_READY_RSP || msgType == TDMT_VND_QUERY_HEARTBEAT_RSP;
}
void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->ahandle;
assert(pMsg->ahandle != NULL);
......@@ -647,6 +666,11 @@ void* doFetchRow(SRequestObj* pRequest, bool setupOneRowPtr) {
}
}
if (pResultInfo->completed) {
pResultInfo->numOfRows = 0;
return NULL;
}
SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
int64_t transporterId = 0;
......
......@@ -255,6 +255,7 @@ void tmqClearUnhandleMsg(tmq_t* tmq) {
break;
}
msg = NULL;
taosReadAllQitems(tmq->mqueue, tmq->qall);
while (1) {
taosGetQitem(tmq->qall, (void**)&msg);
......@@ -787,7 +788,7 @@ void tmqShowMsg(tmq_message_t* tmq_message) {
static bool noPrintSchema;
char pBuf[128];
SMqPollRsp* pRsp = &tmq_message->msg;
int32_t colNum = pRsp->schema->nCols;
int32_t colNum = 2;
if (!noPrintSchema) {
printf("|");
for (int32_t i = 0; i < colNum; i++) {
......@@ -838,6 +839,7 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
int32_t msgEpoch = ((SMqRspHead*)pMsg->pData)->epoch;
int32_t tmqEpoch = atomic_load_32(&tmq->epoch);
if (msgEpoch < tmqEpoch) {
/*printf("discard rsp epoch %d, current epoch %d\n", msgEpoch, tmqEpoch);*/
tsem_post(&tmq->rspSem);
tscWarn("discard rsp epoch %d, current epoch %d", msgEpoch, tmqEpoch);
return 0;
......@@ -886,6 +888,9 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
goto WRITE_QUEUE_FAIL;
}
tscError("tmq recv poll: vg %d, req offset %ld, rsp offset %ld", pParam->pVg->vgId, pRsp->msg.reqOffset,
pRsp->msg.rspOffset);
pRsp->vg = pParam->pVg;
taosWriteQitem(tmq->mqueue, pRsp);
atomic_add_fetch_32(&tmq->readyRequest, 1);
......@@ -902,6 +907,7 @@ WRITE_QUEUE_FAIL:
bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) {
/*printf("call update ep %d\n", epoch);*/
/*printf("tmq update ep epoch %d to epoch %d\n", tmq->epoch, epoch);*/
bool set = false;
int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
char vgKey[TSDB_TOPIC_FNAME_LEN + 22];
......@@ -932,6 +938,7 @@ bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) {
for (int32_t k = 0; k < vgNumCur; k++) {
SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, k);
sprintf(vgKey, "%s:%d", topic.topicName, pVgCur->vgId);
/*printf("epoch %d vg %d build %s\n", epoch, pVgCur->vgId, vgKey);*/
taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(int64_t));
}
break;
......@@ -945,9 +952,12 @@ bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) {
sprintf(vgKey, "%s:%d", topic.topicName, pVgEp->vgId);
int64_t* pOffset = taosHashGet(pHash, vgKey, strlen(vgKey));
int64_t offset = pVgEp->offset;
/*printf("epoch %d vg %d offset og to %ld\n", epoch, pVgEp->vgId, offset);*/
if (pOffset != NULL) {
offset = *pOffset;
/*printf("epoch %d vg %d found %s\n", epoch, pVgEp->vgId, vgKey);*/
}
/*printf("epoch %d vg %d offset set to %ld\n", epoch, pVgEp->vgId, offset);*/
SMqClientVg clientVg = {
.pollCnt = 0,
.currentOffset = offset,
......@@ -1195,6 +1205,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);
int32_t vgStatus = atomic_val_compare_exchange_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE, TMQ_VG_STATUS__WAIT);
if (vgStatus != TMQ_VG_STATUS__IDLE) {
/*printf("skip vg %d\n", pVg->vgId);*/
continue;
}
SMqPollReq* pReq = tmqBuildConsumeReqImpl(tmq, blockingTime, pTopic, pVg);
......@@ -1238,6 +1249,8 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
int64_t transporterId = 0;
/*printf("send poll\n");*/
atomic_add_fetch_32(&tmq->waitingRequest, 1);
/*tscDebug("tmq send poll: vg %d, req offset %ld", pVg->vgId, pVg->currentOffset);*/
/*printf("send vg %d %ld\n", pVg->vgId, pVg->currentOffset);*/
asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, sendInfo);
pVg->pollCnt++;
tmq->pollCnt++;
......
......@@ -1629,6 +1629,7 @@ int32_t tSerializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
if (tEncodeI8(&encoder, pReq->walLevel) < 0) return -1;
if (tEncodeI8(&encoder, pReq->quorum) < 0) return -1;
if (tEncodeI8(&encoder, pReq->cacheLastRow) < 0) return -1;
if (tEncodeI8(&encoder, pReq->replications) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
......@@ -1650,6 +1651,7 @@ int32_t tDeserializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
if (tDecodeI8(&decoder, &pReq->walLevel) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->quorum) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->cacheLastRow) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->replications) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
......
......@@ -361,6 +361,18 @@ int32_t parseLocaltimeDst(char* timestr, int64_t* time, int32_t timePrec) {
return 0;
}
char getPrecisionUnit(int32_t precision) {
static char units[3] = {TIME_UNIT_MILLISECOND, TIME_UNIT_MICROSECOND, TIME_UNIT_NANOSECOND};
switch (precision) {
case TSDB_TIME_PRECISION_MILLI:
case TSDB_TIME_PRECISION_MICRO:
case TSDB_TIME_PRECISION_NANO:
return units[precision];
default:
return 0;
}
}
int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision) {
assert(fromPrecision == TSDB_TIME_PRECISION_MILLI || fromPrecision == TSDB_TIME_PRECISION_MICRO ||
fromPrecision == TSDB_TIME_PRECISION_NANO);
......@@ -370,6 +382,33 @@ int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrec
return (int64_t)((double)time * factors[fromPrecision][toPrecision]);
}
int64_t convertTimeFromPrecisionToUnit(int64_t time, int32_t fromPrecision, char toUnit) {
assert(fromPrecision == TSDB_TIME_PRECISION_MILLI || fromPrecision == TSDB_TIME_PRECISION_MICRO ||
fromPrecision == TSDB_TIME_PRECISION_NANO);
static double factors[3] = {1000000., 1000., 1.};
switch (toUnit) {
case 's':
return time * factors[fromPrecision] / NANOSECOND_PER_SEC;
case 'm':
return time * factors[fromPrecision] / NANOSECOND_PER_MINUTE;
case 'h':
return time * factors[fromPrecision] / NANOSECOND_PER_HOUR;
case 'd':
return time * factors[fromPrecision] / NANOSECOND_PER_DAY;
case 'w':
return time * factors[fromPrecision] / NANOSECOND_PER_WEEK;
case 'a':
return time * factors[fromPrecision] / NANOSECOND_PER_MSEC;
case 'u':
return time * factors[fromPrecision] / NANOSECOND_PER_USEC;
case 'b':
return time * factors[fromPrecision];
default: {
return -1;
}
}
}
static int32_t getDuration(int64_t val, char unit, int64_t* result, int32_t timePrecision) {
switch (unit) {
case 's':
......@@ -688,4 +727,4 @@ void taosFormatUtcTime(char* buf, int32_t bufLen, int64_t t, int32_t precision)
length += (int32_t)strftime(ts + length, 40 - length, "%z", ptm);
tstrncpy(buf, ts, bufLen);
}
\ No newline at end of file
}
......@@ -130,7 +130,7 @@ int32_t dmReadFile(SDnodeMgmt *pMgmt) {
}
code = 0;
dInfo("succcessed to read file %s", file);
dDebug("succcessed to read file %s", file);
dmPrintDnodes(pMgmt);
PRASE_DNODE_OVER:
......@@ -200,7 +200,7 @@ int32_t dmWriteFile(SDnodeMgmt *pMgmt) {
taosMemoryFree(content);
char realfile[PATH_MAX];
snprintf(realfile, sizeof(realfile), "%s%smnode.json", pMgmt->path, TD_DIRSEP);
snprintf(realfile, sizeof(realfile), "%s%sdnode.json", pMgmt->path, TD_DIRSEP);
if (taosRenameFile(file, realfile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
......
......@@ -112,6 +112,16 @@ int32_t dmInit(SMgmtWrapper *pWrapper) {
return -1;
}
if (dndInitServer(pDnode) != 0) {
dError("failed to init trans server since %s", terrstr());
return -1;
}
if (dndInitClient(pDnode) != 0) {
dError("failed to init trans client since %s", terrstr());
return -1;
}
pWrapper->pMgmt = pMgmt;
dInfo("dnode-mgmt is initialized");
return 0;
......@@ -122,6 +132,7 @@ void dmCleanup(SMgmtWrapper *pWrapper) {
if (pMgmt == NULL) return;
dInfo("dnode-mgmt start to clean up");
SDnode *pDnode = pMgmt->pDnode;
dmStopWorker(pMgmt);
taosWLockLatch(&pMgmt->latch);
......@@ -140,6 +151,9 @@ void dmCleanup(SMgmtWrapper *pWrapper) {
taosMemoryFree(pMgmt);
pWrapper->pMgmt = NULL;
dndCleanupServer(pDnode);
dndCleanupClient(pDnode);
dInfo("dnode-mgmt is cleaned up");
}
......
......@@ -53,7 +53,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
tSerializeSStatusReq(pHead, contLen, &req);
taosArrayDestroy(req.pVloads);
SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)9527};
SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)0x9527};
pMgmt->statusSent = 1;
dTrace("send req:%s to mnode, app:%p", TMSG_INFO(rpcMsg.msgType), rpcMsg.ahandle);
......
......@@ -30,18 +30,13 @@ static struct {
} global = {0};
static void dndStopDnode(int signum, void *info, void *ctx) {
dInfo("signal:%d is received", signum);
dInfo("system signal:%d received", signum);
SDnode *pDnode = atomic_val_compare_exchange_ptr(&global.pDnode, 0, global.pDnode);
if (pDnode != NULL) {
dndHandleEvent(pDnode, DND_EVENT_STOP);
}
}
static void dndHandleChild(int signum, void *info, void *ctx) {
dInfo("signal:%d is received", signum);
dndHandleEvent(global.pDnode, DND_EVENT_CHILD);
}
static void dndSetSignalHandle() {
taosSetSignal(SIGTERM, dndStopDnode);
taosSetSignal(SIGHUP, dndStopDnode);
......@@ -50,15 +45,10 @@ static void dndSetSignalHandle() {
taosSetSignal(SIGBREAK, dndStopDnode);
if (!tsMultiProcess) {
// Set the single process signal
} else if (global.ntype == DNODE) {
// Set the parent process signal
// When the child process exits, the parent process receives a signal
taosSetSignal(SIGCHLD, dndHandleChild);
} else if (global.ntype == DNODE || global.ntype == NODE_MAX) {
taosIgnSignal(SIGCHLD);
} else {
// Set child process signal
// When the parent process exits, the child process will receive the SIGKILL signal
prctl(PR_SET_PDEATHSIG, SIGKILL);
taosKillChildOnParentStopped();
}
}
......@@ -79,10 +69,14 @@ static int32_t dndParseArgs(int32_t argc, char const *argv[]) {
tstrncpy(global.apolloUrl, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-e") == 0) {
tstrncpy(global.envFile, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-k") == 0) {
global.generateGrant = true;
} else if (strcmp(argv[i], "-n") == 0) {
global.ntype = atoi(argv[++i]);
if (global.ntype <= DNODE || global.ntype > NODE_MAX) {
printf("'-n' range is [1-5], default is 0\n");
return -1;
}
} else if (strcmp(argv[i], "-k") == 0) {
global.generateGrant = true;
} else if (strcmp(argv[i], "-C") == 0) {
global.dumpConfig = true;
} else if (strcmp(argv[i], "-V") == 0) {
......@@ -138,24 +132,24 @@ static int32_t dndInitLog() {
return taosCreateLog(logName, 1, configDir, global.envFile, global.apolloUrl, global.pArgs, 0);
}
static void dndSetProcName(char **argv) {
if (global.ntype != DNODE) {
static void dndSetProcInfo(int32_t argc, char **argv) {
taosSetProcPath(argc, argv);
if (global.ntype != DNODE && global.ntype != NODE_MAX) {
const char *name = dndNodeProcStr(global.ntype);
prctl(PR_SET_NAME, name);
strcpy(argv[0], name);
taosSetProcName(argc, argv, name);
}
}
static int32_t dndRunDnode() {
if (dndInit() != 0) {
dError("failed to initialize environment since %s", terrstr());
dError("failed to init environment since %s", terrstr());
return -1;
}
SDnodeOpt option = dndGetOpt();
SDnode *pDnode = dndCreate(&option);
if (pDnode == NULL) {
dError("failed to to create dnode object since %s", terrstr());
dError("failed to to create dnode since %s", terrstr());
return -1;
} else {
global.pDnode = pDnode;
......@@ -212,6 +206,6 @@ int main(int argc, char const *argv[]) {
return 0;
}
dndSetProcName((char **)argv);
dndSetProcInfo(argc, (char **)argv);
return dndRunDnode();
}
......@@ -95,13 +95,14 @@ typedef struct SMgmtWrapper {
bool deployed;
bool required;
EProcType procType;
int32_t procId;
SProcObj *pProc;
SShm shm;
void *pMgmt;
SDnode *pDnode;
NodeMsgFp msgFps[TDMT_MAX];
int8_t msgVgIds[TDMT_MAX]; // Handle the case where the same message type is distributed to qnode or vnode
SMgmtFp fp;
int8_t msgVgIds[TDMT_MAX]; // Handle the case where the same message type is distributed to qnode or vnode
NodeMsgFp msgFps[TDMT_MAX];
} SMgmtWrapper;
typedef struct {
......@@ -128,26 +129,31 @@ typedef struct SDnode {
EDndStatus status;
EDndEvent event;
SStartupReq startup;
TdFilePtr runtimeFile;
TdFilePtr lockfile;
STransMgmt trans;
SMgmtWrapper wrappers[NODE_MAX];
} SDnode;
const char *dndNodeLogStr(ENodeType ntype);
const char *dndNodeProcStr(ENodeType ntype);
const char *dndEventStr(EDndEvent ev);
EDndStatus dndGetStatus(SDnode *pDnode);
void dndSetStatus(SDnode *pDnode, EDndStatus stat);
void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId);
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dndSendMonitorReport(SDnode *pDnode);
int32_t dndInitServer(SDnode *pDnode);
void dndCleanupServer(SDnode *pDnode);
int32_t dndInitClient(SDnode *pDnode);
void dndCleanupClient(SDnode *pDnode);
int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg);
int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pMsg);
void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp);
void dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper);
int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg);
int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed);
int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed);
......
......@@ -50,17 +50,13 @@ void dndClose(SDnode *pDnode);
void dndHandleEvent(SDnode *pDnode, EDndEvent event);
// dndTransport.c
int32_t dndInitServer(SDnode *pDnode);
void dndCleanupServer(SDnode *pDnode);
int32_t dndInitClient(SDnode *pDnode);
void dndCleanupClient(SDnode *pDnode);
int32_t dndInitMsgHandle(SDnode *pDnode);
void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp);
// dndFile.c
int32_t dndOpenRuntimeFile(SDnode *pDnode);
int32_t dndWriteRuntimeFile(SDnode *pDnode);
void dndCloseRuntimeFile(SDnode *pDnode);
TdFilePtr dndCheckRunning(const char *dataDir);
int32_t dndReadShmFile(SDnode *pDnode);
int32_t dndWriteShmFile(SDnode *pDnode);
#ifdef __cplusplus
}
......
/*
* 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/>.
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
#include "wal.h"
static int8_t once = DND_ENV_INIT;
int32_t dndInit() {
dDebug("start to init dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
terrno = TSDB_CODE_REPEAT_INIT;
dError("failed to init dnode env since %s", terrstr());
return -1;
}
taosIgnSIGPIPE();
taosBlockSIGPIPE();
taosResolveCRC();
SMonCfg monCfg = {0};
monCfg.maxLogs = tsMonitorMaxLogs;
monCfg.port = tsMonitorPort;
monCfg.server = tsMonitorFqdn;
monCfg.comp = tsMonitorComp;
if (monInit(&monCfg) != 0) {
dError("failed to init monitor since %s", terrstr());
return -1;
}
dInfo("dnode env is initialized");
return 0;
}
void dndCleanup() {
dDebug("start to cleanup dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
dError("dnode env is already cleaned up");
return;
}
monCleanup();
walCleanUp();
taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up");
}
void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId) {
pWrapper->msgFps[TMSG_INDEX(msgType)] = nodeMsgFp;
pWrapper->msgVgIds[TMSG_INDEX(msgType)] = vgId;
}
EDndStatus dndGetStatus(SDnode *pDnode) { return pDnode->status; }
void dndSetStatus(SDnode *pDnode, EDndStatus status) {
if (pDnode->status != status) {
dDebug("dnode status set from %s to %s", dndStatStr(pDnode->status), dndStatStr(status));
pDnode->status = status;
}
}
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
SStartupReq *pStartup = &pDnode->startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);
pStartup->finished = 0;
}
void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) {
memcpy(pStartup, &pDnode->startup, sizeof(SStartupReq));
pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING);
}
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("startup req is received");
SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq));
dndGetStartup(pDnode, pStartup);
dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished);
SRpcMsg rpcRsp = {
.handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq), .ahandle = pReq->ahandle};
rpcSendResponse(&rpcRsp);
}
......@@ -16,35 +16,30 @@
#define _DEFAULT_SOURCE
#include "dndInt.h"
static void dndResetLog(SMgmtWrapper *pMgmt) {
char logname[24] = {0};
snprintf(logname, sizeof(logname), "%slog", pMgmt->name);
dInfo("node:%s, reset log to %s in child process", pMgmt->name, logname);
taosCloseLog();
taosInitLog(logname, 1);
}
static bool dndRequireNode(SMgmtWrapper *pWrapper) {
bool required = false;
int32_t code =(*pWrapper->fp.requiredFp)(pWrapper, &required);
bool required = false;
int32_t code = (*pWrapper->fp.requiredFp)(pWrapper, &required);
if (!required) {
dDebug("node:%s, no need to start", pWrapper->name);
dDebug("node:%s, does not require startup", pWrapper->name);
} else {
dDebug("node:%s, need to start", pWrapper->name);
dDebug("node:%s, needs to be started", pWrapper->name);
}
return required;
}
int32_t dndOpenNode(SMgmtWrapper *pWrapper) {
int32_t code = (*pWrapper->fp.openFp)(pWrapper);
if (code != 0) {
if (taosMkDir(pWrapper->path) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("node:%s, failed to create dir:%s since %s", pWrapper->name, pWrapper->path, terrstr());
return -1;
}
if ((*pWrapper->fp.openFp)(pWrapper) != 0) {
dError("node:%s, failed to open since %s", pWrapper->name, terrstr());
return -1;
} else {
dDebug("node:%s, has been opened", pWrapper->name);
}
dDebug("node:%s, has been opened", pWrapper->name);
pWrapper->deployed = true;
return 0;
}
......@@ -70,54 +65,6 @@ void dndCloseNode(SMgmtWrapper *pWrapper) {
dDebug("node:%s, has been closed", pWrapper->name);
}
static int32_t dndRunInSingleProcess(SDnode *pDnode) {
dInfo("dnode run in single process mode");
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
pWrapper->required = dndRequireNode(pWrapper);
if (!pWrapper->required) continue;
SMsgCb msgCb = dndCreateMsgcb(pWrapper);
tmsgSetDefaultMsgCb(&msgCb);
if (taosMkDir(pWrapper->path) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to create dir:%s since %s", pWrapper->path, terrstr());
return -1;
}
dInfo("node:%s, will start in single process", pWrapper->name);
pWrapper->procType = PROC_SINGLE;
if (dndOpenNode(pWrapper) != 0) {
dError("node:%s, failed to start since %s", pWrapper->name, terrstr());
return -1;
}
}
dndSetStatus(pDnode, DND_STAT_RUNNING);
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (pWrapper->fp.startFp == NULL) continue;
if ((*pWrapper->fp.startFp)(pWrapper) != 0) {
dError("node:%s, failed to start since %s", pWrapper->name, terrstr());
return -1;
}
}
return 0;
}
static void dndClearNodesExecpt(SDnode *pDnode, ENodeType except) {
// dndCleanupServer(pDnode);
for (ENodeType n = 0; n < NODE_MAX; ++n) {
if (except == n) continue;
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
pWrapper->required = false;
}
}
static void dndConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen,
ProcFuncType ftype) {
SRpcMsg *pRpc = &pMsg->rpcMsg;
......@@ -163,116 +110,234 @@ static void dndConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
taosMemoryFree(pMsg);
}
static int32_t dndRunInMultiProcess(SDnode *pDnode) {
dInfo("dnode run in multi process mode");
static int32_t dndNewProc(SMgmtWrapper *pWrapper, ENodeType n) {
char tstr[8] = {0};
char *args[6] = {0};
snprintf(tstr, sizeof(tstr), "%d", n);
args[1] = "-c";
args[2] = configDir;
args[3] = "-n";
args[4] = tstr;
args[5] = NULL;
int32_t pid = taosNewProc(args);
if (pid <= 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("node:%s, failed to exec in new process since %s", pWrapper->name, terrstr());
return -1;
}
for (ENodeType n = 0; n < NODE_MAX; ++n) {
pWrapper->procId = pid;
dInfo("node:%s, continue running in new process:%d", pWrapper->name, pid);
return 0;
}
static SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper) {
SProcCfg cfg = {.childConsumeFp = (ProcConsumeFp)dndConsumeChildQueue,
.childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
.childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
.childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.parentConsumeFp = (ProcConsumeFp)dndConsumeParentQueue,
.parentMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
.parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
.parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.shm = pWrapper->shm,
.pParent = pWrapper,
.name = pWrapper->name};
return cfg;
}
static int32_t dndRunInSingleProcess(SDnode *pDnode) {
dInfo("dnode run in single process");
for (ENodeType n = DNODE; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
pWrapper->required = dndRequireNode(pWrapper);
if (!pWrapper->required) continue;
SMsgCb msgCb = dndCreateMsgcb(pWrapper);
tmsgSetDefaultMsgCb(&msgCb);
if (dndOpenNode(pWrapper) != 0) {
dError("node:%s, failed to start since %s", pWrapper->name, terrstr());
return -1;
}
}
dndSetStatus(pDnode, DND_STAT_RUNNING);
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (pWrapper->fp.startFp == NULL) continue;
if ((*pWrapper->fp.startFp)(pWrapper) != 0) {
dError("node:%s, failed to start since %s", pWrapper->name, terrstr());
return -1;
}
}
dInfo("TDengine initialized successfully");
dndReportStartup(pDnode, "TDengine", "initialized successfully");
while (1) {
if (pDnode->event == DND_EVENT_STOP) {
dInfo("dnode is about to stop");
break;
}
taosMsleep(100);
}
return 0;
}
if (taosMkDir(pWrapper->path) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to create dir:%s since %s", pWrapper->path, terrstr());
static int32_t dndRunInParentProcess(SDnode *pDnode) {
dInfo("dnode run in parent process");
SMgmtWrapper *pDWrapper = &pDnode->wrappers[DNODE];
if (dndOpenNode(pDWrapper) != 0) {
dError("node:%s, failed to start since %s", pDWrapper->name, terrstr());
return -1;
}
for (ENodeType n = DNODE + 1; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
pWrapper->required = dndRequireNode(pWrapper);
if (!pWrapper->required) continue;
int32_t shmsize = 1024 * 1024 * 2; // size will be a configuration item
if (taosCreateShm(&pWrapper->shm, shmsize) != 0) {
terrno = TAOS_SYSTEM_ERROR(terrno);
dError("node:%s, failed to create shm size:%d since %s", pWrapper->name, shmsize, terrstr());
return -1;
}
dInfo("node:%s, shm:%d is created, size:%d", pWrapper->name, pWrapper->shm.id, shmsize);
SProcCfg cfg = dndGenProcCfg(pWrapper);
cfg.isChild = false;
pWrapper->procType = PROC_PARENT;
pWrapper->pProc = taosProcInit(&cfg);
if (pWrapper->pProc == NULL) {
dError("node:%s, failed to create proc since %s", pWrapper->name, terrstr());
return -1;
}
}
if (dndWriteShmFile(pDnode) != 0) {
dError("failed to write runtime file since %s", terrstr());
return -1;
}
for (ENodeType n = DNODE + 1; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (n == DNODE) {
dInfo("node:%s, will start in parent process", pWrapper->name);
pWrapper->procType = PROC_SINGLE;
if (dndOpenNode(pWrapper) != 0) {
dError("node:%s, failed to start since %s", pWrapper->name, terrstr());
if (pDnode->ntype == NODE_MAX) {
dInfo("node:%s, should be started manually", pWrapper->name);
} else {
if (dndNewProc(pWrapper, n) != 0) {
return -1;
}
continue;
}
SProcCfg cfg = {.childQueueSize = 1024 * 1024 * 2, // size will be a configuration item
.childConsumeFp = (ProcConsumeFp)dndConsumeChildQueue,
.childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
.childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
.childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.parentQueueSize = 1024 * 1024 * 2, // size will be a configuration item
.parentConsumeFp = (ProcConsumeFp)dndConsumeParentQueue,
.parentdMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
.parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
.parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.pParent = pWrapper,
.name = pWrapper->name};
SProcObj *pProc = taosProcInit(&cfg);
if (pProc == NULL) {
dError("node:%s, failed to fork since %s", pWrapper->name, terrstr());
if (taosProcRun(pWrapper->pProc) != 0) {
dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr());
return -1;
}
}
pWrapper->pProc = pProc;
dndSetStatus(pDnode, DND_STAT_RUNNING);
if (taosProcIsChild(pProc)) {
dInfo("node:%s, will start in child process", pWrapper->name);
pWrapper->procType = PROC_CHILD;
dndResetLog(pWrapper);
if ((*pDWrapper->fp.startFp)(pDWrapper) != 0) {
dError("node:%s, failed to start since %s", pDWrapper->name, terrstr());
return -1;
}
dInfo("node:%s, clean up resources inherited from parent", pWrapper->name);
dndClearNodesExecpt(pDnode, n);
dInfo("TDengine initialized successfully");
dndReportStartup(pDnode, "TDengine", "initialized successfully");
dInfo("node:%s, will be initialized in child process", pWrapper->name);
if (dndOpenNode(pWrapper) != 0) {
dInfo("node:%s, failed to init in child process since %s", pWrapper->name, terrstr());
return -1;
while (1) {
if (pDnode->event == DND_EVENT_STOP) {
dInfo("dnode is about to stop");
for (ENodeType n = DNODE + 1; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (pDnode->ntype == NODE_MAX) continue;
if (pWrapper->procId > 0 && taosProcExist(pWrapper->procId)) {
dInfo("node:%s, send kill signal to the child process:%d", pWrapper->name, pWrapper->procId);
taosKillProc(pWrapper->procId);
}
}
if (taosProcRun(pProc) != 0) {
dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr());
return -1;
for (ENodeType n = DNODE + 1; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (pDnode->ntype == NODE_MAX) continue;
if (pWrapper->procId > 0 && taosProcExist(pWrapper->procId)) {
dInfo("node:%s, wait for child process:%d to stop", pWrapper->name, pWrapper->procId);
taosWaitProc(pWrapper->procId);
dInfo("node:%s, child process:%d is stopped", pWrapper->name, pWrapper->procId);
}
}
break;
} else {
dInfo("node:%s, will not start in parent process, child pid:%d", pWrapper->name, taosProcChildId(pProc));
pWrapper->procType = PROC_PARENT;
if (taosProcRun(pProc) != 0) {
dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr());
return -1;
for (ENodeType n = DNODE + 1; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (pDnode->ntype == NODE_MAX) continue;
if (pWrapper->procId <= 0 || !taosProcExist(pWrapper->procId)) {
dInfo("node:%s, process:%d is killed and needs to be restarted", pWrapper->name, pWrapper->procId);
dndNewProc(pWrapper, n);
}
}
}
taosMsleep(100);
}
dndSetStatus(pDnode, DND_STAT_RUNNING);
return 0;
}
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (pWrapper->fp.startFp == NULL) continue;
if (pWrapper->procType == PROC_PARENT && n != DNODE) continue;
if (pWrapper->procType == PROC_CHILD && n == DNODE) continue;
static int32_t dndRunInChildProcess(SDnode *pDnode) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
dInfo("%s run in child process", pWrapper->name);
pWrapper->required = dndRequireNode(pWrapper);
if (!pWrapper->required) {
dError("%s does not require startup", pWrapper->name);
return -1;
}
SMsgCb msgCb = dndCreateMsgcb(pWrapper);
tmsgSetDefaultMsgCb(&msgCb);
pWrapper->procType = PROC_CHILD;
if (dndOpenNode(pWrapper) != 0) {
dError("node:%s, failed to start since %s", pWrapper->name, terrstr());
return -1;
}
SProcCfg cfg = dndGenProcCfg(pWrapper);
cfg.isChild = true;
pWrapper->pProc = taosProcInit(&cfg);
if (pWrapper->pProc == NULL) {
dError("node:%s, failed to create proc since %s", pWrapper->name, terrstr());
return -1;
}
if (pWrapper->fp.startFp != NULL) {
if ((*pWrapper->fp.startFp)(pWrapper) != 0) {
dError("node:%s, failed to start since %s", pWrapper->name, terrstr());
return -1;
}
}
return 0;
}
int32_t dndRun(SDnode *pDnode) {
if (!tsMultiProcess) {
if (dndRunInSingleProcess(pDnode) != 0) {
dError("failed to run dnode in single process mode since %s", terrstr());
return -1;
}
} else {
if (dndRunInMultiProcess(pDnode) != 0) {
dError("failed to run dnode in multi process mode since %s", terrstr());
return -1;
}
if (taosProcRun(pWrapper->pProc) != 0) {
dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr());
return -1;
}
dInfo("TDengine initialized successfully");
dndReportStartup(pDnode, "TDengine", "initialized successfully");
while (1) {
if (pDnode->event == DND_EVENT_STOP) {
dInfo("dnode is about to stop");
......@@ -280,6 +345,16 @@ int32_t dndRun(SDnode *pDnode) {
}
taosMsleep(100);
}
}
int32_t dndRun(SDnode *pDnode) {
if (!tsMultiProcess) {
return dndRunInSingleProcess(pDnode);
} else if (pDnode->ntype == DNODE || pDnode->ntype == NODE_MAX) {
return dndRunInParentProcess(pDnode);
} else {
return dndRunInChildProcess(pDnode);
}
return 0;
}
......@@ -117,7 +117,30 @@ _OVER:
return code;
}
int32_t dndOpenRuntimeFile(SDnode *pDnode) {
TdFilePtr dndCheckRunning(const char *dataDir) {
char filepath[PATH_MAX] = {0};
snprintf(filepath, sizeof(filepath), "%s%s.running", dataDir, TD_DIRSEP);
TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", filepath, terrstr());
return NULL;
}
int32_t ret = taosLockFile(pFile);
if (ret != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", filepath, terrstr());
taosCloseFile(&pFile);
return NULL;
}
dDebug("file:%s is locked", filepath);
return pFile;
}
int32_t dndReadShmFile(SDnode *pDnode) {
int32_t code = -1;
char itemName[24] = {0};
char content[MAXLEN + 1] = {0};
......@@ -125,17 +148,11 @@ int32_t dndOpenRuntimeFile(SDnode *pDnode) {
cJSON *root = NULL;
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s.running", pDnode->dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
snprintf(file, sizeof(file), "%s%s.shmfile", pDnode->dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_READ);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to open file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosLockFile(pFile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", file, terrstr());
dDebug("file %s not exist", file);
code = 0;
goto _OVER;
}
......@@ -162,10 +179,10 @@ int32_t dndOpenRuntimeFile(SDnode *pDnode) {
}
}
if (tsMultiProcess || pDnode->ntype == DNODE) {
if (!tsMultiProcess || pDnode->ntype == DNODE || pDnode->ntype == NODE_MAX) {
for (ENodeType ntype = DNODE; ntype < NODE_MAX; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
if (pWrapper->shm.id > 0) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
if (pWrapper->shm.id >= 0) {
dDebug("shmid:%d, is closed, size:%d", pWrapper->shm.id, pWrapper->shm.size);
taosDropShm(&pWrapper->shm);
}
......@@ -174,27 +191,23 @@ int32_t dndOpenRuntimeFile(SDnode *pDnode) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
if (taosAttachShm(&pWrapper->shm) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("shmid:%d, failed to attach since %s", pWrapper->shm.id, terrstr());
dError("shmid:%d, failed to attach shm since %s", pWrapper->shm.id, terrstr());
goto _OVER;
}
dDebug("shmid:%d, is attached, size:%d", pWrapper->shm.id, pWrapper->shm.size);
dInfo("node:%s, shmid:%d is attached, size:%d", pWrapper->name, pWrapper->shm.id, pWrapper->shm.size);
}
dDebug("successed to open %s", file);
dDebug("successed to load %s", file);
code = 0;
_OVER:
if (root != NULL) cJSON_Delete(root);
if (code != 0) {
if (pFile != NULL) taosCloseFile(&pFile);
} else {
pDnode->runtimeFile = pFile;
}
if (pFile != NULL) taosCloseFile(&pFile);
return code;
}
int32_t dndWriteRuntimeFile(SDnode *pDnode) {
int32_t dndWriteShmFile(SDnode *pDnode) {
int32_t code = -1;
int32_t len = 0;
char content[MAXLEN + 1] = {0};
......@@ -202,8 +215,8 @@ int32_t dndWriteRuntimeFile(SDnode *pDnode) {
char realfile[PATH_MAX] = {0};
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s.running.bak", pDnode->dataDir, TD_DIRSEP);
snprintf(realfile, sizeof(realfile), "%s%s.running", pDnode->dataDir, TD_DIRSEP);
snprintf(file, sizeof(file), "%s%s.shmfile.bak", pDnode->dataDir, TD_DIRSEP);
snprintf(realfile, sizeof(realfile), "%s%s.shmfile", pDnode->dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
......@@ -214,12 +227,12 @@ int32_t dndWriteRuntimeFile(SDnode *pDnode) {
len += snprintf(content + len, MAXLEN - len, "{\n");
for (ENodeType ntype = DNODE + 1; ntype < NODE_MAX; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
len += snprintf(content + len, MAXLEN - len, " \"%s_shmid\": %d,\n", dndNodeProcStr(ntype), pWrapper->shm.id);
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
len += snprintf(content + len, MAXLEN - len, " \"%s_shmid\":%d,\n", dndNodeProcStr(ntype), pWrapper->shm.id);
if (ntype == NODE_MAX - 1) {
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\": %d\n", dndNodeProcStr(ntype), pWrapper->shm.size);
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\":%d\n", dndNodeProcStr(ntype), pWrapper->shm.size);
} else {
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\": %d,\n", dndNodeProcStr(ntype), pWrapper->shm.size);
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\":%d,\n", dndNodeProcStr(ntype), pWrapper->shm.size);
}
}
len += snprintf(content + len, MAXLEN - len, "}\n");
......@@ -244,7 +257,7 @@ int32_t dndWriteRuntimeFile(SDnode *pDnode) {
return -1;
}
dDebug("successed to write %s", realfile);
dInfo("successed to write %s", realfile);
code = 0;
_OVER:
......@@ -254,11 +267,3 @@ _OVER:
return code;
}
void dndCloseRuntimeFile(SDnode *pDnode) {
if (pDnode->runtimeFile) {
taosUnLockFile(pDnode->runtimeFile);
taosCloseFile(&pDnode->runtimeFile);
pDnode->runtimeFile = NULL;
}
}
\ No newline at end of file
......@@ -15,82 +15,186 @@
#define _DEFAULT_SOURCE
#include "dndInt.h"
#include "wal.h"
static int8_t once = DND_ENV_INIT;
static int32_t dndInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
pDnode->numOfSupportVnodes = pOption->numOfSupportVnodes;
pDnode->serverPort = pOption->serverPort;
pDnode->dataDir = strdup(pOption->dataDir);
pDnode->localEp = strdup(pOption->localEp);
pDnode->localFqdn = strdup(pOption->localFqdn);
pDnode->firstEp = strdup(pOption->firstEp);
pDnode->secondEp = strdup(pOption->secondEp);
pDnode->disks = pOption->disks;
pDnode->numOfDisks = pOption->numOfDisks;
pDnode->ntype = pOption->ntype;
pDnode->rebootTime = taosGetTimestampMs();
int32_t dndInit() {
dDebug("start to init dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
terrno = TSDB_CODE_REPEAT_INIT;
dError("failed to init dnode env since %s", terrstr());
if (pDnode->dataDir == NULL || pDnode->localEp == NULL || pDnode->localFqdn == NULL || pDnode->firstEp == NULL ||
pDnode->secondEp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
taosIgnSIGPIPE();
taosBlockSIGPIPE();
taosResolveCRC();
SMonCfg monCfg = {0};
monCfg.maxLogs = tsMonitorMaxLogs;
monCfg.port = tsMonitorPort;
monCfg.server = tsMonitorFqdn;
monCfg.comp = tsMonitorComp;
if (monInit(&monCfg) != 0) {
dError("failed to init monitor since %s", terrstr());
return -1;
if (!tsMultiProcess || pDnode->ntype == DNODE || pDnode->ntype == NODE_MAX) {
pDnode->lockfile = dndCheckRunning(pDnode->dataDir);
if (pDnode->lockfile == NULL) {
return -1;
}
}
dDebug("dnode env is initialized");
return 0;
}
void dndCleanup() {
dDebug("start to cleanup dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
dError("dnode env is already cleaned up");
return;
static void dndClearVars(SDnode *pDnode) {
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pMgmt = &pDnode->wrappers[n];
taosMemoryFreeClear(pMgmt->path);
}
monCleanup();
walCleanUp();
taosStopCacheRefreshWorker();
dDebug("dnode env is cleaned up");
if (pDnode->lockfile != NULL) {
taosUnLockFile(pDnode->lockfile);
taosCloseFile(&pDnode->lockfile);
pDnode->lockfile = NULL;
}
taosMemoryFreeClear(pDnode->localEp);
taosMemoryFreeClear(pDnode->localFqdn);
taosMemoryFreeClear(pDnode->firstEp);
taosMemoryFreeClear(pDnode->secondEp);
taosMemoryFreeClear(pDnode->dataDir);
taosMemoryFree(pDnode);
dDebug("dnode memory is cleared, data:%p", pDnode);
}
void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId) {
pWrapper->msgFps[TMSG_INDEX(msgType)] = nodeMsgFp;
pWrapper->msgVgIds[TMSG_INDEX(msgType)] = vgId;
SDnode *dndCreate(const SDnodeOpt *pOption) {
dDebug("start to create dnode object");
int32_t code = -1;
char path[PATH_MAX] = {0};
SDnode *pDnode = NULL;
pDnode = taosMemoryCalloc(1, sizeof(SDnode));
if (pDnode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;
}
if (dndInitVars(pDnode, pOption) != 0) {
dError("failed to init variables since %s", terrstr());
goto _OVER;
}
dndSetStatus(pDnode, DND_STAT_INIT);
dmGetMgmtFp(&pDnode->wrappers[DNODE]);
mmGetMgmtFp(&pDnode->wrappers[MNODE]);
vmGetMgmtFp(&pDnode->wrappers[VNODES]);
qmGetMgmtFp(&pDnode->wrappers[QNODE]);
smGetMgmtFp(&pDnode->wrappers[SNODE]);
bmGetMgmtFp(&pDnode->wrappers[BNODE]);
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
snprintf(path, sizeof(path), "%s%s%s", pDnode->dataDir, TD_DIRSEP, pWrapper->name);
pWrapper->path = strdup(path);
pWrapper->shm.id = -1;
pWrapper->pDnode = pDnode;
if (pWrapper->path == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;
}
pWrapper->procType = PROC_SINGLE;
taosInitRWLatch(&pWrapper->latch);
}
if (dndInitMsgHandle(pDnode) != 0) {
dError("failed to msg handles since %s", terrstr());
goto _OVER;
}
if (dndReadShmFile(pDnode) != 0) {
dError("failed to read shm file since %s", terrstr());
goto _OVER;
}
SMsgCb msgCb = dndCreateMsgcb(&pDnode->wrappers[0]);
tmsgSetDefaultMsgCb(&msgCb);
dInfo("dnode is created, data:%p", pDnode);
code = 0;
_OVER:
if (code != 0 && pDnode) {
dndClearVars(pDnode);
pDnode = NULL;
dError("failed to create dnode since %s", terrstr());
}
return pDnode;
}
EDndStatus dndGetStatus(SDnode *pDnode) { return pDnode->status; }
void dndClose(SDnode *pDnode) {
if (pDnode == NULL) return;
void dndSetStatus(SDnode *pDnode, EDndStatus status) {
if (pDnode->status != status) {
dDebug("dnode status set from %s to %s", dndStatStr(pDnode->status), dndStatStr(status));
pDnode->status = status;
if (dndGetStatus(pDnode) == DND_STAT_STOPPED) {
dError("dnode is shutting down, data:%p", pDnode);
return;
}
dInfo("start to close dnode, data:%p", pDnode);
dndSetStatus(pDnode, DND_STAT_STOPPED);
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
dndCloseNode(pWrapper);
}
dndClearVars(pDnode);
dInfo("dnode is closed, data:%p", pDnode);
}
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
SStartupReq *pStartup = &pDnode->startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);
pStartup->finished = 0;
void dndHandleEvent(SDnode *pDnode, EDndEvent event) {
dInfo("dnode receive %s event, data:%p", dndEventStr(event), pDnode);
if (event == DND_EVENT_STOP) {
pDnode->event = event;
}
}
void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) {
memcpy(pStartup, &pDnode->startup, sizeof(SStartupReq));
pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING);
SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
SMgmtWrapper *pRetWrapper = pWrapper;
taosRLockLatch(&pWrapper->latch);
if (pWrapper->deployed) {
int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
dTrace("node:%s, is acquired, refCount:%d", pWrapper->name, refCount);
} else {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
pRetWrapper = NULL;
}
taosRUnLockLatch(&pWrapper->latch);
return pRetWrapper;
}
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("startup req is received");
SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq));
dndGetStartup(pDnode, pStartup);
int32_t dndMarkWrapper(SMgmtWrapper *pWrapper) {
int32_t code = 0;
dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished);
SRpcMsg rpcRsp = {
.handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq), .ahandle = pReq->ahandle};
rpcSendResponse(&rpcRsp);
taosRLockLatch(&pWrapper->latch);
if (pWrapper->deployed || (pWrapper->procType == PROC_PARENT && pWrapper->required)) {
int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
dTrace("node:%s, is marked, refCount:%d", pWrapper->name, refCount);
} else {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
code = -1;
}
taosRUnLockLatch(&pWrapper->latch);
return code;
}
void dndReleaseWrapper(SMgmtWrapper *pWrapper) {
if (pWrapper == NULL) return;
taosRLockLatch(&pWrapper->latch);
int32_t refCount = atomic_sub_fetch_32(&pWrapper->refCount, 1);
taosRUnLockLatch(&pWrapper->latch);
dTrace("node:%s, is released, refCount:%d", pWrapper->name, refCount);
}
\ No newline at end of file
......@@ -66,7 +66,7 @@ void dndProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSet) {
dTrace("msg:%p, is created, handle:%p app:%p user:%s", pMsg, pRpc->handle, pRpc->ahandle, pMsg->user);
code = (*msgFp)(pWrapper, pMsg);
} else if (pWrapper->procType == PROC_PARENT) {
dTrace("msg:%p, is created and will put into child queue, handle:%p app:%p user:%s", pMsg, pRpc->handle,
dTrace("msg:%p, is created and put into child queue, handle:%p app:%p user:%s", pMsg, pRpc->handle,
pRpc->ahandle, pMsg->user);
code = taosProcPutToChildQ(pWrapper->pProc, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, PROC_REQ);
} else {
......
/*
* 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/>.
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
static int32_t dndInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
pDnode->numOfSupportVnodes = pOption->numOfSupportVnodes;
pDnode->serverPort = pOption->serverPort;
pDnode->dataDir = strdup(pOption->dataDir);
pDnode->localEp = strdup(pOption->localEp);
pDnode->localFqdn = strdup(pOption->localFqdn);
pDnode->firstEp = strdup(pOption->firstEp);
pDnode->secondEp = strdup(pOption->secondEp);
pDnode->disks = pOption->disks;
pDnode->numOfDisks = pOption->numOfDisks;
pDnode->ntype = pOption->ntype;
pDnode->rebootTime = taosGetTimestampMs();
if (pDnode->dataDir == NULL || pDnode->localEp == NULL || pDnode->localFqdn == NULL || pDnode->firstEp == NULL ||
pDnode->secondEp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return 0;
}
static void dndClearVars(SDnode *pDnode) {
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pMgmt = &pDnode->wrappers[n];
taosMemoryFreeClear(pMgmt->path);
}
dndCloseRuntimeFile(pDnode);
taosMemoryFreeClear(pDnode->localEp);
taosMemoryFreeClear(pDnode->localFqdn);
taosMemoryFreeClear(pDnode->firstEp);
taosMemoryFreeClear(pDnode->secondEp);
taosMemoryFreeClear(pDnode->dataDir);
taosMemoryFree(pDnode);
dDebug("dnode object memory is cleared, data:%p", pDnode);
}
SDnode *dndCreate(const SDnodeOpt *pOption) {
dInfo("start to create dnode object");
int32_t code = -1;
char path[PATH_MAX] = {0};
SDnode *pDnode = NULL;
pDnode = taosMemoryCalloc(1, sizeof(SDnode));
if (pDnode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;
}
if (dndInitVars(pDnode, pOption) != 0) {
dError("failed to init variables since %s", terrstr());
goto _OVER;
}
dndSetStatus(pDnode, DND_STAT_INIT);
dmGetMgmtFp(&pDnode->wrappers[DNODE]);
mmGetMgmtFp(&pDnode->wrappers[MNODE]);
vmGetMgmtFp(&pDnode->wrappers[VNODES]);
qmGetMgmtFp(&pDnode->wrappers[QNODE]);
smGetMgmtFp(&pDnode->wrappers[SNODE]);
bmGetMgmtFp(&pDnode->wrappers[BNODE]);
if (dndOpenRuntimeFile(pDnode) != 0) {
dError("failed to open runtime file since %s", terrstr());
goto _OVER;
}
if (dndInitServer(pDnode) != 0) {
dError("failed to init trans server since %s", terrstr());
goto _OVER;
}
if (dndInitClient(pDnode) != 0) {
dError("failed to init trans client since %s", terrstr());
goto _OVER;
}
if (dndInitMsgHandle(pDnode) != 0) {
goto _OVER;
}
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
snprintf(path, sizeof(path), "%s%s%s", pDnode->dataDir, TD_DIRSEP, pWrapper->name);
pWrapper->path = strdup(path);
pWrapper->pDnode = pDnode;
if (pWrapper->path == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;
}
pWrapper->procType = PROC_SINGLE;
taosInitRWLatch(&pWrapper->latch);
}
code = 0;
_OVER:
if (code != 0 && pDnode) {
dndClearVars(pDnode);
pDnode = NULL;
dError("failed to create dnode object since %s", terrstr());
} else {
dInfo("dnode object is created, data:%p", pDnode);
}
return pDnode;
}
void dndClose(SDnode *pDnode) {
if (pDnode == NULL) return;
if (dndGetStatus(pDnode) == DND_STAT_STOPPED) {
dError("dnode is shutting down, data:%p", pDnode);
return;
}
dInfo("start to close dnode, data:%p", pDnode);
dndSetStatus(pDnode, DND_STAT_STOPPED);
dndCleanupServer(pDnode);
dndCleanupClient(pDnode);
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
dndCloseNode(pWrapper);
}
dndClearVars(pDnode);
dInfo("dnode object is closed, data:%p", pDnode);
}
void dndHandleEvent(SDnode *pDnode, EDndEvent event) {
dInfo("dnode object receive event %d, data:%p", event, pDnode);
pDnode->event = event;
}
SMgmtWrapper *dndAcquireWrapper(SDnode *pDnode, ENodeType ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
SMgmtWrapper *pRetWrapper = pWrapper;
taosRLockLatch(&pWrapper->latch);
if (pWrapper->deployed) {
int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
dTrace("node:%s, is acquired, refCount:%d", pWrapper->name, refCount);
} else {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
pRetWrapper = NULL;
}
taosRUnLockLatch(&pWrapper->latch);
return pRetWrapper;
}
int32_t dndMarkWrapper(SMgmtWrapper *pWrapper) {
int32_t code = 0;
taosRLockLatch(&pWrapper->latch);
if (pWrapper->deployed || (pWrapper->procType == PROC_PARENT && pWrapper->required)) {
int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
dTrace("node:%s, is marked, refCount:%d", pWrapper->name, refCount);
} else {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
code = -1;
}
taosRUnLockLatch(&pWrapper->latch);
return code;
}
void dndReleaseWrapper(SMgmtWrapper *pWrapper) {
if (pWrapper == NULL) return;
taosRLockLatch(&pWrapper->latch);
int32_t refCount = atomic_sub_fetch_32(&pWrapper->refCount, 1);
taosRUnLockLatch(&pWrapper->latch);
dTrace("node:%s, is released, refCount:%d", pWrapper->name, refCount);
}
\ No newline at end of file
......@@ -62,3 +62,16 @@ const char *dndNodeProcStr(ENodeType ntype) {
return "taosd";
}
}
const char *dndEventStr(EDndEvent ev) {
switch (ev) {
case DND_EVENT_START:
return "start";
case DND_EVENT_STOP:
return "stop";
case DND_EVENT_CHILD:
return "child";
default:
return "UNKNOWN";
}
}
\ No newline at end of file
......@@ -187,7 +187,7 @@ static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueTyp
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL) {
dError("vgId:%d, failed to write msg:%p to vnode-queue since %s", pHead->vgId, pMsg, terrstr());
return -1;
return terrno;
}
int32_t code = 0;
......
......@@ -735,6 +735,9 @@ typedef struct {
int8_t createdBy; // STREAM_CREATED_BY__USER or SMA
int32_t fixedSinkVgId; // 0 for shuffle
int64_t smaId; // 0 for unused
int8_t trigger;
int32_t triggerParam;
int64_t waterMark;
char* sql;
char* logicalPlan;
char* physicalPlan;
......
......@@ -1309,7 +1309,7 @@ static int32_t mndGetDbMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMet
cols++;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "days");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
......@@ -1444,7 +1444,7 @@ static void dumpDbInfoToPayload(char *data, SDbObj *pDb, SShowObj *pShow, int32_
cols++;
pWrite = getDataPosition(data, pShow, cols, rows, rowCapacity);
*(int16_t *)pWrite = pDb->cfg.daysPerFile;
*(int32_t *)pWrite = pDb->cfg.daysPerFile;
cols++;
pWrite = getDataPosition(data, pShow, cols, rows, rowCapacity);
......
......@@ -53,7 +53,7 @@ static const SInfosTableSchema userDBSchema[] = {
{.name = "ntables", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
{.name = "replica", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
{.name = "quorum", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
{.name = "days", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
{.name = "days", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "keep", .bytes = 24 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
{.name = "cache", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "blocks", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
......@@ -89,7 +89,6 @@ static const SInfosTableSchema userStbsSchema[] = {
{.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "columns", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "tags", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "tables", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "last_update", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "table_comment", .bytes = 1024 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_INT},
};
......
......@@ -310,6 +310,8 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
mError("failed to process show-retrieve req:%p since %s", pShow, terrstr());
return -1;
}
pShow->numOfReads = 0;
}
ShowRetrieveFp retrieveFp = pMgmt->retrieveFps[pShow->type];
......@@ -374,7 +376,7 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
pReq->pRsp = pRsp;
pReq->rspLen = size;
if (rowsRead == 0 || rowsToRead == 0 || (rowsRead == rowsToRead && pShow->numOfRows == pShow->numOfReads)) {
if (rowsRead == 0 || rowsToRead == 0 || (rowsRead < rowsToRead)) {
pRsp->completed = 1;
mDebug("show:0x%" PRIx64 ", retrieve completed", pShow->id);
mndReleaseShowObj((SShowObj*) pShow, true);
......
......@@ -1608,7 +1608,6 @@ static int32_t mndRetrieveStb(SNodeMsg *pReq, SShowObj *pShow, char *data, int32
SStbObj *pStb = NULL;
int32_t cols = 0;
char *pWrite;
char prefix[TSDB_DB_FNAME_LEN] = {0};
SDbObj* pDb = NULL;
if (strlen(pShow->db) > 0) {
......@@ -1653,10 +1652,6 @@ static int32_t mndRetrieveStb(SNodeMsg *pReq, SShowObj *pShow, char *data, int32
*(int32_t *)pWrite = pStb->numOfTags;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int32_t *)pWrite = 0; // number of tables
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = pStb->updateTime; // number of tables
cols++;
......
......@@ -42,10 +42,10 @@ void* MndTestSma::BuildCreateDbReq(const char* dbname, int32_t* pContLen) {
createReq.numOfVgroups = 2;
createReq.cacheBlockSize = 16;
createReq.totalBlocks = 10;
createReq.daysPerFile = 10;
createReq.daysToKeep0 = 3650;
createReq.daysToKeep1 = 3650;
createReq.daysToKeep2 = 3650;
createReq.daysPerFile = 10 * 1440;
createReq.daysToKeep0 = 3650 * 1440;
createReq.daysToKeep1 = 3650 * 1440;
createReq.daysToKeep2 = 3650 * 1440;
createReq.minRows = 100;
createReq.maxRows = 4096;
createReq.commitTime = 3600;
......
......@@ -35,10 +35,10 @@ void* MndTestTopic::BuildCreateDbReq(const char* dbname, int32_t* pContLen) {
createReq.numOfVgroups = 2;
createReq.cacheBlockSize = 16;
createReq.totalBlocks = 10;
createReq.daysPerFile = 10;
createReq.daysToKeep0 = 3650;
createReq.daysToKeep1 = 3650;
createReq.daysToKeep2 = 3650;
createReq.daysPerFile = 10 * 1440;
createReq.daysToKeep0 = 3650 * 1440;
createReq.daysToKeep1 = 3650 * 1440;
createReq.daysToKeep2 = 3650 * 1440;
createReq.minRows = 100;
createReq.maxRows = 4096;
createReq.commitTime = 3600;
......
......@@ -324,10 +324,10 @@ TEST_F(MndTestUser, 03_Alter_User) {
createReq.numOfVgroups = 2;
createReq.cacheBlockSize = 16;
createReq.totalBlocks = 10;
createReq.daysPerFile = 10;
createReq.daysToKeep0 = 3650;
createReq.daysToKeep1 = 3650;
createReq.daysToKeep2 = 3650;
createReq.daysPerFile = 10 * 1440;
createReq.daysToKeep0 = 3650 * 1440;
createReq.daysToKeep1 = 3650 * 1440;
createReq.daysToKeep2 = 3650 * 1440;
createReq.minRows = 100;
createReq.maxRows = 4096;
createReq.commitTime = 3600;
......
......@@ -74,4 +74,4 @@ int tsdbApplyRtn(STsdbRepo *pRepo);
}
#endif
#endif /* _TD_TSDB_COMMIT_H_ */
\ No newline at end of file
#endif /* _TD_TSDB_COMMIT_H_ */
......@@ -79,4 +79,4 @@ static FORCE_INLINE STSchema *tsdbGetTableSchemaImpl(STable *pTable, bool lock,
}
#endif
#endif /*_TD_TSDB_DEF_H_*/
\ No newline at end of file
#endif /*_TD_TSDB_DEF_H_*/
......@@ -51,7 +51,7 @@ static FORCE_INLINE int32_t tsdbEncodeTSmaKey(int64_t groupId, TSKEY tsKey, void
return len;
}
static FORCE_INLINE int tsdbRLockSma(SSmaEnv *pEnv) {
static FORCE_INLINE int32_t tsdbRLockSma(SSmaEnv *pEnv) {
int code = taosThreadRwlockRdlock(&(pEnv->lock));
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(code);
......@@ -60,7 +60,7 @@ static FORCE_INLINE int tsdbRLockSma(SSmaEnv *pEnv) {
return 0;
}
static FORCE_INLINE int tsdbWLockSma(SSmaEnv *pEnv) {
static FORCE_INLINE int32_t tsdbWLockSma(SSmaEnv *pEnv) {
int code = taosThreadRwlockWrlock(&(pEnv->lock));
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(code);
......@@ -69,7 +69,7 @@ static FORCE_INLINE int tsdbWLockSma(SSmaEnv *pEnv) {
return 0;
}
static FORCE_INLINE int tsdbUnLockSma(SSmaEnv *pEnv) {
static FORCE_INLINE int32_t tsdbUnLockSma(SSmaEnv *pEnv) {
int code = taosThreadRwlockUnlock(&(pEnv->lock));
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(code);
......
......@@ -194,7 +194,7 @@ void tqClose(STQ*);
int tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t version);
int tqCommit(STQ*);
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId);
int32_t tqProcessSetConnReq(STQ* pTq, char* msg);
int32_t tqProcessRebReq(STQ* pTq, char* msg);
int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId);
......
......@@ -27,5 +27,5 @@ void metaCloseUidGnrt(SMeta *pMeta) { /* TODO */
tb_uid_t metaGenerateUid(SMeta *pMeta) {
// Generate a new table UID
return ++(pMeta->uidGnrt.nextUid);
return tGenIdPI32();
}
......@@ -250,7 +250,7 @@ int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsu
return 0;
}
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
SMqPollReq* pReq = pMsg->pCont;
int64_t consumerId = pReq->consumerId;
int64_t fetchOffset;
......@@ -264,6 +264,8 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
fetchOffset = pReq->currentOffset + 1;
}
/*printf("tmq poll vg %d req %ld %ld\n", pTq->pVnode->vgId, pReq->currentOffset, fetchOffset);*/
SMqPollRsp rsp = {
/*.consumerId = consumerId,*/
.numOfTopics = 0,
......@@ -288,62 +290,77 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
rsp.reqOffset = pReq->currentOffset;
rsp.skipLogNum = 0;
SWalHead* pHead;
while (1) {
/*if (fetchOffset > walGetLastVer(pTq->pWal) || walReadWithHandle(pTopic->pReadhandle, fetchOffset) < 0) {*/
if (walReadWithHandle(pTopic->pReadhandle, fetchOffset) < 0) {
SWalReadHead* pHead;
if (walReadWithHandle_s(pTopic->pReadhandle, fetchOffset, &pHead) < 0) {
// TODO: no more log, set timer to wait blocking time
// if data inserted during waiting, launch query and
// response to user
break;
}
int8_t pos = fetchOffset % TQ_BUFFER_SIZE;
pHead = pTopic->pReadhandle->pHead;
if (pHead->head.msgType == TDMT_VND_SUBMIT) {
SSubmitReq* pCont = (SSubmitReq*)&pHead->head.body;
qTaskInfo_t task = pTopic->buffer.output[pos].task;
/*printf("vg %d offset %ld msgType %d from epoch %d\n", pTq->pVnode->vgId, fetchOffset, pHead->msgType,
* pReq->epoch);*/
/*int8_t pos = fetchOffset % TQ_BUFFER_SIZE;*/
/*pHead = pTopic->pReadhandle->pHead;*/
if (pHead->msgType == TDMT_VND_SUBMIT) {
SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
/*printf("from topic %s from consumer\n", pTopic->topicName, consumerId);*/
qTaskInfo_t task = pTopic->buffer.output[workerId].task;
ASSERT(task);
qSetStreamInput(task, pCont, STREAM_DATA_TYPE_SUBMIT_BLOCK);
SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
while (1) {
SSDataBlock* pDataBlock;
SSDataBlock* pDataBlock = NULL;
uint64_t ts;
if (qExecTask(task, &pDataBlock, &ts) < 0) {
ASSERT(false);
}
if (pDataBlock == NULL) {
fetchOffset++;
pos = fetchOffset % TQ_BUFFER_SIZE;
rsp.skipLogNum++;
/*pos = fetchOffset % TQ_BUFFER_SIZE;*/
break;
}
taosArrayPush(pRes, pDataBlock);
rsp.schema = pTopic->buffer.output[pos].pReadHandle->pSchemaWrapper;
rsp.rspOffset = fetchOffset;
}
if (taosArrayGetSize(pRes) == 0) {
fetchOffset++;
rsp.skipLogNum++;
taosArrayDestroy(pRes);
continue;
}
rsp.schema = pTopic->buffer.output[workerId].pReadHandle->pSchemaWrapper;
rsp.rspOffset = fetchOffset;
rsp.numOfTopics = 1;
rsp.pBlockData = pRes;
rsp.numOfTopics = 1;
rsp.pBlockData = pRes;
int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqPollRsp(NULL, &rsp);
void* buf = rpcMallocCont(tlen);
if (buf == NULL) {
pMsg->code = -1;
return -1;
}
((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
((SMqRspHead*)buf)->epoch = pReq->epoch;
((SMqRspHead*)buf)->consumerId = consumerId;
void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
tEncodeSMqPollRsp(&abuf, &rsp);
/*taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);*/
pMsg->pCont = buf;
pMsg->contLen = tlen;
pMsg->code = 0;
tmsgSendRsp(pMsg);
return 0;
int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqPollRsp(NULL, &rsp);
void* buf = rpcMallocCont(tlen);
if (buf == NULL) {
pMsg->code = -1;
taosMemoryFree(pHead);
return -1;
}
((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
((SMqRspHead*)buf)->epoch = pReq->epoch;
((SMqRspHead*)buf)->consumerId = consumerId;
void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
tEncodeSMqPollRsp(&abuf, &rsp);
/*taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);*/
pMsg->pCont = buf;
pMsg->contLen = tlen;
pMsg->code = 0;
/*printf("vg %d offset %ld msgType %d from epoch %d actual rsp\n", pTq->pVnode->vgId, fetchOffset,
* pHead->msgType,*/
/*pReq->epoch);*/
tmsgSendRsp(pMsg);
taosMemoryFree(pHead);
return 0;
} else {
taosMemoryFree(pHead);
fetchOffset++;
rsp.skipLogNum++;
}
......@@ -368,6 +385,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
pMsg->contLen = tlen;
pMsg->code = 0;
tmsgSendRsp(pMsg);
/*printf("vg %d offset %ld from epoch %d not rsp\n", pTq->pVnode->vgId, fetchOffset, pReq->epoch);*/
/*}*/
return 0;
......@@ -432,7 +450,9 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
};
pTopic->buffer.output[i].pReadHandle = pReadHandle;
pTopic->buffer.output[i].task = qCreateStreamExecTaskInfo(req.qmsg, &handle);
ASSERT(pTopic->buffer.output[i].task);
}
printf("set topic %s to consumer %ld\n", pTopic->topicName, req.consumerId);
taosArrayPush(pConsumer->topics, pTopic);
tqHandleMovePut(pTq->tqMeta, req.consumerId, pConsumer);
tqHandleCommit(pTq->tqMeta, req.consumerId);
......
......@@ -167,7 +167,8 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
break;
}
if (colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) {
if (colDataAppend(pColData, curRow, sVal.val, false) < 0) {
/*if (colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) {*/
taosArrayDestroyEx(pArray, (void (*)(void*))tDeleteSSDataBlock);
return NULL;
}
......
......@@ -1394,7 +1394,7 @@ int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDF
tsdbDebug("vgId:%d uid:%" PRId64 " a block of data is written to file %s, offset %" PRId64
" numOfRows %d len %d numOfCols %" PRId16 " keyFirst %" PRId64 " keyLast %" PRId64,
REPO_ID(pRepo), TABLE_TID(pTable), TSDB_FILE_FULL_NAME(pDFile), offset, rowsToWrite, pBlock->len,
REPO_ID(pRepo), TABLE_UID(pTable), TSDB_FILE_FULL_NAME(pDFile), offset, rowsToWrite, pBlock->len,
pBlock->numOfCols, pBlock->keyFirst, pBlock->keyLast);
return 0;
......
......@@ -4068,4 +4068,4 @@ void getTableListfromSkipList(tExprNode *pExpr, SSkipList *pSkipList, SArray *re
//apply the hierarchical filter expression to every node in skiplist to find the qualified nodes
applyFilterToSkipListNode(pSkipList, pExpr, result, param);
}
#endif
\ No newline at end of file
#endif
......@@ -651,9 +651,8 @@ static int32_t tsdbGetSmaStorageLevel(int64_t interval, int8_t intervalUnit) {
*/
static int32_t tsdbInsertTSmaBlocks(STSmaWriteH *pSmaH, void *smaKey, uint32_t keyLen, void *pData, uint32_t dataLen) {
SDBFile *pDBFile = &pSmaH->dFile;
tsdbDebug("vgId:%d insert sma data blocks into %s: smaKey %" PRIx64 "-%" PRIu16 "-%" PRIx64 ", dataLen %d",
REPO_ID(pSmaH->pTsdb), pDBFile->path, *(tb_uid_t *)smaKey, *(uint16_t *)POINTER_SHIFT(smaKey, 8),
*(int64_t *)POINTER_SHIFT(smaKey, 10), dataLen);
printf("\nvgId:%d insert sma data blocks into %s: smaKey %" PRIx64 "-%" PRIx64 ", dataLen %" PRIu32 "\n",
REPO_ID(pSmaH->pTsdb), pDBFile->path, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), dataLen);
// TODO: insert sma data blocks into B+Tree(TDB)
if (tsdbSaveSmaToDB(pDBFile, smaKey, keyLen, pData, dataLen) != 0) {
......@@ -874,7 +873,6 @@ static int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, int64_t indexUid, const char
// key: skey + groupId
char smaKey[SMA_KEY_LEN] = {0};
void *pSmaKey = &smaKey;
char dataBuf[512] = {0};
void *pDataBuf = &dataBuf;
int32_t sz = taosArrayGetSize(pDataBlocks);
......@@ -887,6 +885,7 @@ static int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, int64_t indexUid, const char
for (int32_t j = 0; j < rows; ++j) {
printf("|");
TSKEY skey = TSKEY_INITIAL_VAL; // the start key of TS window by interval
void *pSmaKey = &smaKey;
int32_t tlen = 0;
for (int32_t k = 0; k < colNum; ++k) {
SColumnInfoData *pColInfoData = *(SColumnInfoData **)taosArrayGet(pDataBlock->pDataBlock, k);
......@@ -894,7 +893,7 @@ static int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, int64_t indexUid, const char
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_TIMESTAMP:
skey = *(TSKEY *)var;
printf(" skey = %" PRIi64 " groupId = %" PRId64 "|", skey, groupId);
printf("==> skey = %" PRIi64 " groupId = %" PRId64 "|", skey, groupId);
tsdbEncodeTSmaKey(groupId, skey, &pSmaKey);
break;
case TSDB_DATA_TYPE_BOOL:
......@@ -976,7 +975,7 @@ static int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, int64_t indexUid, const char
}
}
if (tsdbInsertTSmaBlocks(&tSmaH, pSmaKey, SMA_KEY_LEN, pDataBuf, tlen) != 0) {
if (tsdbInsertTSmaBlocks(&tSmaH, &smaKey, SMA_KEY_LEN, pDataBuf, tlen) != 0) {
tsdbWarn("vgId:%d insert tSma data blocks failed for index %" PRIi64 ", skey %" PRIi64 ", groupId %" PRIi64
" since %s",
REPO_ID(pTsdb), indexUid, skey, groupId, tstrerror(terrno));
......@@ -1308,22 +1307,19 @@ static int32_t tsdbGetTSmaDataImpl(STsdb *pTsdb, char *pData, int64_t indexUid,
return TSDB_CODE_FAILED;
}
char smaKey[SMA_KEY_LEN] = {0};
void *pSmaKey = &smaKey;
char smaKey[SMA_KEY_LEN] = {0};
void *pSmaKey = &smaKey;
int64_t queryGroupId = 1;
tsdbEncodeTSmaKey(queryGroupId, querySKey, (void **)&pSmaKey);
tsdbDebug("vgId:%d get sma data from %s: smaKey %" PRIx64 "-%" PRIu16 "-%" PRIx64 ", keyLen %d", REPO_ID(pTsdb),
tReadH.dFile.path, *(tb_uid_t *)smaKey, *(uint16_t *)POINTER_SHIFT(smaKey, 8),
*(int64_t *)POINTER_SHIFT(smaKey, 10), SMA_KEY_LEN);
tsdbDebug("vgId:%d get sma data from %s: smaKey %" PRIx64 "-%" PRIx64 ", keyLen %d", REPO_ID(pTsdb),
tReadH.dFile.path, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), SMA_KEY_LEN);
void *result = NULL;
uint32_t valueSize = 0;
if ((result = tsdbGetSmaDataByKey(&tReadH.dFile, smaKey, SMA_KEY_LEN, &valueSize)) == NULL) {
tsdbWarn("vgId:%d get sma data failed from smaIndex %" PRIi64 ", smaKey %" PRIx64 "-%" PRIu16 "-%" PRIx64
" since %s",
REPO_ID(pTsdb), indexUid, *(tb_uid_t *)smaKey, *(uint16_t *)POINTER_SHIFT(smaKey, 8),
*(int64_t *)POINTER_SHIFT(smaKey, 10), tstrerror(terrno));
tsdbWarn("vgId:%d get sma data failed from smaIndex %" PRIi64 ", smaKey %" PRIx64 "-%" PRIx64 " since %s",
REPO_ID(pTsdb), indexUid, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), tstrerror(terrno));
tsdbCloseDBF(&tReadH.dFile);
return TSDB_CODE_FAILED;
}
......
......@@ -66,12 +66,12 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
case TDMT_VND_TABLE_META:
return vnodeGetTableMeta(pVnode, pMsg);
case TDMT_VND_CONSUME:
return tqProcessPollReq(pVnode->pTq, pMsg);
return tqProcessPollReq(pVnode->pTq, pMsg, pInfo->workerId);
case TDMT_VND_TASK_PIPE_EXEC:
case TDMT_VND_TASK_MERGE_EXEC:
return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen, pInfo->workerId);
return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen, 0);
case TDMT_VND_STREAM_TRIGGER:
return tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen, pInfo->workerId);
return tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen, 0);
case TDMT_VND_QUERY_HEARTBEAT:
return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg);
default:
......
......@@ -165,6 +165,7 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
// }
break;
case TDMT_VND_SUBMIT:
/*printf("vnode %d write data %ld\n", pVnode->vgId, ver);*/
if (pVnode->config.streamMode == 0) {
if (tsdbInsertData(pVnode->pTsdb, (SSubmitReq *)ptr, NULL) < 0) {
// TODO: handle error
......
......@@ -59,20 +59,21 @@ TEST(testCase, unionEncodeDecodeTest) {
void *buf = taosMemoryMalloc(1024);
void *pBuf = buf;
void *qBuf = buf;
int32_t tlen = 0;
tlen += taosEncodeFixedU8(&buf, sut.info);
tlen += taosEncodeFixedI16(&buf, sut.nBSmaCols);
tlen += taosEncodeFixedU8(&pBuf, sut.info);
tlen += taosEncodeFixedI16(&pBuf, sut.nBSmaCols);