xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/3.0' into feature/config

This commit is contained in:
Shengliang Guan 2022-02-21 13:47:48 +08:00
parent f22d662b88 c5854b83d1
commit 84fd629b9c
56 changed files with 4157 additions and 1918 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
example/src/tmq.c
View File

@ -28,7 +28,7 @@ int32_t init_env() {
return -1;
}
TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2");
TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 1");
if (taos_errno(pRes) != 0) {
printf("error in create db, reason:%s\n", taos_errstr(pRes));
return -1;
@ -62,6 +62,23 @@ int32_t init_env() {
return -1;
}
taos_free_result(pRes);
return 0;
}
int32_t create_topic() {
printf("create topic");
TAOS_RES* pRes;
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
if (pConn == NULL) {
return -1;
}
pRes = taos_query(pConn, "use abc1");
if (taos_errno(pRes) != 0) {
printf("error in use db, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
const char* sql = "select * from tu1";
pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));
@ -193,6 +210,7 @@ int main(int argc, char* argv[]) {
printf("env init\n");
code = init_env();
}
create_topic();
tmq_t* tmq = build_consumer();
tmq_list_t* topic_list = build_topic_list();
/*perf_loop(tmq, topic_list);*/

12
include/common/tmsg.h
View File

@ -524,6 +524,7 @@ typedef struct {
int8_t update;
int8_t cacheLastRow;
int8_t ignoreExist;
int8_t streamMode;
} SCreateDbReq;
int32_t tSerializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq);
@ -744,6 +745,7 @@ typedef struct {
int8_t cacheLastRow;
int8_t replica;
int8_t selfIndex;
int8_t streamMode;
SReplica replicas[TSDB_MAX_REPLICA];
} SCreateVnodeReq, SAlterVnodeReq;
@ -1758,6 +1760,11 @@ typedef struct {
char cgroup[TSDB_CONSUMER_GROUP_LEN];
} SMqOffset;
typedef struct {
int32_t vgId;
SArray* offsets; // SArray<SMqOffset>
} SMqVgOffsets;
typedef struct {
int32_t num;
SMqOffset* offsets;
@ -1768,8 +1775,8 @@ typedef struct {
} SMqCMResetOffsetRsp;
typedef struct {
int32_t num;
SMqOffset* offsets;
int64_t leftForVer;
SMqVgOffsets offsets;
} SMqMVResetOffsetReq;
typedef struct {
@ -1780,7 +1787,6 @@ int32_t tEncodeSMqOffset(SCoder* encoder, const SMqOffset* pOffset);
int32_t tDecodeSMqOffset(SCoder* decoder, SMqOffset* pOffset);
int32_t tEncodeSMqCMResetOffsetReq(SCoder* encoder, const SMqCMResetOffsetReq* pReq);
int32_t tDecodeSMqCMResetOffsetReq(SCoder* decoder, SMqCMResetOffsetReq* pReq);
int32_t tEncodeSMqMVResetOffsetReq(SCoder* encoder, const SMqMVResetOffsetReq* pReq);
int32_t tDecodeSMqMVResetOffsetReq(SCoder* decoder, SMqMVResetOffsetReq* pReq);

161
include/common/ttokendef.h
View File

@ -126,86 +126,87 @@
#define TK_PRECISION 108
#define TK_UPDATE 109
#define TK_CACHELAST 110
#define TK_UNSIGNED 111
#define TK_TAGS 112
#define TK_USING 113
#define TK_NULL 114
#define TK_NOW 115
#define TK_SELECT 116
#define TK_UNION 117
#define TK_ALL 118
#define TK_DISTINCT 119
#define TK_FROM 120
#define TK_VARIABLE 121
#define TK_INTERVAL 122
#define TK_EVERY 123
#define TK_SESSION 124
#define TK_STATE_WINDOW 125
#define TK_FILL 126
#define TK_SLIDING 127
#define TK_ORDER 128
#define TK_BY 129
#define TK_ASC 130
#define TK_GROUP 131
#define TK_HAVING 132
#define TK_LIMIT 133
#define TK_OFFSET 134
#define TK_SLIMIT 135
#define TK_SOFFSET 136
#define TK_WHERE 137
#define TK_RESET 138
#define TK_QUERY 139
#define TK_SYNCDB 140
#define TK_ADD 141
#define TK_COLUMN 142
#define TK_MODIFY 143
#define TK_TAG 144
#define TK_CHANGE 145
#define TK_SET 146
#define TK_KILL 147
#define TK_CONNECTION 148
#define TK_STREAM 149
#define TK_COLON 150
#define TK_ABORT 151
#define TK_AFTER 152
#define TK_ATTACH 153
#define TK_BEFORE 154
#define TK_BEGIN 155
#define TK_CASCADE 156
#define TK_CLUSTER 157
#define TK_CONFLICT 158
#define TK_COPY 159
#define TK_DEFERRED 160
#define TK_DELIMITERS 161
#define TK_DETACH 162
#define TK_EACH 163
#define TK_END 164
#define TK_EXPLAIN 165
#define TK_FAIL 166
#define TK_FOR 167
#define TK_IGNORE 168
#define TK_IMMEDIATE 169
#define TK_INITIALLY 170
#define TK_INSTEAD 171
#define TK_KEY 172
#define TK_OF 173
#define TK_RAISE 174
#define TK_REPLACE 175
#define TK_RESTRICT 176
#define TK_ROW 177
#define TK_STATEMENT 178
#define TK_TRIGGER 179
#define TK_VIEW 180
#define TK_SEMI 181
#define TK_NONE 182
#define TK_PREV 183
#define TK_LINEAR 184
#define TK_IMPORT 185
#define TK_TBNAME 186
#define TK_JOIN 187
#define TK_INSERT 188
#define TK_INTO 189
#define TK_VALUES 190
#define TK_STREAM 111
#define TK_MODE 112
#define TK_UNSIGNED 113
#define TK_TAGS 114
#define TK_USING 115
#define TK_NULL 116
#define TK_NOW 117
#define TK_SELECT 118
#define TK_UNION 119
#define TK_ALL 120
#define TK_DISTINCT 121
#define TK_FROM 122
#define TK_VARIABLE 123
#define TK_INTERVAL 124
#define TK_EVERY 125
#define TK_SESSION 126
#define TK_STATE_WINDOW 127
#define TK_FILL 128
#define TK_SLIDING 129
#define TK_ORDER 130
#define TK_BY 131
#define TK_ASC 132
#define TK_GROUP 133
#define TK_HAVING 134
#define TK_LIMIT 135
#define TK_OFFSET 136
#define TK_SLIMIT 137
#define TK_SOFFSET 138
#define TK_WHERE 139
#define TK_RESET 140
#define TK_QUERY 141
#define TK_SYNCDB 142
#define TK_ADD 143
#define TK_COLUMN 144
#define TK_MODIFY 145
#define TK_TAG 146
#define TK_CHANGE 147
#define TK_SET 148
#define TK_KILL 149
#define TK_CONNECTION 150
#define TK_COLON 151
#define TK_ABORT 152
#define TK_AFTER 153
#define TK_ATTACH 154
#define TK_BEFORE 155
#define TK_BEGIN 156
#define TK_CASCADE 157
#define TK_CLUSTER 158
#define TK_CONFLICT 159
#define TK_COPY 160
#define TK_DEFERRED 161
#define TK_DELIMITERS 162
#define TK_DETACH 163
#define TK_EACH 164
#define TK_END 165
#define TK_EXPLAIN 166
#define TK_FAIL 167
#define TK_FOR 168
#define TK_IGNORE 169
#define TK_IMMEDIATE 170
#define TK_INITIALLY 171
#define TK_INSTEAD 172
#define TK_KEY 173
#define TK_OF 174
#define TK_RAISE 175
#define TK_REPLACE 176
#define TK_RESTRICT 177
#define TK_ROW 178
#define TK_STATEMENT 179
#define TK_TRIGGER 180
#define TK_VIEW 181
#define TK_SEMI 182
#define TK_NONE 183
#define TK_PREV 184
#define TK_LINEAR 185
#define TK_IMPORT 186
#define TK_TBNAME 187
#define TK_JOIN 188
#define TK_INSERT 189
#define TK_INTO 190
#define TK_VALUES 191
#define NEW_TK_OR 1
#define NEW_TK_AND 2

8
include/libs/nodes/nodes.h
View File

@ -62,6 +62,7 @@ typedef enum ENodeType {
QUERY_NODE_NODE_LIST,
QUERY_NODE_FILL,
QUERY_NODE_COLUMN_REF,
QUERY_NODE_TARGET,
// Only be used in parser module.
QUERY_NODE_RAW_EXPR,
@ -72,8 +73,10 @@ typedef enum ENodeType {
QUERY_NODE_SHOW_STMT,
QUERY_NODE_LOGIC_PLAN_SCAN,
QUERY_NODE_LOGIC_PLAN_JOIN,
QUERY_NODE_LOGIC_PLAN_FILTER,
QUERY_NODE_LOGIC_PLAN_AGG
QUERY_NODE_LOGIC_PLAN_AGG,
QUERY_NODE_LOGIC_PLAN_PROJECT
} ENodeType;
/**
@ -91,7 +94,7 @@ typedef struct SListCell {
} SListCell;
typedef struct SNodeList {
int16_t length;
int32_t length;
SListCell* pHead;
SListCell* pTail;
} SNodeList;
@ -101,6 +104,7 @@ void nodesDestroyNode(SNode* pNode);
SNodeList* nodesMakeList();
int32_t nodesListAppend(SNodeList* pList, SNode* pNode);
int32_t nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc);
SListCell* nodesListErase(SNodeList* pList, SListCell* pCell);
SNode* nodesListGetNode(SNodeList* pList, int32_t index);
void nodesDestroyList(SNodeList* pList);

65
include/libs/nodes/plannodes.h Normal file
View File

@ -0,0 +1,65 @@
/*
* 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_PLANN_NODES_H_
#define _TD_PLANN_NODES_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "querynodes.h"
typedef struct SLogicNode {
ENodeType type;
int32_t id;
SNodeList* pTargets; // SColumnNode
SNode* pConditions;
SNodeList* pChildren;
struct SLogicNode* pParent;
} SLogicNode;
typedef struct SScanLogicNode {
SLogicNode node;
SNodeList* pScanCols;
struct STableMeta* pMeta;
} SScanLogicNode;
typedef struct SJoinLogicNode {
SLogicNode node;
EJoinType joinType;
SNode* pOnConditions;
} SJoinLogicNode;
typedef struct SFilterLogicNode {
SLogicNode node;
} SFilterLogicNode;
typedef struct SAggLogicNode {
SLogicNode node;
SNodeList* pGroupKeys;
SNodeList* pAggFuncs;
} SAggLogicNode;
typedef struct SProjectLogicNode {
SLogicNode node;
SNodeList* pProjections;
} SProjectLogicNode;
#ifdef __cplusplus
}
#endif
#endif /*_TD_PLANN_NODES_H_*/

14
include/libs/nodes/querynodes.h
View File

@ -62,8 +62,10 @@ typedef struct SColumnNode {
typedef struct SColumnRefNode {
ENodeType type;
int32_t tupleId;
int32_t slotId;
SDataType dataType;
int16_t tupleId;
int16_t slotId;
int16_t columnId;
} SColumnRefNode;
typedef struct SValueNode {
@ -106,6 +108,12 @@ typedef enum EOperatorType {
OP_TYPE_NMATCH,
OP_TYPE_IS_NULL,
OP_TYPE_IS_NOT_NULL,
OP_TYPE_IS_TRUE,
OP_TYPE_IS_FALSE,
OP_TYPE_IS_UNKNOWN,
OP_TYPE_IS_NOT_TRUE,
OP_TYPE_IS_NOT_FALSE,
OP_TYPE_IS_NOT_UNKNOWN,
// json operator
OP_TYPE_JSON_GET_VALUE,
@ -285,7 +293,7 @@ typedef enum ESqlClause {
void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);
void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewriter rewriter, void* pContext);
int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, uint64_t tableId, bool realCol, SNodeList** pCols);
int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, const char* pTableAlias, SNodeList** pCols);
typedef bool (*FFuncClassifier)(int32_t funcId);
int32_t nodesCollectFuncs(SSelectStmt* pSelect, FFuncClassifier classifier, SNodeList** pFuncs);

34
source/libs/tdb/src/inc/tdb_db.h → include/libs/parser/newParser.h
View File

@ -13,33 +13,31 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_TDB_DB_H_
#define _TD_TDB_DB_H_
#include "tdb_mpool.h"
#ifndef _TD_NEW_PARSER_H_
#define _TD_NEW_PARSER_H_
#ifdef __cplusplus
extern "C" {
#endif
typedef struct TDB TDB;
#include "parser.h"
struct TDB {
char * fname;
char * dbname;
TDB_MPFILE *mpf;
// union {
// TDB_BTREE *btree;
// TDB_HASH * hash;
// TDB_HEAP * heap;
// } dbam; // db access method
};
typedef enum EStmtType {
STMT_TYPE_CMD = 1,
STMT_TYPE_QUERY
} EStmtType;
int tdbOpen(TDB **dbpp, const char *fname, const char *dbname, uint32_t flags);
int tdbClose(TDB *dbp, uint32_t flags);
typedef struct SQuery {
EStmtType stmtType;
SNode* pRoot;
int32_t numOfResCols;
SSchema* pResSchema;
} SQuery;
int32_t parser(SParseContext* pParseCxt, SQuery* pQuery);
#ifdef __cplusplus
}
#endif
#endif /*_TD_TDB_DB_H_*/
#endif /*_TD_NEW_PARSER_H_*/

2
include/util/tdef.h
View File

@ -333,6 +333,8 @@ do { \
#define TSDB_QUERY_TYPE_NON_TYPE 0x00u // none type
#define TSDB_QUERY_TYPE_FREE_RESOURCE 0x01u // free qhandle at vnode
#define TSDB_QUERY_TYPE_NON_TYPE 0x00u // none type
#define TSDB_QUERY_TYPE_FREE_RESOURCE 0x01u // free qhandle at vnode
#define TSDB_META_COMPACT_RATIO 0 // disable tsdb meta compact by default

50
include/util/tjson.h Normal file
View File

@ -0,0 +1,50 @@
/*
* 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_UTIL_JSON_H_
#define _TD_UTIL_JSON_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "os.h"
typedef void SJson;
SJson* tjsonCreateObject();
void tjsonDelete(SJson* pJson);
SJson* tjsonAddArrayToObject(SJson* pJson, const char* pName);
int32_t tjsonAddIntegerToObject(SJson* pJson, const char* pName, const uint64_t number);
int32_t tjsonAddStringToObject(SJson* pJson, const char* pName, const char* pVal);
int32_t tjsonAddItemToObject(SJson* pJson, const char* pName, SJson* pItem);
int32_t tjsonAddItemToArray(SJson* pJson, SJson* pItem);
typedef int32_t (*FToJson)(const void* pObj, SJson* pJson);
int32_t tjsonAddObject(SJson* pJson, const char* pName, FToJson func, const void* pObj);
int32_t tjsonAddItem(SJson* pJson, FToJson func, const void* pObj);
typedef int32_t (*FFromJson)(const SJson* pJson, void* pObj);
char* tjsonToString(const SJson* pJson);
#ifdef __cplusplus
}
#endif
#endif /*_TD_UTIL_JSON_H_*/

8
source/client/src/tmq.c
View File

@ -75,7 +75,7 @@ struct tmq_message_t {
SMqConsumeRsp rsp;
};
typedef struct SMqClientVg {
typedef struct {
// statistics
int64_t pollCnt;
// offset
@ -86,7 +86,7 @@ typedef struct SMqClientVg {
SEpSet epSet;
} SMqClientVg;
typedef struct SMqClientTopic {
typedef struct {
// subscribe info
int32_t sqlLen;
char* sql;
@ -269,7 +269,7 @@ tmq_resp_err_t tmq_reset_offset(tmq_t* tmq, const tmq_topic_vgroup_list_t* offse
tsem_wait(&param.rspSem);
tsem_destroy(&param.rspSem);
return TMQ_RESP_ERR__SUCCESS;
return param.rspErr;
}
tmq_resp_err_t tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
@ -779,11 +779,9 @@ tmq_message_t* tmq_consumer_poll(tmq_t* tmq, int64_t blocking_time) {
param->pVg = pVg;
tsem_init(&param->rspSem, 0, 0);
SRequestObj* pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_VND_CONSUME);
pRequest->body.requestMsg = (SDataBuf){.pData = pReq, .len = sizeof(SMqConsumeReq), .handle = NULL};
SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
sendInfo->requestObjRefId = 0;
sendInfo->param = param;

42
source/common/src/tmsg.c
View File

@ -34,6 +34,7 @@ int32_t tInitSubmitMsgIter(SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
}
pIter->totalLen = pMsg->length;
ASSERT(pIter->totalLen > 0);
pIter->len = 0;
pIter->pMsg = pMsg;
if (pMsg->length <= sizeof(SSubmitReq)) {
@ -45,11 +46,14 @@ int32_t tInitSubmitMsgIter(SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
}
int32_t tGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
ASSERT(pIter->len >= 0);
if (pIter->len == 0) {
pIter->len += sizeof(SSubmitReq);
} else {
SSubmitBlk *pSubmitBlk = (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
pIter->len += (sizeof(SSubmitBlk) + pSubmitBlk->dataLen + pSubmitBlk->schemaLen);
ASSERT(pIter->len > 0);
}
if (pIter->len > pIter->totalLen) {
@ -1275,6 +1279,7 @@ int32_t tSerializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq) {
if (tEncodeI8(&encoder, pReq->update) < 0) return -1;
if (tEncodeI8(&encoder, pReq->cacheLastRow) < 0) return -1;
if (tEncodeI8(&encoder, pReq->ignoreExist) < 0) return -1;
if (tEncodeI8(&encoder, pReq->streamMode) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -1307,6 +1312,7 @@ int32_t tDeserializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq)
if (tDecodeI8(&decoder, &pReq->update) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->cacheLastRow) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->ignoreExist) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->streamMode) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
@ -2107,6 +2113,7 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
if (tEncodeI8(&encoder, pReq->cacheLastRow) < 0) return -1;
if (tEncodeI8(&encoder, pReq->replica) < 0) return -1;
if (tEncodeI8(&encoder, pReq->selfIndex) < 0) return -1;
if (tEncodeI8(&encoder, pReq->streamMode) < 0) return -1;
for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
SReplica *pReplica = &pReq->replicas[i];
if (tEncodeSReplica(&encoder, pReplica) < 0) return -1;
@ -2146,6 +2153,7 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
if (tDecodeI8(&decoder, &pReq->cacheLastRow) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->replica) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->selfIndex) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->streamMode) < 0) return -1;
for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
SReplica *pReplica = &pReq->replicas[i];
if (tDecodeSReplica(&decoder, pReplica) < 0) return -1;
@ -2348,6 +2356,34 @@ int32_t tDecodeSMqOffset(SCoder *decoder, SMqOffset *pOffset) {
return 0;
}
int32_t tEncodeSMqVgOffsets(SCoder *encoder, const SMqVgOffsets *pOffsets) {
if (tStartEncode(encoder) < 0) return -1;
if (tEncodeI32(encoder, pOffsets->vgId) < 0) return -1;
int32_t sz = taosArrayGetSize(pOffsets->offsets);
if (tEncodeI32(encoder, sz) < 0) return -1;
for (int32_t i = 0; i < sz; i++) {
SMqOffset *offset = taosArrayGet(pOffsets->offsets, i);
if (tEncodeSMqOffset(encoder, offset) < 0) return -1;
}
tEndEncode(encoder);
return encoder->pos;
}
int32_t tDecodeSMqVgOffsets(SCoder *decoder, SMqVgOffsets *pOffsets) {
int32_t sz;
if (tStartDecode(decoder) < 0) return -1;
if (tDecodeI32(decoder, &pOffsets->vgId) < 0) return -1;
if (tDecodeI32(decoder, &sz) < 0) return -1;
pOffsets->offsets = taosArrayInit(sz, sizeof(SMqOffset));
for (int32_t i = 0; i < sz; i++) {
SMqOffset offset;
if (tDecodeSMqOffset(decoder, &offset) < 0) return -1;
taosArrayPush(pOffsets->offsets, &offset);
}
tEndDecode(decoder);
return 0;
}
int32_t tEncodeSMqCMResetOffsetReq(SCoder *encoder, const SMqCMResetOffsetReq *pReq) {
if (tStartEncode(encoder) < 0) return -1;
if (tEncodeI32(encoder, pReq->num) < 0) return -1;
@ -2359,17 +2395,20 @@ int32_t tEncodeSMqCMResetOffsetReq(SCoder *encoder, const SMqCMResetOffsetReq *p
}
int32_t tDecodeSMqCMResetOffsetReq(SCoder *decoder, SMqCMResetOffsetReq *pReq) {
if (tStartDecode(decoder) < 0) return -1;
if (tDecodeI32(decoder, &pReq->num) < 0) return -1;
pReq->offsets = TCODER_MALLOC(pReq->num * sizeof(SMqOffset), decoder);
if (pReq->offsets == NULL) return -1;
for (int32_t i = 0; i < pReq->num; i++) {
tDecodeSMqOffset(decoder, &pReq->offsets[i]);
}
tEndDecode(decoder);
return 0;
}
#if 0
int32_t tEncodeSMqMVResetOffsetReq(SCoder *encoder, const SMqMVResetOffsetReq *pReq) {
if (tEncodeI32(encoder, pReq->num) < 0) return -1;
if (tEncodeI64(encoder, pReq->leftForVer) < 0) return -1;
for (int32_t i = 0; i < pReq->num; i++) {
tEncodeSMqOffset(encoder, &pReq->offsets[i]);
}
@ -2385,3 +2424,4 @@ int32_t tDecodeSMqMVResetOffsetReq(SCoder *decoder, SMqMVResetOffsetReq *pReq) {
}
return 0;
}
#endif

1
source/dnode/mgmt/impl/src/dndVnodes.c
View File

@ -507,6 +507,7 @@ static void dndGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
pCfg->isHeapAllocator = true;
pCfg->ttl = 4;
pCfg->keep = pCreate->daysToKeep0;
pCfg->streamMode = pCreate->streamMode;
pCfg->isWeak = true;
pCfg->tsdbCfg.keep = pCreate->daysToKeep0;
pCfg->tsdbCfg.keep1 = pCreate->daysToKeep2;

16
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -300,6 +300,7 @@ typedef struct {
int8_t quorum;
int8_t update;
int8_t cacheLastRow;
int8_t streamMode;
} SDbCfg;
typedef struct {
@ -336,6 +337,7 @@ typedef struct {
int64_t pointsWritten;
int8_t compact;
int8_t replica;
int8_t streamMode;
SVnodeGid vnodeGid[TSDB_MAX_REPLICA];
} SVgObj;
@ -467,9 +469,9 @@ typedef struct {
char key[TSDB_SUBSCRIBE_KEY_LEN];
int32_t status;
int32_t vgNum;
SArray* consumers; // SArray<SMqSubConsumer>
SArray* lostConsumers; // SArray<SMqSubConsumer>
SArray* unassignedVg; // SArray<SMqConsumerEp>
SArray* consumers; // SArray<SMqSubConsumer>
SArray* lostConsumers; // SArray<SMqSubConsumer>
SArray* unassignedVg; // SArray<SMqConsumerEp>
} SMqSubscribeObj;
static FORCE_INLINE SMqSubscribeObj* tNewSubscribeObj() {
@ -583,13 +585,13 @@ static FORCE_INLINE void* tDecodeSubscribeObj(void* buf, SMqSubscribeObj* pSub)
static FORCE_INLINE void tDeleteSMqSubscribeObj(SMqSubscribeObj* pSub) {
if (pSub->consumers) {
taosArrayDestroyEx(pSub->consumers, (void (*)(void*))tDeleteSMqSubConsumer);
//taosArrayDestroy(pSub->consumers);
// taosArrayDestroy(pSub->consumers);
pSub->consumers = NULL;
}
if (pSub->unassignedVg) {
taosArrayDestroyEx(pSub->unassignedVg, (void (*)(void*))tDeleteSMqConsumerEp);
//taosArrayDestroy(pSub->unassignedVg);
// taosArrayDestroy(pSub->unassignedVg);
pSub->unassignedVg = NULL;
}
}
@ -614,8 +616,8 @@ typedef struct {
int64_t connId;
SRWLatch lock;
char cgroup[TSDB_CONSUMER_GROUP_LEN];
SArray* currentTopics; // SArray<char*>
SArray* recentRemovedTopics; // SArray<char*>
SArray* currentTopics; // SArray<char*>
SArray* recentRemovedTopics; // SArray<char*>
int32_t epoch;
// stat
int64_t pollCnt;

41
source/dnode/mnode/impl/src/mndDb.c
View File

@ -395,24 +395,27 @@ static int32_t mndCreateDb(SMnode *pMnode, SMnodeMsg *pReq, SCreateDbReq *pCreat
dbObj.vgVersion = 1;
dbObj.hashMethod = 1;
memcpy(dbObj.createUser, pUser->user, TSDB_USER_LEN);
dbObj.cfg = (SDbCfg){.numOfVgroups = pCreate->numOfVgroups,
.cacheBlockSize = pCreate->cacheBlockSize,
.totalBlocks = pCreate->totalBlocks,
.daysPerFile = pCreate->daysPerFile,
.daysToKeep0 = pCreate->daysToKeep0,
.daysToKeep1 = pCreate->daysToKeep1,
.daysToKeep2 = pCreate->daysToKeep2,
.minRows = pCreate->minRows,
.maxRows = pCreate->maxRows,
.fsyncPeriod = pCreate->fsyncPeriod,
.commitTime = pCreate->commitTime,
.precision = pCreate->precision,
.compression = pCreate->compression,
.walLevel = pCreate->walLevel,
.replications = pCreate->replications,
.quorum = pCreate->quorum,
.update = pCreate->update,
.cacheLastRow = pCreate->cacheLastRow};
dbObj.cfg = (SDbCfg){
.numOfVgroups = pCreate->numOfVgroups,
.cacheBlockSize = pCreate->cacheBlockSize,
.totalBlocks = pCreate->totalBlocks,
.daysPerFile = pCreate->daysPerFile,
.daysToKeep0 = pCreate->daysToKeep0,
.daysToKeep1 = pCreate->daysToKeep1,
.daysToKeep2 = pCreate->daysToKeep2,
.minRows = pCreate->minRows,
.maxRows = pCreate->maxRows,
.fsyncPeriod = pCreate->fsyncPeriod,
.commitTime = pCreate->commitTime,
.precision = pCreate->precision,
.compression = pCreate->compression,
.walLevel = pCreate->walLevel,
.replications = pCreate->replications,
.quorum = pCreate->quorum,
.update = pCreate->update,
.cacheLastRow = pCreate->cacheLastRow,
.streamMode = pCreate->streamMode,
};
mndSetDefaultDbCfg(&dbObj.cfg);
@ -1400,4 +1403,4 @@ static int32_t mndRetrieveDbs(SMnodeMsg *pReq, SShowObj *pShow, char *data, int3
static void mndCancelGetNextDb(SMnode *pMnode, void *pIter) {
SSdb *pSdb = pMnode->pSdb;
sdbCancelFetch(pSdb, pIter);
}
}

42
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -53,6 +53,7 @@ static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pMsg);
static int32_t mndProcessMqTimerMsg(SMnodeMsg *pMsg);
static int32_t mndProcessGetSubEpReq(SMnodeMsg *pMsg);
static int32_t mndProcessDoRebalanceMsg(SMnodeMsg *pMsg);
static int32_t mndProcessResetOffsetReq(SMnodeMsg *pMsg);
static int32_t mndPersistMqSetConnReq(SMnode *pMnode, STrans *pTrans, const SMqTopicObj *pTopic, const char *cgroup,
const SMqConsumerEp *pConsumerEp);
@ -205,6 +206,45 @@ static int32_t mndPersistCancelConnReq(SMnode *pMnode, STrans *pTrans, const SMq
return 0;
}
#if 0
static int32_t mndProcessResetOffsetReq(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
uint8_t *str = pMsg->rpcMsg.pCont;
SMqCMResetOffsetReq req;
SCoder decoder;
tCoderInit(&decoder, TD_LITTLE_ENDIAN, str, pMsg->rpcMsg.contLen, TD_DECODER);
tDecodeSMqCMResetOffsetReq(&decoder, &req);
SHashObj *pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
if (pHash == NULL) {
return -1;
}
for (int32_t i = 0; i < req.num; i++) {
SMqOffset *pOffset = &req.offsets[i];
SMqVgOffsets *pVgOffset = taosHashGet(pHash, &pOffset->vgId, sizeof(int32_t));
if (pVgOffset == NULL) {
pVgOffset = malloc(sizeof(SMqVgOffsets));
if (pVgOffset == NULL) {
return -1;
}
pVgOffset->offsets = taosArrayInit(0, sizeof(void *));
taosArrayPush(pVgOffset->offsets, &pOffset);
}
taosHashPut(pHash, &pOffset->vgId, sizeof(int32_t), &pVgOffset, sizeof(void *));
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("mq-reset-offset: failed since %s", terrstr());
return -1;
}
return 0;
}
#endif
static int32_t mndProcessGetSubEpReq(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->rpcMsg.pCont;
@ -1059,6 +1099,8 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
pConsumerEp->consumerId = consumerId;
taosArrayPush(mqSubConsumer.vgInfo, pConsumerEp);
if (pConsumerEp->oldConsumerId == -1) {
mInfo("mq set conn: assign vgroup %d of topic %s to consumer %ld", pConsumerEp->vgId, newTopicName,
pConsumerEp->consumerId);
mndPersistMqSetConnReq(pMnode, pTrans, pTopic, cgroup, pConsumerEp);
} else {
mndPersistRebalanceMsg(pMnode, pTrans, pConsumerEp);

9
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -21,7 +21,7 @@
#include "mndShow.h"
#include "mndTrans.h"
#define TSDB_VGROUP_VER_NUMBER 1
#define TSDB_VGROUP_VER_NUMBER 1
#define TSDB_VGROUP_RESERVE_SIZE 64
static SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw);
@ -214,6 +214,7 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
createReq.cacheLastRow = pDb->cfg.cacheLastRow;
createReq.replica = pVgroup->replica;
createReq.selfIndex = -1;
createReq.streamMode = pVgroup->streamMode;
for (int32_t v = 0; v < pVgroup->replica; ++v) {
SReplica *pReplica = &createReq.replicas[v];
@ -255,8 +256,7 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
return pReq;
}
void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup,
int32_t *pContLen) {
void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
SDropVnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
dropReq.vgId = pVgroup->vgId;
@ -399,6 +399,7 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
pVgroup->createdTime = taosGetTimestampMs();
pVgroup->updateTime = pVgroups->createdTime;
pVgroup->version = 1;
pVgroup->streamMode = pDb->cfg.streamMode;
pVgroup->hashBegin = hashMin + hashInterval * v;
if (v == pDb->cfg.numOfVgroups - 1) {
pVgroup->hashEnd = hashMax;
@ -700,4 +701,4 @@ static int32_t mndRetrieveVnodes(SMnodeMsg *pReq, SShowObj *pShow, char *data, i
static void mndCancelGetNextVnode(SMnode *pMnode, void *pIter) {
SSdb *pSdb = pMnode->pSdb;
sdbCancelFetch(pSdb, pIter);
}
}

24
source/dnode/vnode/CMakeLists.txt
View File

@ -1,7 +1,24 @@
set(META_DB_IMPL_LIST "BDB" "TDB")
set(META_DB_IMPL "BDB" CACHE STRING "Use BDB as the default META implementation")
set_property(CACHE META_DB_IMPL PROPERTY STRINGS ${META_DB_IMPL_LIST})
if(META_DB_IMPL IN_LIST META_DB_IMPL_LIST)
message(STATUS "META DB Impl: ${META_DB_IMPL}==============")
else()
message(FATAL_ERROR "Invalid META DB IMPL: ${META_DB_IMPL}==============")
endif()
aux_source_directory(src/meta META_SRC)
if(${META_DB_IMPL} STREQUAL "BDB")
list(REMOVE_ITEM META_SRC "src/meta/metaTDBImpl.c")
elseif(${META_DB_IMPL} STREQUAL "TDB")
list(REMOVE_ITEM META_SRC "src/meta/metaBDBImpl.c")
endif()
aux_source_directory(src/tq TQ_SRC)
aux_source_directory(src/tsdb TSDB_SRC)
aux_source_directory(src/vnd VND_SRC)
list(APPEND
VNODE_SRC
${META_SRC}
@ -22,7 +39,6 @@ target_link_libraries(
PUBLIC util
PUBLIC common
PUBLIC transport
PUBLIC bdb
PUBLIC tfs
PUBLIC wal
PUBLIC scheduler
@ -31,6 +47,12 @@ target_link_libraries(
PUBLIC sync
)
if(${META_DB_IMPL} STREQUAL "BDB")
target_link_libraries(vnode PUBLIC bdb)
elseif(${META_DB_IMPL} STREQUAL "TDB")
target_link_libraries(vnode PUBLIC tdb)
endif()
if(${BUILD_TEST})
# add_subdirectory(test)
endif(${BUILD_TEST})

7
source/dnode/vnode/inc/vnode.h
View File

@ -51,6 +51,7 @@ typedef struct {
bool isHeapAllocator;
uint32_t ttl;
uint32_t keep;
int8_t streamMode;
bool isWeak;
STsdbCfg tsdbCfg;
SMetaCfg metaCfg;
@ -225,9 +226,9 @@ static FORCE_INLINE int tqReadHandleSetTbUidList(STqReadHandle *pHandle, const S
return 0;
}
void tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReadHandle *pHandle);
int tqRetrieveDataBlockInfo(STqReadHandle *pHandle, SDataBlockInfo *pBlockInfo);
int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReadHandle *pHandle);
int tqRetrieveDataBlockInfo(STqReadHandle *pHandle, SDataBlockInfo *pBlockInfo);
// return SArray<SColumnInfoData>
SArray *tqRetrieveDataBlock(STqReadHandle *pHandle);

3
source/dnode/vnode/src/inc/vnd.h
View File

@ -23,8 +23,8 @@
#include "tlist.h"
#include "tlockfree.h"
#include "tmacro.h"
#include "wal.h"
#include "tq.h"
#include "wal.h"
#include "vnode.h"
@ -175,7 +175,6 @@ void* vmaMalloc(SVMemAllocator* pVMA, uint64_t size);
void vmaFree(SVMemAllocator* pVMA, void* ptr);
bool vmaIsFull(SVMemAllocator* pVMA);
#ifdef __cplusplus
}
#endif

145
source/dnode/vnode/src/meta/metaTDBImpl.c Normal file
View File

@ -0,0 +1,145 @@
/*
* 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 "metaDef.h"
#include "tdb.h"
struct SMetaDB {
TENV *pEnv;
TDB * pTbDB;
TDB * pSchemaDB;
TDB * pNameIdx;
TDB * pStbIdx;
TDB * pNtbIdx;
TDB * pCtbIdx;
// tag index hash table
// suid+colid --> TDB *
struct {
} tagIdxHt;
};
#define A(op, flag) \
do { \
if ((ret = op) != 0) goto flag; \
} while (0)
int metaOpenDB(SMeta *pMeta) {
SMetaDB *pDb;
TENV * pEnv;
TDB * pTbDB;
TDB * pSchemaDB;
TDB * pNameIdx;
TDB * pStbIdx;
TDB * pNtbIdx;
TDB * pCtbIdx;
int ret;
pDb = (SMetaDB *)calloc(1, sizeof(*pDb));
if (pDb == NULL) {
return -1;
}
// Create and open the ENV
A((tdbEnvCreate(&pEnv)), _err);
#if 0
// Set options of the environment
A(tdbEnvSetPageSize(pEnv, 8192), _err);
A(tdbEnvSetCacheSize(pEnv, 16 * 1024 * 1024), _err);
#endif
A((tdbEnvOpen(&pEnv)), _err);
// Create and open each DB
A(tdbCreate(&pTbDB), _err);
A(tdbOpen(&pTbDB, "table.db", NULL, pEnv), _err);
A(tdbCreate(&pSchemaDB), _err);
A(tdbOpen(&pSchemaDB, "schema.db", NULL, pEnv), _err);
A(tdbCreate(&pNameIdx), _err);
A(tdbOpen(&pNameIdx, "name.db", NULL, pEnv), _err);
// tdbAssociate();
pDb->pEnv = pEnv;
pDb->pTbDB = pTbDB;
pDb->pSchemaDB = pSchemaDB;
pMeta->pDB = pDb;
return 0;
_err:
return -1;
}
void metaCloseDB(SMeta *pMeta) {
// TODO
}
int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
// TODO
return 0;
}
int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
// TODO
return 0;
}
STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
// TODO
return NULL;
}
STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid) {
// TODO
return NULL;
}
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
// TODO
return NULL;
}
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
// TODO
return NULL;
}
SMTbCursor *metaOpenTbCursor(SMeta *pMeta) {
// TODO
return NULL;
}
void metaCloseTbCursor(SMTbCursor *pTbCur) {
// TODO
}
char *metaTbCursorNext(SMTbCursor *pTbCur) {
// TODO
return NULL;
}
SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) {
// TODO
return NULL;
}
void metaCloseCtbCurosr(SMCtbCursor *pCtbCur) {
// TODO
}
tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {
// TODO
return 0;
}

33
source/dnode/vnode/src/tq/tqRead.c
View File

@ -31,13 +31,28 @@ STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
return pReadHandle;
}
void tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitReq* pMsg, int64_t ver) {
int32_t tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitReq* pMsg, int64_t ver) {
pReadHandle->pMsg = pMsg;
pMsg->length = htonl(pMsg->length);
pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter);
if (tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter) < 0) return -1;
while (true) {
if (tGetSubmitMsgNext(&pReadHandle->msgIter, &pReadHandle->pBlock) < 0) return -1;
if (pReadHandle->pBlock == NULL) break;
pReadHandle->pBlock->uid = htobe64(pReadHandle->pBlock->uid);
pReadHandle->pBlock->tid = htonl(pReadHandle->pBlock->tid);
pReadHandle->pBlock->sversion = htonl(pReadHandle->pBlock->sversion);
pReadHandle->pBlock->dataLen = htonl(pReadHandle->pBlock->dataLen);
pReadHandle->pBlock->schemaLen = htonl(pReadHandle->pBlock->schemaLen);
pReadHandle->pBlock->numOfRows = htons(pReadHandle->pBlock->numOfRows);
}
if (tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter) < 0) return -1;
pReadHandle->ver = ver;
memset(&pReadHandle->blkIter, 0, sizeof(SSubmitBlkIter));
return 0;
}
bool tqNextDataBlock(STqReadHandle* pHandle) {
@ -47,19 +62,19 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
}
if (pHandle->pBlock == NULL) return false;
pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);
/*pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);*/
/*if (pHandle->tbUid == pHandle->pBlock->uid) {*/
ASSERT(pHandle->tbIdHash);
void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->pBlock->uid, sizeof(int64_t));
if (ret != NULL) {
/*printf("retrieve one tb %ld\n", pHandle->pBlock->uid);*/
pHandle->pBlock->tid = htonl(pHandle->pBlock->tid);
pHandle->pBlock->sversion = htonl(pHandle->pBlock->sversion);
pHandle->pBlock->dataLen = htonl(pHandle->pBlock->dataLen);
pHandle->pBlock->schemaLen = htonl(pHandle->pBlock->schemaLen);
pHandle->pBlock->numOfRows = htons(pHandle->pBlock->numOfRows);
/*pHandle->pBlock->tid = htonl(pHandle->pBlock->tid);*/
/*pHandle->pBlock->sversion = htonl(pHandle->pBlock->sversion);*/
/*pHandle->pBlock->dataLen = htonl(pHandle->pBlock->dataLen);*/
/*pHandle->pBlock->schemaLen = htonl(pHandle->pBlock->schemaLen);*/
/*pHandle->pBlock->numOfRows = htons(pHandle->pBlock->numOfRows);*/
return true;
} else {
/*} else {*/
/*printf("skip one tb %ld\n", pHandle->pBlock->uid);*/
}
}

26
source/dnode/vnode/src/vnd/vnodeWrite.c
View File

@ -43,13 +43,17 @@ int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
SVCreateTbReq vCreateTbReq;
SVCreateTbBatchReq vCreateTbBatchReq;
void *ptr = vnodeMalloc(pVnode, pMsg->contLen);
if (ptr == NULL) {
// TODO: handle error
}
void *ptr = NULL;
// TODO: copy here need to be extended
memcpy(ptr, pMsg->pCont, pMsg->contLen);
if (pVnode->config.streamMode == 0) {
ptr = vnodeMalloc(pVnode, pMsg->contLen);
if (ptr == NULL) {
// TODO: handle error
}
// TODO: copy here need to be extended
memcpy(ptr, pMsg->pCont, pMsg->contLen);
}
// todo: change the interface here
int64_t ver;
@ -109,17 +113,19 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
// }
break;
case TDMT_VND_SUBMIT:
if (tsdbInsertData(pVnode->pTsdb, (SSubmitReq *)ptr, NULL) < 0) {
// TODO: handle error
if (pVnode->config.streamMode == 0) {
if (tsdbInsertData(pVnode->pTsdb, (SSubmitReq *)ptr, NULL) < 0) {
// TODO: handle error
}
}
break;
case TDMT_VND_MQ_SET_CONN: {
if (tqProcessSetConnReq(pVnode->pTq, POINTER_SHIFT(ptr, sizeof(SMsgHead))) < 0) {
if (tqProcessSetConnReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead))) < 0) {
// TODO: handle error
}
} break;
case TDMT_VND_MQ_REB: {
if (tqProcessRebReq(pVnode->pTq, POINTER_SHIFT(ptr, sizeof(SMsgHead))) < 0) {
if (tqProcessRebReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead))) < 0) {
}
} break;
default:

11
source/libs/executor/src/executor.c
View File

@ -14,9 +14,9 @@
*/
#include "executor.h"
#include "tq.h"
#include "executorimpl.h"
#include "planner.h"
#include "tq.h"
static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id) {
ASSERT(pOperator != NULL);
@ -34,7 +34,10 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id)
return doSetStreamBlock(pOperator->pDownstream[0], input, id);
} else {
SStreamBlockScanInfo* pInfo = pOperator->info;
tqReadHandleSetMsg(pInfo->readerHandle, input, 0);
if (tqReadHandleSetMsg(pInfo->readerHandle, input, 0) < 0) {
qError("submit msg error while set stream msg, %s" PRIx64, id);
return TSDB_CODE_QRY_APP_ERROR;
}
return TSDB_CODE_SUCCESS;
}
}
@ -48,9 +51,9 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input) {
return TSDB_CODE_SUCCESS;
}
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo;
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void*) input, GET_TASKID(pTaskInfo));
int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void*)input, GET_TASKID(pTaskInfo));
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
} else {

2
source/libs/nodes/CMakeLists.txt
View File

@ -7,7 +7,7 @@ target_include_directories(
)
target_link_libraries(
nodes
PRIVATE os util
PRIVATE os util common qcom
)
if(${BUILD_TEST})

231
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -13,10 +13,179 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "nodes.h"
#include "plannodes.h"
#include "querynodes.h"
#include "query.h"
#include "taoserror.h"
#include "tjson.h"
int32_t nodesNodeToString(const SNode* pNode, char** pStr, int32_t* pLen) {
switch (nodeType(pNode)) {
static int32_t nodeToJson(const void* pObj, SJson* pJson);
static char* nodeName(ENodeType type) {
switch (type) {
case QUERY_NODE_COLUMN:
return "Column";
case QUERY_NODE_VALUE:
case QUERY_NODE_OPERATOR:
case QUERY_NODE_LOGIC_CONDITION:
case QUERY_NODE_FUNCTION:
case QUERY_NODE_REAL_TABLE:
case QUERY_NODE_TEMP_TABLE:
case QUERY_NODE_JOIN_TABLE:
case QUERY_NODE_GROUPING_SET:
case QUERY_NODE_ORDER_BY_EXPR:
case QUERY_NODE_LIMIT:
case QUERY_NODE_STATE_WINDOW:
case QUERY_NODE_SESSION_WINDOW:
case QUERY_NODE_INTERVAL_WINDOW:
case QUERY_NODE_NODE_LIST:
case QUERY_NODE_FILL:
case QUERY_NODE_COLUMN_REF:
case QUERY_NODE_TARGET:
case QUERY_NODE_RAW_EXPR:
case QUERY_NODE_SET_OPERATOR:
case QUERY_NODE_SELECT_STMT:
case QUERY_NODE_SHOW_STMT:
break;
case QUERY_NODE_LOGIC_PLAN_SCAN:
return "LogicScan";
case QUERY_NODE_LOGIC_PLAN_JOIN:
return "LogicJoin";
case QUERY_NODE_LOGIC_PLAN_FILTER:
return "LogicFilter";
case QUERY_NODE_LOGIC_PLAN_AGG:
return "LogicAgg";
case QUERY_NODE_LOGIC_PLAN_PROJECT:
return "LogicProject";
default:
break;
}
return "Unknown";
}
static int32_t addNodeList(SJson* pJson, const char* pName, FToJson func, const SNodeList* pList) {
if (LIST_LENGTH(pList) > 0) {
SJson* jList = tjsonAddArrayToObject(pJson, pName);
if (NULL == jList) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SNode* pNode;
FOREACH(pNode, pList) {
int32_t code = tjsonAddItem(jList, func, pNode);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
}
}
return TSDB_CODE_SUCCESS;
}
static const char* jkTableMetaUid = "TableMetaUid";
static const char* jkTableMetaSuid = "TableMetaSuid";
static int32_t tableMetaToJson(const void* pObj, SJson* pJson) {
const STableMeta* pNode = (const STableMeta*)pObj;
int32_t code = tjsonAddIntegerToObject(pJson, jkTableMetaUid, pNode->uid);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkTableMetaSuid, pNode->suid);
}
return code;
}
static const char* jkLogicPlanId = "Id";
static const char* jkLogicPlanTargets = "Targets";
static const char* jkLogicPlanConditions = "Conditions";
static const char* jkLogicPlanChildren = "Children";
static int32_t logicPlanNodeToJson(const void* pObj, SJson* pJson) {
const SLogicNode* pNode = (const SLogicNode*)pObj;
int32_t code = tjsonAddIntegerToObject(pJson, jkLogicPlanId, pNode->id);
if (TSDB_CODE_SUCCESS == code) {
code = addNodeList(pJson, jkLogicPlanTargets, nodeToJson, pNode->pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkLogicPlanConditions, nodeToJson, pNode->pConditions);
}
if (TSDB_CODE_SUCCESS == code) {
code = addNodeList(pJson, jkLogicPlanChildren, nodeToJson, pNode->pChildren);
}
return code;
}
static const char* jkScanLogicPlanScanCols = "ScanCols";
static const char* jkScanLogicPlanTableMeta = "TableMeta";
static int32_t logicScanToJson(const void* pObj, SJson* pJson) {
const SScanLogicNode* pNode = (const SScanLogicNode*)pObj;
int32_t code = logicPlanNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = addNodeList(pJson, jkScanLogicPlanScanCols, nodeToJson, pNode->pScanCols);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkScanLogicPlanTableMeta, tableMetaToJson, pNode->pMeta);
}
return code;
}
static const char* jkProjectLogicPlanProjections = "Projections";
static int32_t logicProjectToJson(const void* pObj, SJson* pJson) {
const SProjectLogicNode* pNode = (const SProjectLogicNode*)pObj;
int32_t code = logicPlanNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = addNodeList(pJson, jkProjectLogicPlanProjections, nodeToJson, pNode->pProjections);
}
return code;
}
static const char* jkJoinLogicPlanJoinType = "JoinType";
static const char* jkJoinLogicPlanOnConditions = "OnConditions";
static int32_t logicJoinToJson(const void* pObj, SJson* pJson) {
const SJoinLogicNode* pNode = (const SJoinLogicNode*)pObj;
int32_t code = logicPlanNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinLogicPlanJoinType, pNode->joinType);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkJoinLogicPlanOnConditions, nodeToJson, pNode->pOnConditions);
}
return code;
}
static int32_t logicFilterToJson(const void* pObj, SJson* pJson) {
return logicPlanNodeToJson(pObj, pJson);
}
static const char* jkAggLogicPlanGroupKeys = "GroupKeys";
static const char* jkAggLogicPlanAggFuncs = "AggFuncs";
static int32_t logicAggToJson(const void* pObj, SJson* pJson) {
const SAggLogicNode* pNode = (const SAggLogicNode*)pObj;
int32_t code = logicPlanNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = addNodeList(pJson, jkAggLogicPlanGroupKeys, nodeToJson, pNode->pGroupKeys);
}
if (TSDB_CODE_SUCCESS == code) {
code = addNodeList(pJson, jkAggLogicPlanAggFuncs, nodeToJson, pNode->pAggFuncs);
}
return code;
}
static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
switch (nodeType(pObj)) {
case QUERY_NODE_COLUMN:
case QUERY_NODE_VALUE:
case QUERY_NODE_OPERATOR:
@ -31,14 +200,68 @@ int32_t nodesNodeToString(const SNode* pNode, char** pStr, int32_t* pLen) {
case QUERY_NODE_STATE_WINDOW:
case QUERY_NODE_SESSION_WINDOW:
case QUERY_NODE_INTERVAL_WINDOW:
case QUERY_NODE_NODE_LIST:
case QUERY_NODE_FILL:
case QUERY_NODE_COLUMN_REF:
case QUERY_NODE_TARGET:
case QUERY_NODE_RAW_EXPR:
case QUERY_NODE_SET_OPERATOR:
case QUERY_NODE_SELECT_STMT:
case QUERY_NODE_SHOW_STMT:
break;
case QUERY_NODE_LOGIC_PLAN_SCAN:
return logicScanToJson(pObj, pJson);
case QUERY_NODE_LOGIC_PLAN_JOIN:
return logicJoinToJson(pObj, pJson);
case QUERY_NODE_LOGIC_PLAN_FILTER:
return logicFilterToJson(pObj, pJson);
case QUERY_NODE_LOGIC_PLAN_AGG:
return logicAggToJson(pObj, pJson);
case QUERY_NODE_LOGIC_PLAN_PROJECT:
return logicProjectToJson(pObj, pJson);
default:
break;
}
return TSDB_CODE_SUCCESS;
}
static const char* jkNodeType = "Type";
static int32_t nodeToJson(const void* pObj, SJson* pJson) {
const SNode* pNode = (const SNode*)pObj;
char* pNodeName = nodeName(nodeType(pNode));
int32_t code = tjsonAddStringToObject(pJson, jkNodeType, pNodeName);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, pNodeName, specificNodeToJson, pNode);
}
return code;
}
int32_t nodesNodeToString(const SNode* pNode, char** pStr, int32_t* pLen) {
if (NULL == pNode || NULL == pStr || NULL == pLen) {
return TSDB_CODE_SUCCESS;
}
SJson* pJson = tjsonCreateObject();
if (NULL == pJson) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t code = nodeToJson(pNode, pJson);
if (TSDB_CODE_SUCCESS != code) {
terrno = code;
return code;
}
*pStr = tjsonToString(pJson);
tjsonDelete(pJson);
*pLen = strlen(*pStr) + 1;
return TSDB_CODE_SUCCESS;
}
int32_t nodesStringToNode(const char* pStr, SNode** pNode) {
return TSDB_CODE_SUCCESS;
}

56
source/libs/nodes/src/nodesUtilFuncs.c
View File

@ -14,7 +14,7 @@
*/
#include "querynodes.h"
#include "nodesShowStmts.h"
#include "plannodes.h"
#include "taos.h"
#include "taoserror.h"
#include "thash.h"
@ -68,8 +68,18 @@ SNode* nodesMakeNode(ENodeType type) {
return makeNode(type, sizeof(SSetOperator));
case QUERY_NODE_SELECT_STMT:
return makeNode(type, sizeof(SSelectStmt));
case QUERY_NODE_SHOW_STMT:
return makeNode(type, sizeof(SShowStmt));
// case QUERY_NODE_SHOW_STMT:
// return makeNode(type, sizeof(SShowStmt));
case QUERY_NODE_LOGIC_PLAN_SCAN:
return makeNode(type, sizeof(SScanLogicNode));
case QUERY_NODE_LOGIC_PLAN_JOIN:
return makeNode(type, sizeof(SJoinLogicNode));
case QUERY_NODE_LOGIC_PLAN_FILTER:
return makeNode(type, sizeof(SFilterLogicNode));
case QUERY_NODE_LOGIC_PLAN_AGG:
return makeNode(type, sizeof(SAggLogicNode));
case QUERY_NODE_LOGIC_PLAN_PROJECT:
return makeNode(type, sizeof(SProjectLogicNode));
default:
break;
}
@ -121,6 +131,15 @@ int32_t nodesListAppend(SNodeList* pList, SNode* pNode) {
return TSDB_CODE_SUCCESS;
}
int32_t nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc) {
pTarget->pTail->pNext = pSrc->pHead;
pSrc->pHead->pPrev = pTarget->pTail;
pTarget->pTail = pSrc->pTail;
pTarget->length += pSrc->length;
tfree(pSrc);
return TSDB_CODE_SUCCESS;
}
SListCell* nodesListErase(SNodeList* pList, SListCell* pCell) {
if (NULL == pCell->pPrev) {
pList->pHead = pCell->pNext;
@ -129,6 +148,7 @@ SListCell* nodesListErase(SNodeList* pList, SListCell* pCell) {
pCell->pNext->pPrev = pCell->pPrev;
}
SListCell* pNext = pCell->pNext;
nodesDestroyNode(pCell->pNode);
tfree(pCell);
--(pList->length);
return pNext;
@ -185,6 +205,14 @@ bool nodesIsComparisonOp(const SOperatorNode* pOp) {
case OP_TYPE_NOT_LIKE:
case OP_TYPE_MATCH:
case OP_TYPE_NMATCH:
case OP_TYPE_IS_NULL:
case OP_TYPE_IS_NOT_NULL:
case OP_TYPE_IS_TRUE:
case OP_TYPE_IS_FALSE:
case OP_TYPE_IS_UNKNOWN:
case OP_TYPE_IS_NOT_TRUE:
case OP_TYPE_IS_NOT_FALSE:
case OP_TYPE_IS_NOT_UNKNOWN:
return true;
default:
break;
@ -213,8 +241,7 @@ bool nodesIsTimelineQuery(const SNode* pQuery) {
typedef struct SCollectColumnsCxt {
int32_t errCode;
uint64_t tableId;
bool realCol;
const char* pTableAlias;
SNodeList* pCols;
SHashObj* pColIdHash;
} SCollectColumnsCxt;
@ -232,27 +259,24 @@ static EDealRes doCollect(SCollectColumnsCxt* pCxt, int32_t id, SNode* pNode) {
static EDealRes collectColumns(SNode* pNode, void* pContext) {
SCollectColumnsCxt* pCxt = (SCollectColumnsCxt*)pContext;
if (pCxt->realCol && QUERY_NODE_COLUMN == nodeType(pNode)) {
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SColumnNode* pCol = (SColumnNode*)pNode;
int32_t colId = pCol->colId;
if (pCxt->tableId == pCol->tableId && colId > 0) {
if (0 == strcmp(pCxt->pTableAlias, pCol->tableAlias)) {
return doCollect(pCxt, colId, pNode);
}
} else if (!pCxt->realCol && QUERY_NODE_COLUMN_REF == nodeType(pNode)) {
return doCollect(pCxt, ((SColumnRefNode*)pNode)->slotId, pNode);
}
return DEAL_RES_CONTINUE;
}
int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, uint64_t tableId, bool realCol, SNodeList** pCols) {
int32_t nodesCollectColumns(SSelectStmt* pSelect, ESqlClause clause, const char* pTableAlias, SNodeList** pCols) {
if (NULL == pSelect || NULL == pCols) {
return TSDB_CODE_SUCCESS;
}
SCollectColumnsCxt cxt = {
.errCode = TSDB_CODE_SUCCESS,
.realCol = realCol,
.pTableAlias = pTableAlias,
.pCols = nodesMakeList(),
.pColIdHash = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK)
};
@ -303,6 +327,12 @@ int32_t nodesCollectFuncs(SSelectStmt* pSelect, FFuncClassifier classifier, SNod
nodesDestroyList(cxt.pFuncs);
return cxt.errCode;
}
*pFuncs = cxt.pFuncs;
if (LIST_LENGTH(cxt.pFuncs) > 0) {
*pFuncs = cxt.pFuncs;
} else {
nodesDestroyList(cxt.pFuncs);
*pFuncs = NULL;
}
return TSDB_CODE_SUCCESS;
}

1
source/libs/parser/inc/astGenerator.h
View File

@ -170,6 +170,7 @@ typedef struct SCreateDbInfo {
int8_t update;
int8_t cachelast;
SArray *keep;
int8_t streamMode;
} SCreateDbInfo;
typedef struct SCreateFuncInfo {

20
source/libs/parser/inc/parserImpl.h
View File

@ -13,27 +13,15 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_AST_CREATE_FUNCS_H_
#define _TD_AST_CREATE_FUNCS_H_
#ifndef _TD_PARSER_IMPL_H_
#define _TD_PARSER_IMPL_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "querynodes.h"
#include "parser.h"
typedef enum EStmtType {
STMT_TYPE_CMD = 1,
STMT_TYPE_QUERY
} EStmtType;
typedef struct SQuery {
EStmtType stmtType;
SNode* pRoot;
int32_t numOfResCols;
SSchema* pResSchema;
} SQuery;
#include "newParser.h"
int32_t doParse(SParseContext* pParseCxt, SQuery* pQuery);
int32_t doTranslate(SParseContext* pParseCxt, SQuery* pQuery);
@ -42,4 +30,4 @@ int32_t doTranslate(SParseContext* pParseCxt, SQuery* pQuery);
}
#endif
#endif /*_TD_AST_CREATE_FUNCS_H_*/
#endif /*_TD_PARSER_IMPL_H_*/

2
source/libs/parser/inc/sql.y
View File

@ -282,6 +282,7 @@ update(Y) ::= UPDATE INTEGER(X). { Y = X; }
cachelast(Y) ::= CACHELAST INTEGER(X). { Y = X; }
vgroups(Y) ::= VGROUPS INTEGER(X). { Y = X; }
//partitions(Y) ::= PARTITIONS INTEGER(X). { Y = X; }
stream_mode(Y) ::= STREAM MODE INTEGER(X). { Y = X; }
%type db_optr {SCreateDbInfo}
db_optr(Y) ::= . {setDefaultCreateDbOption(&Y);}
@ -302,6 +303,7 @@ db_optr(Y) ::= db_optr(Z) keep(X). { Y = Z; Y.keep = X; }
db_optr(Y) ::= db_optr(Z) update(X). { Y = Z; Y.update = strtol(X.z, NULL, 10); }
db_optr(Y) ::= db_optr(Z) cachelast(X). { Y = Z; Y.cachelast = strtol(X.z, NULL, 10); }
db_optr(Y) ::= db_optr(Z) vgroups(X). { Y = Z; Y.numOfVgroups = strtol(X.z, NULL, 10); }
db_optr(Y) ::= db_optr(Z) stream_mode(X). { Y = Z; Y.streamMode = strtol(X.z, NULL, 10); }
//%type topic_optr {SCreateDbInfo}
//

1
source/libs/parser/src/astToMsg.c
View File

@ -242,6 +242,7 @@ static void doSetDbOptions(SCreateDbReq* pMsg, const SCreateDbInfo* pCreateDb) {
pMsg->update = pCreateDb->update;
pMsg->cacheLastRow = pCreateDb->cachelast;
pMsg->numOfVgroups = pCreateDb->numOfVgroups;
pMsg->streamMode = pCreateDb->streamMode;
}
int32_t setDbOptions(SCreateDbReq* pCreateDbMsg, const SCreateDbInfo* pCreateDbSql, SMsgBuf* pMsgBuf) {

22
source/libs/parser/src/parserImpl.c
View File

@ -349,14 +349,15 @@ static SNodeList* getProjectList(SNode* pNode) {
return NULL;
}
static void setColumnInfoBySchema(const STableNode* pTable, const SSchema* pColSchema, SColumnNode* pCol) {
strcpy(pCol->dbName, pTable->dbName);
strcpy(pCol->tableAlias, pTable->tableAlias);
strcpy(pCol->tableName, pTable->tableName);
static void setColumnInfoBySchema(const SRealTableNode* pTable, const SSchema* pColSchema, SColumnNode* pCol) {
strcpy(pCol->dbName, pTable->table.dbName);
strcpy(pCol->tableAlias, pTable->table.tableAlias);
strcpy(pCol->tableName, pTable->table.tableName);
strcpy(pCol->colName, pColSchema->name);
if ('\0' == pCol->node.aliasName[0]) {
strcpy(pCol->node.aliasName, pColSchema->name);
}
pCol->tableId = pTable->pMeta->uid;
pCol->colId = pColSchema->colId;
// pCol->colType = pColSchema->type;
pCol->node.resType.type = pColSchema->type;
@ -382,7 +383,7 @@ static int32_t createColumnNodeByTable(STranslateContext* pCxt, const STableNode
if (NULL == pCol) {
return generateSyntaxErrMsg(pCxt, TSDB_CODE_OUT_OF_MEMORY);
}
setColumnInfoBySchema(pTable, pMeta->schema + i, pCol);
setColumnInfoBySchema((SRealTableNode*)pTable, pMeta->schema + i, pCol);
nodesListAppend(pList, (SNode*)pCol);
}
} else {
@ -407,7 +408,7 @@ static bool findAndSetColumn(SColumnNode* pCol, const STableNode* pTable) {
int32_t nums = pMeta->tableInfo.numOfTags + pMeta->tableInfo.numOfColumns;
for (int32_t i = 0; i < nums; ++i) {
if (0 == strcmp(pCol->colName, pMeta->schema[i].name)) {
setColumnInfoBySchema(pTable, pMeta->schema + i, pCol);
setColumnInfoBySchema((SRealTableNode*)pTable, pMeta->schema + i, pCol);
found = true;
break;
}
@ -879,7 +880,6 @@ static int32_t translateOrderByPosition(STranslateContext* pCxt, SNodeList* pPro
int32_t pos = getPositionValue(pVal);
if (pos < 0) {
ERASE_NODE(pOrderByList);
nodesDestroyNode(pNode);
continue;
} else if (0 == pos || pos > LIST_LENGTH(pProjectionList)) {
return generateSyntaxErrMsg(pCxt, TSDB_CODE_PAR_WRONG_NUMBER_OF_SELECT);
@ -1057,3 +1057,11 @@ int32_t doTranslate(SParseContext* pParseCxt, SQuery* pQuery) {
}
return code;
}
int32_t parser(SParseContext* pParseCxt, SQuery* pQuery) {
int32_t code = doParse(pParseCxt, pQuery);
if (TSDB_CODE_SUCCESS == code) {
code = doTranslate(pParseCxt, pQuery);
}
return code;
}

2845
source/libs/parser/src/sql.c
View File

File diff suppressed because it is too large Load Diff

1
source/libs/parser/src/ttokenizer.c
View File

@ -230,6 +230,7 @@ static SKeyword keywordTable[] = {
{"PORT", TK_PORT},
{"INNER", NEW_TK_INNER},
{"ON", NEW_TK_ON},
{"MODE", TK_MODE},
};
static const char isIdChar[] = {

27
source/libs/planner/inc/plannerImpl.h
View File

@ -20,33 +20,10 @@
extern "C" {
#endif
#include "querynodes.h"
#include "plannodes.h"
#include "planner.h"
typedef struct SLogicNode {
ENodeType type;
int32_t id;
SNodeList* pTargets;
SNode* pConditions;
SNodeList* pChildren;
struct SLogicNode* pParent;
} SLogicNode;
typedef struct SScanLogicNode {
SLogicNode node;
SNodeList* pScanCols;
struct STableMeta* pMeta;
} SScanLogicNode;
typedef struct SFilterLogicNode {
SLogicNode node;
} SFilterLogicNode;
typedef struct SAggLogicNode {
SLogicNode node;
SNodeList* pGroupKeys;
SNodeList* pAggFuncs;
} SAggLogicNode;
int32_t createLogicPlan(SNode* pNode, SLogicNode** pLogicNode);
#ifdef __cplusplus
}

239
source/libs/planner/src/plannerImpl.c
View File

@ -19,6 +19,7 @@
#define CHECK_ALLOC(p, res) \
do { \
if (NULL == p) { \
printf("%s : %d\n", __FUNCTION__, __LINE__); \
pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY; \
return res; \
} \
@ -28,6 +29,7 @@
do { \
int32_t code = exec; \
if (TSDB_CODE_SUCCESS != code) { \
printf("%s : %d\n", __FUNCTION__, __LINE__); \
pCxt->errCode = code; \
return res; \
} \
@ -39,36 +41,57 @@ typedef struct SPlanContext {
SNodeList* pResource;
} SPlanContext;
static SLogicNode* createQueryLogicNode(SPlanContext* pCxt, SNode* pStmt);
static SLogicNode* createLogicNodeByTable(SPlanContext* pCxt, SSelectStmt* pSelect, SNode* pTable);
typedef struct SRewriteExprCxt {
int32_t errCode;
int32_t planNodeId;
SNodeList* pTargets;
SNodeList* pExprs;
} SRewriteExprCxt;
static EDealRes doRewriteExpr(SNode** pNode, void* pContext) {
SRewriteExprCxt* pCxt = (SRewriteExprCxt*)pContext;
SNode* pTarget;
int32_t index = 0;
FOREACH(pTarget, pCxt->pTargets) {
if (nodesEqualNode(pTarget, *pNode)) {
SColumnRefNode* pCol = (SColumnRefNode*)nodesMakeNode(QUERY_NODE_COLUMN_REF);
if (NULL == pCol) {
pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
switch (nodeType(*pNode)) {
case QUERY_NODE_OPERATOR:
case QUERY_NODE_LOGIC_CONDITION:
case QUERY_NODE_FUNCTION: {
SRewriteExprCxt* pCxt = (SRewriteExprCxt*)pContext;
SNode* pExpr;
int32_t index = 0;
FOREACH(pExpr, pCxt->pExprs) {
if (nodesEqualNode(pExpr, *pNode)) {
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
SExprNode* pToBeRewrittenExpr = (SExprNode*)(*pNode);
pCol->node.resType = pToBeRewrittenExpr->resType;
strcpy(pCol->node.aliasName, pToBeRewrittenExpr->aliasName);
strcpy(pCol->colName, ((SExprNode*)pExpr)->aliasName);
nodesDestroyNode(*pNode);
*pNode = (SNode*)pCol;
return DEAL_RES_IGNORE_CHILD;
}
++index;
}
pCol->tupleId = pCxt->planNodeId;
pCol->slotId = index;
nodesDestroyNode(*pNode);
*pNode = (SNode*)pCol;
return DEAL_RES_IGNORE_CHILD;
break;
}
++index;
default:
break;
}
return DEAL_RES_CONTINUE;
}
static int32_t rewriteExpr(int32_t planNodeId, SNodeList* pTargets, SSelectStmt* pSelect, ESqlClause clause) {
SRewriteExprCxt cxt = { .errCode = TSDB_CODE_SUCCESS, .planNodeId = planNodeId, .pTargets = pTargets };
static int32_t rewriteExpr(int32_t planNodeId, int32_t rewriteId, SNodeList* pExprs, SSelectStmt* pSelect, ESqlClause clause) {
SNode* pNode;
FOREACH(pNode, pExprs) {
if (QUERY_NODE_COLUMN == nodeType(pNode) || QUERY_NODE_VALUE == nodeType(pNode)) {
continue;
}
sprintf(((SExprNode*)pNode)->aliasName, "#expr_%d_%d", planNodeId, rewriteId);
}
SRewriteExprCxt cxt = { .errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs };
nodesRewriteSelectStmt(pSelect, clause, doRewriteExpr, &cxt);
return cxt.errCode;
}
@ -102,23 +125,6 @@ error:
return pRoot;
}
static SNodeList* createScanTargets(int32_t planNodeId, int32_t numOfScanCols) {
SNodeList* pTargets = nodesMakeList();
if (NULL == pTargets) {
return NULL;
}
for (int32_t i = 0; i < numOfScanCols; ++i) {
SColumnRefNode* pCol = (SColumnRefNode*)nodesMakeNode(QUERY_NODE_COLUMN_REF);
if (NULL == pCol || TSDB_CODE_SUCCESS != nodesListAppend(pTargets, (SNode*)pCol)) {
nodesDestroyList(pTargets);
return NULL;
}
pCol->tupleId = planNodeId;
pCol->slotId = i;
}
return pTargets;
}
static SLogicNode* createScanLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect, SRealTableNode* pRealTable) {
SScanLogicNode* pScan = (SScanLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_SCAN);
CHECK_ALLOC(pScan, NULL);
@ -128,24 +134,56 @@ static SLogicNode* createScanLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect,
// set columns to scan
SNodeList* pCols = NULL;
CHECK_CODE(nodesCollectColumns(pSelect, SQL_CLAUSE_FROM, pScan->pMeta->uid, true, &pCols), (SLogicNode*)pScan);
CHECK_CODE(nodesCollectColumns(pSelect, SQL_CLAUSE_FROM, pRealTable->table.tableAlias, &pCols), (SLogicNode*)pScan);
pScan->pScanCols = nodesCloneList(pCols);
CHECK_ALLOC(pScan->pScanCols, (SLogicNode*)pScan);
// pScanCols of SScanLogicNode is equivalent to pTargets of other logic nodes
CHECK_CODE(rewriteExpr(pScan->node.id, pScan->pScanCols, pSelect, SQL_CLAUSE_FROM), (SLogicNode*)pScan);
// set output
pScan->node.pTargets = createScanTargets(pScan->node.id, LIST_LENGTH(pScan->pScanCols));
pScan->node.pTargets = nodesCloneList(pCols);
CHECK_ALLOC(pScan->node.pTargets, (SLogicNode*)pScan);
return (SLogicNode*)pScan;
}
static SLogicNode* createQueryLogicNode(SPlanContext* pCxt, SNode* pStmt);
static SLogicNode* createSubqueryLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect, STempTableNode* pTable) {
return createQueryLogicNode(pCxt, pTable->pSubquery);
SLogicNode* pRoot = createQueryLogicNode(pCxt, pTable->pSubquery);
CHECK_ALLOC(pRoot, NULL);
SNode* pNode;
FOREACH(pNode, pRoot->pTargets) {
strcpy(((SColumnNode*)pNode)->tableAlias, pTable->table.tableAlias);
}
return pRoot;
}
static SLogicNode* createJoinLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect, SJoinTableNode* pJoinTable) {
SJoinLogicNode* pJoin = (SJoinLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_JOIN);
CHECK_ALLOC(pJoin, NULL);
pJoin->node.id = pCxt->planNodeId++;
pJoin->joinType = pJoinTable->joinType;
// set left and right node
pJoin->node.pChildren = nodesMakeList();
CHECK_ALLOC(pJoin->node.pChildren, (SLogicNode*)pJoin);
SLogicNode* pLeft = createLogicNodeByTable(pCxt, pSelect, pJoinTable->pLeft);
CHECK_ALLOC(pLeft, (SLogicNode*)pJoin);
CHECK_CODE(nodesListAppend(pJoin->node.pChildren, (SNode*)pLeft), (SLogicNode*)pJoin);
SLogicNode* pRight = createLogicNodeByTable(pCxt, pSelect, pJoinTable->pRight);
CHECK_ALLOC(pRight, (SLogicNode*)pJoin);
CHECK_CODE(nodesListAppend(pJoin->node.pChildren, (SNode*)pRight), (SLogicNode*)pJoin);
// set on conditions
pJoin->pOnConditions = nodesCloneNode(pJoinTable->pOnCond);
CHECK_ALLOC(pJoin->pOnConditions, (SLogicNode*)pJoin);
// set the output
pJoin->node.pTargets = nodesCloneList(pLeft->pTargets);
CHECK_ALLOC(pJoin->node.pTargets, (SLogicNode*)pJoin);
SNodeList* pTargets = nodesCloneList(pRight->pTargets);
CHECK_ALLOC(pTargets, (SLogicNode*)pJoin);
nodesListAppendList(pJoin->node.pTargets, pTargets);
return (SLogicNode*)pJoin;
}
static SLogicNode* createLogicNodeByTable(SPlanContext* pCxt, SSelectStmt* pSelect, SNode* pTable) {
@ -155,14 +193,15 @@ static SLogicNode* createLogicNodeByTable(SPlanContext* pCxt, SSelectStmt* pSele
case QUERY_NODE_TEMP_TABLE:
return createSubqueryLogicNode(pCxt, pSelect, (STempTableNode*)pTable);
case QUERY_NODE_JOIN_TABLE:
return createJoinLogicNode(pCxt, pSelect, (SJoinTableNode*)pTable);
default:
break;
}
return NULL;
}
static SLogicNode* createFilterLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect, SNode* pWhere) {
if (NULL == pWhere) {
static SLogicNode* createWhereFilterLogicNode(SPlanContext* pCxt, SLogicNode* pChild, SSelectStmt* pSelect) {
if (NULL == pSelect->pWhere) {
return NULL;
}
@ -171,23 +210,51 @@ static SLogicNode* createFilterLogicNode(SPlanContext* pCxt, SSelectStmt* pSelec
pFilter->node.id = pCxt->planNodeId++;
// set filter conditions
pFilter->node.pConditions = nodesCloneNode(pWhere);
pFilter->node.pConditions = nodesCloneNode(pSelect->pWhere);
CHECK_ALLOC(pFilter->node.pConditions, (SLogicNode*)pFilter);
// set the output and rewrite the expression in subsequent clauses with the output
SNodeList* pCols = NULL;
CHECK_CODE(nodesCollectColumns(pSelect, SQL_CLAUSE_WHERE, 0, false, &pCols), (SLogicNode*)pFilter);
pFilter->node.pTargets = nodesCloneList(pCols);
// set the output
pFilter->node.pTargets = nodesCloneList(pChild->pTargets);
CHECK_ALLOC(pFilter->node.pTargets, (SLogicNode*)pFilter);
CHECK_CODE(rewriteExpr(pFilter->node.id, pFilter->node.pTargets, pSelect, SQL_CLAUSE_WHERE), (SLogicNode*)pFilter);
return (SLogicNode*)pFilter;
}
static SLogicNode* createAggLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect, SNodeList* pGroupByList, SNode* pHaving) {
static SNodeList* createColumnByRewriteExps(SPlanContext* pCxt, SNodeList* pExprs) {
SNodeList* pList = nodesMakeList();
CHECK_ALLOC(pList, NULL);
SNode* pNode;
FOREACH(pNode, pExprs) {
if (QUERY_NODE_VALUE == nodeType(pNode)) {
continue;
} else if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SNode* pCol = nodesCloneNode(pNode);
if (NULL == pCol) {
goto error;
}
if (TSDB_CODE_SUCCESS != nodesListAppend(pList, pCol)) {
goto error;
}
} else {
SExprNode* pExpr = (SExprNode*)pNode;
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
goto error;
}
pCol->node.resType = pExpr->resType;
strcpy(pCol->colName, pExpr->aliasName);
}
}
return pList;
error:
nodesDestroyList(pList);
return NULL;
}
static SLogicNode* createAggLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect) {
SNodeList* pAggFuncs = NULL;
CHECK_CODE(nodesCollectFuncs(pSelect, fmIsAggFunc, &pAggFuncs), NULL);
if (NULL == pAggFuncs && NULL == pGroupByList) {
if (NULL == pAggFuncs && NULL == pSelect->pGroupByList) {
return NULL;
}
@ -196,32 +263,74 @@ static SLogicNode* createAggLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect,
pAgg->node.id = pCxt->planNodeId++;
// set grouyp keys, agg funcs and having conditions
pAgg->pGroupKeys = nodesCloneList(pGroupByList);
pAgg->pGroupKeys = nodesCloneList(pSelect->pGroupByList);
CHECK_ALLOC(pAgg->pGroupKeys, (SLogicNode*)pAgg);
pAgg->pAggFuncs = nodesCloneList(pAggFuncs);
CHECK_ALLOC(pAgg->pAggFuncs, (SLogicNode*)pAgg);
pAgg->node.pConditions = nodesCloneNode(pHaving);
// rewrite the expression in subsequent clauses
CHECK_CODE(rewriteExpr(pAgg->node.id, 1, pAgg->pGroupKeys, pSelect, SQL_CLAUSE_GROUP_BY), (SLogicNode*)pAgg);
CHECK_CODE(rewriteExpr(pAgg->node.id, 1 + LIST_LENGTH(pAgg->pGroupKeys), pAgg->pAggFuncs, pSelect, SQL_CLAUSE_GROUP_BY), (SLogicNode*)pAgg);
pAgg->node.pConditions = nodesCloneNode(pSelect->pHaving);
CHECK_ALLOC(pAgg->node.pConditions, (SLogicNode*)pAgg);
// set the output and rewrite the expression in subsequent clauses with the output
SNodeList* pCols = NULL;
CHECK_CODE(nodesCollectColumns(pSelect, SQL_CLAUSE_HAVING, 0, false, &pCols), (SLogicNode*)pAgg);
pAgg->node.pTargets = nodesCloneList(pCols);
// set the output
pAgg->node.pTargets = createColumnByRewriteExps(pCxt, pAgg->pGroupKeys);
CHECK_ALLOC(pAgg->node.pTargets, (SLogicNode*)pAgg);
CHECK_CODE(rewriteExpr(pAgg->node.id, pAgg->node.pTargets, pSelect, SQL_CLAUSE_HAVING), (SLogicNode*)pAgg);
SNodeList* pTargets = createColumnByRewriteExps(pCxt, pAgg->pAggFuncs);
CHECK_ALLOC(pTargets, (SLogicNode*)pAgg);
nodesListAppendList(pAgg->node.pTargets, pTargets);
return (SLogicNode*)pAgg;
}
static SNodeList* createColumnByProjections(SPlanContext* pCxt, SNodeList* pExprs) {
SNodeList* pList = nodesMakeList();
CHECK_ALLOC(pList, NULL);
SNode* pNode;
FOREACH(pNode, pExprs) {
SExprNode* pExpr = (SExprNode*)pNode;
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
goto error;
}
pCol->node.resType = pExpr->resType;
strcpy(pCol->colName, pExpr->aliasName);
if (TSDB_CODE_SUCCESS != nodesListAppend(pList, (SNode*)pCol)) {
goto error;
}
}
return pList;
error:
nodesDestroyList(pList);
return NULL;
}
static SLogicNode* createProjectLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect) {
SProjectLogicNode* pProject = (SProjectLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_PROJECT);
CHECK_ALLOC(pProject, NULL);
pProject->node.id = pCxt->planNodeId++;
pProject->pProjections = nodesCloneList(pSelect->pProjectionList);
pProject->node.pTargets = createColumnByProjections(pCxt,pSelect->pProjectionList);
CHECK_ALLOC(pProject->node.pTargets, (SLogicNode*)pProject);
return (SLogicNode*)pProject;
}
static SLogicNode* createSelectLogicNode(SPlanContext* pCxt, SSelectStmt* pSelect) {
SLogicNode* pRoot = createLogicNodeByTable(pCxt, pSelect, pSelect->pFromTable);
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pRoot = pushLogicNode(pCxt, pRoot, createFilterLogicNode(pCxt, pSelect, pSelect->pWhere));
pRoot = pushLogicNode(pCxt, pRoot, createWhereFilterLogicNode(pCxt, pRoot, pSelect));
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pRoot = pushLogicNode(pCxt, pRoot, createAggLogicNode(pCxt, pSelect, pSelect->pGroupByList, pSelect->pHaving));
pRoot = pushLogicNode(pCxt, pRoot, createAggLogicNode(pCxt, pSelect));
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pRoot = pushLogicNode(pCxt, pRoot, createProjectLogicNode(pCxt, pSelect));
}
// pRoot = pushLogicNode(pCxt, pRoot, createProjectLogicNode(pSelect, pSelect->pProjectionList));
return pRoot;
}

2
source/libs/planner/test/CMakeLists.txt
View File

@ -13,7 +13,7 @@ ADD_EXECUTABLE(plannerTest
TARGET_LINK_LIBRARIES(
plannerTest
PUBLIC os util common planner parser catalog transport gtest function qcom
PUBLIC os util common nodes planner parser catalog transport gtest function qcom
)
TARGET_INCLUDE_DIRECTORIES(

91
source/libs/planner/test/newPlannerTest.cpp Normal file
View File

@ -0,0 +1,91 @@
/*
* 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 <algorithm>
#include <gtest/gtest.h>
#include "plannerImpl.h"
#include "newParser.h"
using namespace std;
using namespace testing;
class NewPlannerTest : public Test {
protected:
void setDatabase(const string& acctId, const string& db) {
acctId_ = acctId;
db_ = db;
}
void bind(const char* sql) {
reset();
cxt_.acctId = atoi(acctId_.c_str());
cxt_.db = db_.c_str();
sqlBuf_ = string(sql);
transform(sqlBuf_.begin(), sqlBuf_.end(), sqlBuf_.begin(), ::tolower);
cxt_.sqlLen = strlen(sql);
cxt_.pSql = sqlBuf_.c_str();
}
bool run() {
int32_t code = parser(&cxt_, &query_);
// cout << "parser return " << code << endl;
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] parser code:" << tstrerror(code) << ", msg:" << errMagBuf_ << endl;
return false;
}
SLogicNode* pLogicPlan = nullptr;
code = createLogicPlan(query_.pRoot, &pLogicPlan);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] plan code:" << tstrerror(code) << endl;
return false;
}
char* pStr = NULL;
int32_t len = 0;
code = nodesNodeToString((const SNode*)pLogicPlan, &pStr, &len);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] toString code:" << tstrerror(code) << endl;
return false;
}
cout << "logic plan : " << endl;
cout << pStr << endl;
return true;
}
private:
static const int max_err_len = 1024;
void reset() {
memset(&cxt_, 0, sizeof(cxt_));
memset(errMagBuf_, 0, max_err_len);
cxt_.pMsg = errMagBuf_;
cxt_.msgLen = max_err_len;
}
string acctId_;
string db_;
char errMagBuf_[max_err_len];
string sqlBuf_;
SParseContext cxt_;
SQuery query_;
};
TEST_F(NewPlannerTest, simple) {
setDatabase("root", "test");
bind("SELECT * FROM t1");
ASSERT_TRUE(run());
}

59
source/libs/tdb/inc/tdb.h
View File

@ -22,41 +22,44 @@
extern "C" {
#endif
// #define TDB_EXTERN
// #define TDB_PUBLIC
// #define TDB_STATIC static
typedef struct STDb TDB;
typedef struct STDbEnv TENV;
typedef struct STDbCurosr TDBC;
// typedef enum { TDB_BTREE_T = 0, TDB_HASH_T = 1, TDB_HEAP_T = 2 } tdb_db_t;
typedef int32_t pgsz_t;
typedef int32_t cachesz_t;
// // Forward declarations
// typedef struct TDB TDB;
// // typedef struct TDB_MPOOL TDB_MPOOL;
// // typedef struct TDB_MPFILE TDB_MPFILE;
// // typedef struct TDB_CURSOR TDB_CURSOR;
typedef int (*TdbKeyCmprFn)(int keyLen1, const void *pKey1, int keyLen2, const void *pKey2);
// typedef struct {
// void* bdata;
// uint32_t size;
// } TDB_KEY, TDB_VALUE;
// TEVN
int tdbEnvCreate(TENV **ppEnv, const char *rootDir);
int tdbEnvOpen(TENV *ppEnv);
int tdbEnvClose(TENV *pEnv);
// // TDB Operations
// int tdbCreateDB(TDB** dbpp, tdb_db_t type);
// int tdbOpenDB(TDB* dbp, const char* fname, const char* dbname, uint32_t flags);
// int tdbCloseDB(TDB* dbp, uint32_t flags);
// int tdbPut(TDB* dbp, const TDB_KEY* key, const TDB_VALUE* value, uint32_t flags);
// int tdbGet(TDB* dbp, const TDB_KEY* key, TDB_VALUE* value, uint32_t flags);
int tdbEnvSetCache(TENV *pEnv, pgsz_t pgSize, cachesz_t cacheSize);
pgsz_t tdbEnvGetPageSize(TENV *pEnv);
cachesz_t tdbEnvGetCacheSize(TENV *pEnv);
// // TDB_MPOOL
// int tdbOpenMPool(TDB_MPOOL** mp);
// int tdbCloseMPool(TDB_MPOOL* mp);
int tdbEnvBeginTxn(TENV *pEnv);
int tdbEnvCommit(TENV *pEnv);
// // TDB_MPFILE
// int tdbOpenMPFile(TDB_MPFILE** mpf, TDB_MPOOL* mp);
// int tdbCloseMPFile(TDB_MPFILE** mpf);
// TDB
int tdbCreate(TDB **ppDb);
int tdbOpen(TDB *pDb, const char *fname, const char *dbname, TENV *pEnv);
int tdbClose(TDB *pDb);
int tdbDrop(TDB *pDb);
// // TDB_CURSOR
// int tdbOpenCursor(TDB* dbp, TDB_CURSOR** tdbcpp);
// int tdbCloseCurosr(TDB_CURSOR* tdbcp);
int tdbSetKeyLen(TDB *pDb, int klen);
int tdbSetValLen(TDB *pDb, int vlen);
int tdbSetDup(TDB *pDb, int dup);
int tdbSetCmprFunc(TDB *pDb, TdbKeyCmprFn fn);
int tdbGetKeyLen(TDB *pDb);
int tdbGetValLen(TDB *pDb);
int tdbGetDup(TDB *pDb);
int tdbInsert(TDB *pDb, const void *pKey, int nKey, const void *pData, int nData);
// TDBC
#ifdef __cplusplus
}

205
source/libs/tdb/src/db/tdb.c Normal file
View File

@ -0,0 +1,205 @@
/*
* 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 "tdbInt.h"
struct STDb {
char dbname[TDB_MAX_DBNAME_LEN];
SBTree * pBt; // current access method (may extend)
SPgFile * pPgFile; // backend page file this DB is using
TENV * pEnv; // TENV containing the DB
int klen; // key length if know
int vlen; // value length if know
bool dup; // dup mode
TdbKeyCmprFn cFn; // compare function
};
struct STDbCurosr {
SBtCursor *pBtCur;
};
static int tdbDefaultKeyCmprFn(int keyLen1, const void *pKey1, int keyLen2, const void *pKey2);
int tdbCreate(TDB **ppDb) {
TDB *pDb;
// create the handle
pDb = (TDB *)calloc(1, sizeof(*pDb));
if (pDb == NULL) {
return -1;
}
pDb->klen = TDB_VARIANT_LEN;
pDb->vlen = TDB_VARIANT_LEN;
pDb->dup = false;
pDb->cFn = tdbDefaultKeyCmprFn;
*ppDb = pDb;
return 0;
}
static int tdbDestroy(TDB *pDb) {
if (pDb) {
free(pDb);
}
return 0;
}
int tdbOpen(TDB *pDb, const char *fname, const char *dbname, TENV *pEnv) {
int ret;
uint8_t fileid[TDB_FILE_ID_LEN];
SPgFile * pPgFile;
SPgCache *pPgCache;
SBTree * pBt;
bool fileExist;
size_t dbNameLen;
pgno_t dbRootPgno;
char dbfname[128]; // TODO: make this as a macro or malloc on the heap
ASSERT(pDb != NULL);
ASSERT(fname != NULL);
// TODO: Here we simply put an assert here. In the future, make `pEnv`
// can be set as NULL.
ASSERT(pEnv != NULL);
// check the DB name
dbNameLen = 0;
if (dbname) {
dbNameLen = strlen(dbname);
if (dbNameLen >= TDB_MAX_DBNAME_LEN) {
return -1;
}
memcpy(pDb->dbname, dbname, dbNameLen);
}
pDb->dbname[dbNameLen] = '\0';
// get page file from the env, if not opened yet, open it
pPgFile = NULL;
snprintf(dbfname, 128, "%s/%s", tdbEnvGetRootDir(pEnv), fname);
fileExist = (tdbCheckFileAccess(fname, TDB_F_OK) == 0);
if (fileExist) {
tdbGnrtFileID(dbfname, fileid, false);
pPgFile = tdbEnvGetPageFile(pEnv, fileid);
}
if (pPgFile == NULL) {
ret = pgFileOpen(&pPgFile, dbfname, pEnv);
if (ret != 0) {
// TODO: handle error
return -1;
}
}
// TODO: get the root page number from the master DB of the page file
// tdbGet(&dbRootPgno);
if (dbRootPgno == 0) {
// DB not exist, create one
ret = pgFileAllocatePage(pPgFile, &dbRootPgno);
if (ret != 0) {
// TODO: handle error
}
// tdbInsert(pPgFile->pMasterDB, dbname, strlen(dbname), &dbRootPgno, sizeof(dbRootPgno));
}
ASSERT(dbRootPgno > 1);
// pDb->pBt->root = dbRootPgno;
// register
pDb->pPgFile = pPgFile;
tdbEnvRgstDB(pEnv, pDb);
pDb->pEnv = pEnv;
return 0;
}
int tdbClose(TDB *pDb) {
if (pDb == NULL) return 0;
return tdbDestroy(pDb);
}
int tdbDrop(TDB *pDb) {
// TODO
return 0;
}
int tdbSetKeyLen(TDB *pDb, int klen) {
// TODO: check `klen`
pDb->klen = klen;
return 0;
}
int tdbSetValLen(TDB *pDb, int vlen) {
// TODO: check `vlen`
pDb->vlen = vlen;
return 0;
}
int tdbSetDup(TDB *pDb, int dup) {
if (dup) {
pDb->dup = true;
} else {
pDb->dup = false;
}
return 0;
}
int tdbSetCmprFunc(TDB *pDb, TdbKeyCmprFn fn) {
if (fn == NULL) {
return -1;
} else {
pDb->cFn = fn;
}
return 0;
}
int tdbGetKeyLen(TDB *pDb) { return pDb->klen; }
int tdbGetValLen(TDB *pDb) { return pDb->vlen; }
int tdbGetDup(TDB *pDb) {
if (pDb->dup) {
return 1;
} else {
return 0;
}
}
int tdbInsert(TDB *pDb, const void *pKey, int nKey, const void *pData, int nData) {
// TODO
return 0;
}
static int tdbDefaultKeyCmprFn(int keyLen1, const void *pKey1, int keyLen2, const void *pKey2) {
int mlen;
int cret;
ASSERT(keyLen1 > 0 && keyLen2 > 0 && pKey1 != NULL && pKey2 != NULL);
mlen = keyLen1 < keyLen2 ? keyLen1 : keyLen2;
cret = memcmp(pKey1, pKey2, mlen);
if (cret == 0) {
if (keyLen1 < keyLen2) {
cret = -1;
} else if (keyLen1 > keyLen2) {
cret = 1;
} else {
cret = 0;
}
}
return cret;
}

164
source/libs/tdb/src/db/tdbBtree.c Normal file
View File

@ -0,0 +1,164 @@
/*
* 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 "tdbInt.h"
struct SBtCursor {
SBTree *pBtree;
pgno_t pgno;
SPage * pPage; // current page traversing
};
typedef struct {
pgno_t pgno;
pgsz_t offset;
} SBtIdx;
// Btree page header definition
typedef struct __attribute__((__packed__)) {
uint8_t flag; // page flag
int32_t vlen; // value length of current page, TDB_VARIANT_LEN for variant length
uint16_t nPayloads; // number of total payloads
pgoff_t freeOff; // free payload offset
pgsz_t fragSize; // total fragment size
pgoff_t offPayload; // payload offset
pgno_t rChildPgno; // right most child page number
} SBtPgHdr;
typedef int (*BtreeCmprFn)(const void *, const void *);
#define BTREE_PAGE_HDR(pPage) NULL /* TODO */
#define BTREE_PAGE_PAYLOAD_AT(pPage, idx) NULL /*TODO*/
#define BTREE_PAGE_IS_LEAF(pPage) 0 /* TODO */
static int btreeCreate(SBTree **ppBt);
static int btreeDestroy(SBTree *pBt);
static int btreeCursorMoveToChild(SBtCursor *pBtCur, pgno_t pgno);
int btreeOpen(SBTree **ppBt, SPgFile *pPgFile) {
SBTree *pBt;
int ret;
ret = btreeCreate(&pBt);
if (ret != 0) {
return -1;
}
*ppBt = pBt;
return 0;
}
int btreeClose(SBTree *pBt) {
// TODO
return 0;
}
static int btreeCreate(SBTree **ppBt) {
SBTree *pBt;
pBt = (SBTree *)calloc(1, sizeof(*pBt));
if (pBt == NULL) {
return -1;
}
// TODO
return 0;
}
static int btreeDestroy(SBTree *pBt) {
if (pBt) {
free(pBt);
}
return 0;
}
int btreeCursorOpen(SBtCursor *pBtCur, SBTree *pBt) {
// TODO
return 0;
}
int btreeCursorClose(SBtCursor *pBtCur) {
// TODO
return 0;
}
int btreeCursorMoveTo(SBtCursor *pBtCur, int kLen, const void *pKey) {
SPage * pPage;
SBtPgHdr * pBtPgHdr;
SPgFile * pPgFile;
pgno_t childPgno;
pgno_t rootPgno;
int nPayloads;
void * pPayload;
BtreeCmprFn cmpFn;
// 1. Move the cursor to the root page
if (rootPgno == TDB_IVLD_PGNO) {
// No any data in this btree, just return not found (TODO)
return 0;
} else {
// Load the page from the file by the SPgFile handle
pPage = pgFileFetch(pPgFile, rootPgno);
pBtCur->pPage = pPage;
}
// 2. Loop to search over the whole tree
for (;;) {
int lidx, ridx, midx, cret;
pPage = pBtCur->pPage;
pBtPgHdr = BTREE_PAGE_HDR(pPage);
nPayloads = pBtPgHdr->nPayloads;
// Binary search the page
lidx = 0;
ridx = nPayloads - 1;
midx = (lidx + ridx) >> 1;
for (;;) {
// get the payload ptr at midx
pPayload = BTREE_PAGE_PAYLOAD_AT(pPage, midx);
// the payload and the key
cret = cmpFn(pKey, pPayload);
if (cret < 0) {
/* TODO */
} else if (cret > 0) {
/* TODO */
} else {
/* TODO */
}
if (lidx > ridx) break;
midx = (lidx + ridx) >> 1;
}
if (BTREE_PAGE_IS_LEAF(pPage)) {
/* TODO */
break;
} else {
/* TODO */
btreeCursorMoveToChild(pBtCur, childPgno);
}
}
return 0;
}
static int btreeCursorMoveToChild(SBtCursor *pBtCur, pgno_t pgno) {
SPgFile *pPgFile;
// TODO
return 0;
}

173
source/libs/tdb/src/db/tdbEnv.c Normal file
View File

@ -0,0 +1,173 @@
/*
* 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 "tdbInt.h"
struct STDbEnv {
char * rootDir; // root directory of the environment
char * jname; // journal file name
int jfd; // journal file fd
pgsz_t pgSize; // page size
cachesz_t cacheSize; // total cache size
STDbList dbList; // TDB List
SPgFileList pgfList; // SPgFile List
SPgCache * pPgCache; // page cache
struct {
#define TDB_ENV_PGF_HASH_BUCKETS 17
SPgFileList buckets[TDB_ENV_PGF_HASH_BUCKETS];
} pgfht; // page file hash table;
};
#define TDB_ENV_PGF_HASH(fileid) \
({ \
uint8_t *tmp = (uint8_t *)(fileid); \
tmp[0] + tmp[1] + tmp[2]; \
})
static int tdbEnvDestroy(TENV *pEnv);
int tdbEnvCreate(TENV **ppEnv, const char *rootDir) {
TENV * pEnv;
size_t slen;
size_t jlen;
ASSERT(rootDir != NULL);
*ppEnv = NULL;
slen = strlen(rootDir);
jlen = slen + strlen(TDB_JOURNAL_NAME) + 1;
pEnv = (TENV *)calloc(1, sizeof(*pEnv) + slen + 1 + jlen + 1);
if (pEnv == NULL) {
return -1;
}
pEnv->rootDir = (char *)(&pEnv[1]);
pEnv->jname = pEnv->rootDir + slen + 1;
pEnv->jfd = -1;
pEnv->pgSize = TDB_DEFAULT_PGSIZE;
pEnv->cacheSize = TDB_DEFAULT_CACHE_SIZE;
memcpy(pEnv->rootDir, rootDir, slen);
pEnv->rootDir[slen] = '\0';
sprintf(pEnv->jname, "%s/%s", rootDir, TDB_JOURNAL_NAME);
TD_DLIST_INIT(&(pEnv->dbList));
TD_DLIST_INIT(&(pEnv->pgfList));
/* TODO */
*ppEnv = pEnv;
return 0;
}
int tdbEnvOpen(TENV *pEnv) {
SPgCache *pPgCache;
int ret;
ASSERT(pEnv != NULL);
/* TODO: here we do not need to create the root directory, more
* work should be done here
*/
mkdir(pEnv->rootDir, 0755);
ret = pgCacheOpen(&pPgCache, pEnv);
if (ret != 0) {
goto _err;
}
pEnv->pPgCache = pPgCache;
return 0;
_err:
return -1;
}
int tdbEnvClose(TENV *pEnv) {
if (pEnv == NULL) return 0;
pgCacheClose(pEnv->pPgCache);
tdbEnvDestroy(pEnv);
return 0;
}
int tdbEnvSetCache(TENV *pEnv, pgsz_t pgSize, cachesz_t cacheSize) {
if (!TDB_IS_PGSIZE_VLD(pgSize) || cacheSize / pgSize < 10) {
return -1;
}
/* TODO */
pEnv->pgSize = pgSize;
pEnv->cacheSize = cacheSize;
return 0;
}
pgsz_t tdbEnvGetPageSize(TENV *pEnv) { return pEnv->pgSize; }
cachesz_t tdbEnvGetCacheSize(TENV *pEnv) { return pEnv->cacheSize; }
SPgFile *tdbEnvGetPageFile(TENV *pEnv, const uint8_t fileid[]) {
SPgFileList *pBucket;
SPgFile * pPgFile;
pBucket = pEnv->pgfht.buckets + (TDB_ENV_PGF_HASH(fileid) % TDB_ENV_PGF_HASH_BUCKETS); // TODO
for (pPgFile = TD_DLIST_HEAD(pBucket); pPgFile != NULL; pPgFile = TD_DLIST_NODE_NEXT_WITH_FIELD(pPgFile, envHash)) {
if (memcmp(fileid, pPgFile->fileid, TDB_FILE_ID_LEN) == 0) break;
};
return pPgFile;
}
SPgCache *tdbEnvGetPgCache(TENV *pEnv) { return pEnv->pPgCache; }
static int tdbEnvDestroy(TENV *pEnv) {
// TODO
return 0;
}
int tdbEnvBeginTxn(TENV *pEnv) {
pEnv->jfd = open(pEnv->jname, O_CREAT | O_RDWR, 0755);
if (pEnv->jfd < 0) {
return -1;
}
return 0;
}
int tdbEnvCommit(TENV *pEnv) {
/* TODO */
close(pEnv->jfd);
pEnv->jfd = -1;
return 0;
}
const char *tdbEnvGetRootDir(TENV *pEnv) { return pEnv->rootDir; }
int tdbEnvRgstPageFile(TENV *pEnv, SPgFile *pPgFile) {
SPgFileList *pBucket;
TD_DLIST_APPEND_WITH_FIELD(&(pEnv->pgfList), pPgFile, envPgfList);
pBucket = pEnv->pgfht.buckets + (TDB_ENV_PGF_HASH(pPgFile->fileid) % TDB_ENV_PGF_HASH_BUCKETS); // TODO
TD_DLIST_APPEND_WITH_FIELD(pBucket, pPgFile, envHash);
return 0;
}
int tdbEnvRgstDB(TENV *pEnv, TDB *pDb) {
// TODO
return 0;
}

247
source/libs/tdb/src/db/tdb_mpool.c → source/libs/tdb/src/db/tdbPgCache.c
View File

@ -12,10 +12,225 @@
* 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 "tdbInt.h"
#include "tdb_mpool.h"
typedef TD_DLIST(SPage) SPgList;
struct SPgCache {
TENV * pEnv; // TENV containing this page cache
pgsz_t pgsize;
int32_t npage;
SPage **pages;
SPgList freeList;
SPgList lru;
struct {
int32_t nbucket;
SPgList *buckets;
} pght; // page hash table
};
static void pgCachePinPage(SPage *pPage);
static void pgCacheUnpinPage(SPage *pPage);
int pgCacheOpen(SPgCache **ppPgCache, TENV *pEnv) {
SPgCache *pPgCache;
SPage * pPage;
void * pData;
pgsz_t pgSize;
cachesz_t cacheSize;
int32_t npage;
int32_t nbucket;
size_t msize;
*ppPgCache = NULL;
pgSize = tdbEnvGetPageSize(pEnv);
cacheSize = tdbEnvGetCacheSize(pEnv);
npage = cacheSize / pgSize;
nbucket = npage;
msize = sizeof(*pPgCache) + sizeof(SPage *) * npage + sizeof(SPgList) * nbucket;
// Allocate the handle
pPgCache = (SPgCache *)calloc(1, msize);
if (pPgCache == NULL) {
return -1;
}
// Init the handle
pPgCache->pEnv = pEnv;
pPgCache->pgsize = pgSize;
pPgCache->npage = npage;
pPgCache->pages = (SPage **)(&pPgCache[1]);
pPgCache->pght.nbucket = nbucket;
pPgCache->pght.buckets = (SPgList *)(&(pPgCache->pages[npage]));
TD_DLIST_INIT(&(pPgCache->freeList));
for (int32_t i = 0; i < npage; i++) {
pData = malloc(pgSize + sizeof(SPage));
if (pData == NULL) {
return -1;
// TODO: handle error
}
pPage = POINTER_SHIFT(pData, pgSize);
pPage->pgid = TDB_IVLD_PGID;
pPage->frameid = i;
pPage->pData = pData;
// add current page to the page cache
pPgCache->pages[i] = pPage;
TD_DLIST_APPEND_WITH_FIELD(&(pPgCache->freeList), pPage, freeNode);
}
#if 0
for (int32_t i = 0; i < nbucket; i++) {
TD_DLIST_INIT(pPgCache->pght.buckets + i);
}
#endif
*ppPgCache = pPgCache;
return 0;
}
int pgCacheClose(SPgCache *pPgCache) {
SPage *pPage;
if (pPgCache) {
for (int32_t i = 0; i < pPgCache->npage; i++) {
pPage = pPgCache->pages[i];
tfree(pPage->pData);
}
free(pPgCache);
}
return 0;
}
#define PG_CACHE_HASH(fileid, pgno) \
({ \
uint64_t *tmp = (uint64_t *)(fileid); \
(tmp[0] + tmp[1] + tmp[2] + (pgno)); \
})
SPage *pgCacheFetch(SPgCache *pPgCache, pgid_t pgid) {
SPage * pPage;
SPgFile *pPgFile;
SPgList *pBucket;
// 1. Search the page hash table SPgCache.pght
pBucket = pPgCache->pght.buckets + (PG_CACHE_HASH(pgid.fileid, pgid.pgno) % pPgCache->pght.nbucket);
pPage = TD_DLIST_HEAD(pBucket);
while (pPage && tdbCmprPgId(&(pPage->pgid), &pgid)) {
pPage = TD_DLIST_NODE_NEXT_WITH_FIELD(pPage, pghtNode);
}
if (pPage) {
// Page is found, pin the page and return the page
pgCachePinPage(pPage);
return pPage;
}
// 2. Check the free list
pPage = TD_DLIST_HEAD(&(pPgCache->freeList));
if (pPage) {
TD_DLIST_POP_WITH_FIELD(&(pPgCache->freeList), pPage, freeNode);
pgCachePinPage(pPage);
return pPage;
}
// 3. Try to recycle a page from the LRU list
pPage = TD_DLIST_HEAD(&(pPgCache->lru));
if (pPage) {
TD_DLIST_POP_WITH_FIELD(&(pPgCache->lru), pPage, lruNode);
// TODO: remove from the hash table
pgCachePinPage(pPage);
return pPage;
}
// 4. If a memory allocator is set, try to allocate from the allocator (TODO)
return NULL;
}
int pgCacheRelease(SPage *pPage) {
// TODO
return 0;
}
static void pgCachePinPage(SPage *pPage) {
// TODO
}
static void pgCacheUnpinPage(SPage *pPage) {
// TODO
}
#if 0
// Exposed handle
typedef struct TDB_MPOOL TDB_MPOOL;
typedef struct TDB_MPFILE TDB_MPFILE;
typedef TD_DLIST_NODE(pg_t) pg_free_dlist_node_t, pg_hash_dlist_node_t;
typedef struct pg_t {
SRWLatch rwLatch;
frame_id_t frameid;
pgid_t pgid;
uint8_t dirty;
uint8_t rbit;
int32_t pinRef;
pg_free_dlist_node_t free;
pg_hash_dlist_node_t hash;
void * p;
} pg_t;
typedef TD_DLIST(pg_t) pg_list_t;
typedef struct {
SRWLatch latch;
TD_DLIST(TDB_MPFILE);
} mpf_bucket_t;
struct TDB_MPOOL {
int64_t cachesize;
pgsz_t pgsize;
int32_t npages;
pg_t * pages;
pg_list_t freeList;
frame_id_t clockHand;
struct {
int32_t nbucket;
pg_list_t *hashtab;
} pgtab; // page table, hash<pgid_t, pg_t>
struct {
#define MPF_HASH_BUCKETS 16
mpf_bucket_t buckets[MPF_HASH_BUCKETS];
} mpfht; // MPF hash table. MPFs using this MP will be put in this hash table
};
#define MP_PAGE_AT(mp, idx) (mp)->pages[idx]
typedef TD_DLIST_NODE(TDB_MPFILE) td_mpf_dlist_node_t;
struct TDB_MPFILE {
char * fname; // file name
int fd; // fd
uint8_t fileid[TDB_FILE_ID_LEN]; // file ID
TDB_MPOOL * mp; // underlying memory pool
td_mpf_dlist_node_t node;
};
/*=================================================== Exposed apis ==================================================*/
// TDB_MPOOL
int tdbMPoolOpen(TDB_MPOOL **mpp, uint64_t cachesize, pgsz_t pgsize);
int tdbMPoolClose(TDB_MPOOL *mp);
int tdbMPoolSync(TDB_MPOOL *mp);
// TDB_MPFILE
int tdbMPoolFileOpen(TDB_MPFILE **mpfp, const char *fname, TDB_MPOOL *mp);
int tdbMPoolFileClose(TDB_MPFILE *mpf);
int tdbMPoolFileNewPage(TDB_MPFILE *mpf, pgno_t *pgno, void *addr);
int tdbMPoolFileFreePage(TDB_MPOOL *mpf, pgno_t *pgno, void *addr);
int tdbMPoolFileGetPage(TDB_MPFILE *mpf, pgno_t pgno, void *addr);
int tdbMPoolFilePutPage(TDB_MPFILE *mpf, pgno_t pgno, void *addr);
int tdbMPoolFileSync(TDB_MPFILE *mpf);
static int tdbGnrtFileID(const char *fname, uint8_t *fileid);
static void tdbMPoolRegFile(TDB_MPOOL *mp, TDB_MPFILE *mpf);
static void tdbMPoolUnregFile(TDB_MPOOL *mp, TDB_MPFILE *mpf);
static TDB_MPFILE *tdbMPoolGetFile(TDB_MPOOL *mp, uint8_t *fileid);
@ -23,7 +238,7 @@ static int tdbMPoolFileReadPage(TDB_MPFILE *mpf, pgno_t pgno, void *p);
static int tdbMPoolFileWritePage(TDB_MPFILE *mpf, pgno_t pgno, const void *p);
static void tdbMPoolClockEvictPage(TDB_MPOOL *mp, pg_t **pagepp);
int tdbMPoolOpen(TDB_MPOOL **mpp, uint64_t cachesize, pgsize_t pgsize) {
int tdbMPoolOpen(TDB_MPOOL **mpp, uint64_t cachesize, pgsz_t pgsize) {
TDB_MPOOL *mp = NULL;
size_t tsize;
pg_t * pagep;
@ -120,7 +335,7 @@ int tdbMPoolFileOpen(TDB_MPFILE **mpfp, const char *fname, TDB_MPOOL *mp) {
goto _err;
}
if (tdbGnrtFileID(fname, mpf->fileid) < 0) {
if (tdbGnrtFileID(fname, mpf->fileid, false) < 0) {
goto _err;
}
@ -230,22 +445,6 @@ int tdbMPoolFilePutPage(TDB_MPFILE *mpf, pgno_t pgno, void *addr) {
return 0;
}
static int tdbGnrtFileID(const char *fname, uint8_t *fileid) {
struct stat statbuf;
if (stat(fname, &statbuf) < 0) {
return -1;
}
memset(fileid, 0, TDB_FILE_ID_LEN);
((uint64_t *)fileid)[0] = (uint64_t)statbuf.st_ino;
((uint64_t *)fileid)[1] = (uint64_t)statbuf.st_dev;
((uint64_t *)fileid)[2] = rand();
return 0;
}
#define MPF_GET_BUCKETID(fileid) \
({ \
uint64_t *tmp = (uint64_t *)fileid; \
@ -317,7 +516,7 @@ static void tdbMPoolUnregFile(TDB_MPOOL *mp, TDB_MPFILE *mpf) {
}
static int tdbMPoolFileReadPage(TDB_MPFILE *mpf, pgno_t pgno, void *p) {
pgsize_t pgsize;
pgsz_t pgsize;
TDB_MPOOL *mp;
off_t offset;
size_t rsize;
@ -334,7 +533,7 @@ static int tdbMPoolFileReadPage(TDB_MPFILE *mpf, pgno_t pgno, void *p) {
}
static int tdbMPoolFileWritePage(TDB_MPFILE *mpf, pgno_t pgno, const void *p) {
pgsize_t pgsize;
pgsz_t pgsize;
TDB_MPOOL *mp;
off_t offset;
@ -376,4 +575,6 @@ static void tdbMPoolClockEvictPage(TDB_MPOOL *mp, pg_t **pagepp) {
} while (1);
*pagepp = pagep;
}
}
#endif

221
source/libs/tdb/src/db/tdbPgFile.c Normal file
View File

@ -0,0 +1,221 @@
/*
* 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 "tdbInt.h"
typedef struct SPage1 {
char magic[64];
pgno_t mdbRootPgno; // master DB root page number
pgno_t freePgno; // free list page number
uint32_t nFree; // number of free pages
} SPage1;
typedef struct SFreePage {
/* TODO */
} SFreePage;
TDB_STATIC_ASSERT(sizeof(SPage1) <= TDB_MIN_PGSIZE, "TDB Page1 definition too large");
static int pgFileRead(SPgFile *pPgFile, pgno_t pgno, uint8_t *pData);
int pgFileOpen(SPgFile **ppPgFile, const char *fname, TENV *pEnv) {
SPgFile * pPgFile;
SPgCache *pPgCache;
size_t fnameLen;
pgno_t fsize;
*ppPgFile = NULL;
// create the handle
fnameLen = strlen(fname);
pPgFile = (SPgFile *)calloc(1, sizeof(*pPgFile) + fnameLen + 1);
if (pPgFile == NULL) {
return -1;
}
ASSERT(pEnv != NULL);
// init the handle
pPgFile->fname = (char *)(&(pPgFile[1]));
memcpy(pPgFile->fname, fname, fnameLen);
pPgFile->fname[fnameLen] = '\0';
pPgFile->fd = -1;
pPgFile->fd = open(fname, O_CREAT | O_RDWR, 0755);
if (pPgFile->fd < 0) {
// TODO: handle error
return -1;
}
tdbGnrtFileID(fname, pPgFile->fileid, false);
tdbGetFileSize(fname, tdbEnvGetPageSize(pEnv), &fsize);
pPgFile->fsize = fsize;
pPgFile->lsize = fsize;
if (pPgFile->fsize == 0) {
// A created file
pgno_t pgno;
pgid_t pgid;
pgFileAllocatePage(pPgFile, &pgno);
ASSERT(pgno == 1);
memcpy(pgid.fileid, pPgFile->fileid, TDB_FILE_ID_LEN);
pgid.pgno = pgno;
pgCacheFetch(pPgCache, pgid);
// Need to allocate the first page as a description page
} else {
// An existing file
}
/* TODO: other open operations */
// add the page file to the environment
tdbEnvRgstPageFile(pEnv, pPgFile);
pPgFile->pEnv = pEnv;
*ppPgFile = pPgFile;
return 0;
}
int pgFileClose(SPgFile *pPgFile) {
if (pPgFile) {
if (pPgFile->fd >= 0) {
close(pPgFile->fd);
}
tfree(pPgFile->fname);
free(pPgFile);
}
return 0;
}
SPage *pgFileFetch(SPgFile *pPgFile, pgno_t pgno) {
SPgCache *pPgCache;
SPage * pPage;
pgid_t pgid;
// 1. Fetch from the page cache
// pgCacheFetch(pPgCache, pgid);
// 2. If only get a page frame, no content, maybe
// need to load from the file
if (1 /*page not initialized*/) {
if (pgno < pPgFile->fsize) {
// load the page content from the disk
// ?? How about the freed pages ??
} else {
// zero the page, make the page as a empty
// page with zero records.
}
}
#if 0
pPgCache = pPgFile->pPgCache;
pPage = NULL;
memcpy(pgid.fileid, pPgFile->fileid, TDB_FILE_ID_LEN);
pgid.pgno = pgno;
if (pgno > pPgFile->pgFileSize) {
// TODO
} else {
pPage = pgCacheFetch(pPgCache, pgid);
if (1 /*Page is cached, no need to load from file*/) {
return pPage;
} else {
// TODO: handle error
if (pgFileRead(pPgFile, pgno, (void *)pPage) < 0) {
// todoerr
}
return pPage;
}
}
#endif
return pPage;
}
int pgFileRelease(SPage *pPage) {
pgCacheRelease(pPage);
return 0;
}
int pgFileWrite(SPage *pPage) {
// TODO
return 0;
}
int pgFileAllocatePage(SPgFile *pPgFile, pgno_t *pPgno) {
pgno_t pgno;
SPage1 * pPage1;
SPgCache *pPgCache;
pgid_t pgid;
SPage * pPage;
if (pPgFile->lsize == 0) {
pgno = ++(pPgFile->lsize);
} else {
if (0) {
// TODO: allocate from the free list
pPage = pgCacheFetch(pPgCache, pgid);
if (pPage1->nFree > 0) {
// TODO
} else {
pgno = ++(pPgFile->lsize);
}
} else {
pgno = ++(pPgFile->lsize);
}
}
*pPgno = pgno;
return 0;
}
static int pgFileRead(SPgFile *pPgFile, pgno_t pgno, uint8_t *pData) {
pgsz_t pgSize;
ssize_t rsize;
uint8_t *pTData;
size_t szToRead;
#if 0
// pgSize = ; (TODO)
pTData = pData;
szToRead = pgSize;
for (; szToRead > 0;) {
rsize = pread(pPgFile->fd, pTData, szToRead, pgno * pgSize);
if (rsize < 0) {
if (errno == EINTR) {
continue;
} else {
return -1;
}
} else if (rsize == 0) {
return -1;
}
szToRead -= rsize;
pTData += rsize;
}
#endif
return 0;
}

67
source/libs/tdb/src/db/tdbUtil.c Normal file
View File

@ -0,0 +1,67 @@
/*
* 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 "tdbInt.h"
int tdbGnrtFileID(const char *fname, uint8_t *fileid, bool unique) {
struct stat statbuf;
if (stat(fname, &statbuf) < 0) {
return -1;
}
memset(fileid, 0, TDB_FILE_ID_LEN);
((uint64_t *)fileid)[0] = (uint64_t)statbuf.st_ino;
((uint64_t *)fileid)[1] = (uint64_t)statbuf.st_dev;
if (unique) {
((uint64_t *)fileid)[2] = rand();
}
return 0;
}
int tdbCheckFileAccess(const char *pathname, int mode) {
int flags = 0;
if (mode & TDB_F_OK) {
flags |= F_OK;
}
if (mode & TDB_R_OK) {
flags |= R_OK;
}
if (mode & TDB_W_OK) {
flags |= W_OK;
}
return access(pathname, flags);
}
int tdbGetFileSize(const char *fname, pgsz_t pgSize, pgno_t *pSize) {
struct stat st;
int ret;
ret = stat(fname, &st);
if (ret != 0) {
return -1;
}
ASSERT(st.st_size % pgSize == 0);
*pSize = st.st_size / pgSize;
return 0;
}

44
source/libs/tdb/src/inc/tdbBtree.h Normal file
View File

@ -0,0 +1,44 @@
/*
* 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_BTREE_H_
#define _TD_BTREE_H_
#ifdef __cplusplus
extern "C" {
#endif
typedef struct SBTree SBTree;
typedef struct SBtCursor SBtCursor;
// SBTree
int btreeOpen(SBTree **ppBt, SPgFile *pPgFile);
int btreeClose(SBTree *pBt);
// SBtCursor
int btreeCursorOpen(SBtCursor *pBtCur, SBTree *pBt);
int btreeCursorClose(SBtCursor *pBtCur);
int btreeCursorMoveTo(SBtCursor *pBtCur, int kLen, const void *pKey);
int btreeCursorNext(SBtCursor *pBtCur);
struct SBTree {
pgno_t root;
};
#ifdef __cplusplus
}
#endif
#endif /*_TD_BTREE_H_*/

23
source/libs/tdb/src/db/tdb_db.c → source/libs/tdb/src/inc/tdbEnv.h
View File

@ -13,14 +13,21 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tdb_db.h"
#ifndef _TDB_ENV_H_
#define _TDB_ENV_H_
int tdbOpen(TDB **dbpp, const char *fname, const char *dbname, uint32_t flags) {
// TODO
return 0;
#ifdef __cplusplus
extern "C" {
#endif
const char* tdbEnvGetRootDir(TENV* pEnv);
SPgFile* tdbEnvGetPageFile(TENV* pEnv, const uint8_t fileid[]);
SPgCache* tdbEnvGetPgCache(TENV* pEnv);
int tdbEnvRgstPageFile(TENV* pEnv, SPgFile* pPgFile);
int tdbEnvRgstDB(TENV* pEnv, TDB* pDb);
#ifdef __cplusplus
}
#endif
int tdbClose(TDB *dbp, uint32_t flags) {
// TODO
return 0;
}
#endif /*_TDB_ENV_H_*/

134
source/libs/tdb/src/inc/tdbInt.h Normal file
View File

@ -0,0 +1,134 @@
/*
* 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_TDB_INTERNAL_H_
#define _TD_TDB_INTERNAL_H_
#include "tlist.h"
#include "tlockfree.h"
#include "tdb.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct SPgFile SPgFile;
// pgno_t
typedef int32_t pgno_t;
#define TDB_IVLD_PGNO ((pgno_t)0)
// fileid
#define TDB_FILE_ID_LEN 24
// pgid_t
typedef struct {
uint8_t fileid[TDB_FILE_ID_LEN];
pgno_t pgno;
} pgid_t;
#define TDB_IVLD_PGID (pgid_t){0, TDB_IVLD_PGNO};
static FORCE_INLINE int tdbCmprPgId(const void *p1, const void *p2) {
pgid_t *pgid1 = (pgid_t *)p1;
pgid_t *pgid2 = (pgid_t *)p2;
int rcode;
rcode = memcmp(pgid1->fileid, pgid2->fileid, TDB_FILE_ID_LEN);
if (rcode) {
return rcode;
} else {
if (pgid1->pgno > pgid2->pgno) {
return 1;
} else if (pgid1->pgno < pgid2->pgno) {
return -1;
} else {
return 0;
}
}
}
// framd_id_t
typedef int32_t frame_id_t;
// pgsz_t
#define TDB_MIN_PGSIZE 512
#define TDB_MAX_PGSIZE 65536
#define TDB_DEFAULT_PGSIZE 4096
#define TDB_IS_PGSIZE_VLD(s) (((s) >= TDB_MIN_PGSIZE) && ((s) <= TDB_MAX_PGSIZE))
// pgoff_t
typedef pgsz_t pgoff_t;
// cache
#define TDB_DEFAULT_CACHE_SIZE (256 * 4096) // 1M
// dbname
#define TDB_MAX_DBNAME_LEN 24
// tdb_log
#define tdbError(var)
typedef TD_DLIST(STDb) STDbList;
typedef TD_DLIST(SPgFile) SPgFileList;
typedef TD_DLIST_NODE(SPgFile) SPgFileListNode;
#define TERR_A(val, op, flag) \
do { \
if (((val) = (op)) != 0) { \
goto flag; \
} \
} while (0)
#define TERR_B(val, op, flag) \
do { \
if (((val) = (op)) == NULL) { \
goto flag; \
} \
} while (0)
#define TDB_VARIANT_LEN (int)-1
// page payload format
// <keyLen> + <valLen> + [key] + [value]
#define TDB_DECODE_PAYLOAD(pPayload, keyLen, pKey, valLen, pVal) \
do { \
if ((keyLen) == TDB_VARIANT_LEN) { \
/* TODO: decode the keyLen */ \
} \
if ((valLen) == TDB_VARIANT_LEN) { \
/* TODO: decode the valLen */ \
} \
/* TODO */ \
} while (0)
#define TDB_JOURNAL_NAME "tdb.journal"
#include "tdbUtil.h"
#include "tdbBtree.h"
#include "tdbPgCache.h"
#include "tdbPgFile.h"
#include "tdbEnv.h"
#ifdef __cplusplus
}
#endif
#endif /*_TD_TDB_INTERNAL_H_*/

48
source/libs/tdb/src/inc/tdbPgCache.h Normal file
View File

@ -0,0 +1,48 @@
/*
* 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_PAGE_CACHE_H_
#define _TD_PAGE_CACHE_H_
#ifdef __cplusplus
extern "C" {
#endif
typedef struct SPgCache SPgCache;
typedef struct SPage SPage;
// SPgCache
int pgCacheOpen(SPgCache **ppPgCache, TENV *pEnv);
int pgCacheClose(SPgCache *pPgCache);
SPage *pgCacheFetch(SPgCache *pPgCache, pgid_t pgid);
int pgCacheRelease(SPage *pPage);
// SPage
typedef TD_DLIST_NODE(SPage) SPgListNode;
struct SPage {
pgid_t pgid; // page id
frame_id_t frameid; // frame id
uint8_t * pData; // real data
SPgListNode freeNode; // for SPgCache.freeList
SPgListNode pghtNode; // for pght
SPgListNode lruNode; // for LRU
};
#ifdef __cplusplus
}
#endif
#endif /*_TD_PAGE_CACHE_H_*/

59
source/libs/tdb/src/inc/tdbPgFile.h Normal file
View File

@ -0,0 +1,59 @@
/*
* 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_PAGE_FILE_H_
#define _TD_PAGE_FILE_H_
#ifdef __cplusplus
extern "C" {
#endif
typedef struct __attribute__((__packed__)) {
char hdrInfo[16]; // info string
pgsz_t szPage; // page size of current file
int32_t cno; // commit number counter
pgno_t freePgno; // freelist page number
uint8_t resv[100]; // reserved space
} SPgFileHdr;
#define TDB_PG_FILE_HDR_SIZE 128
TDB_STATIC_ASSERT(sizeof(SPgFileHdr) == TDB_PG_FILE_HDR_SIZE, "Page file header size if not 128");
struct SPgFile {
TENV * pEnv; // env containing this page file
char * fname; // backend file name
uint8_t fileid[TDB_FILE_ID_LEN]; // file id
pgno_t lsize; // page file logical size (for count)
pgno_t fsize; // real file size on disk (for rollback)
int fd;
SPgFileListNode envHash;
SPgFileListNode envPgfList;
};
int pgFileOpen(SPgFile **ppPgFile, const char *fname, TENV *pEnv);
int pgFileClose(SPgFile *pPgFile);
SPage *pgFileFetch(SPgFile *pPgFile, pgno_t pgno);
int pgFileRelease(SPage *pPage);
int pgFileWrite(SPage *pPage);
int pgFileAllocatePage(SPgFile *pPgFile, pgno_t *pPgno);
#ifdef __cplusplus
}
#endif
#endif /*_TD_PAGE_FILE_H_*/

44
source/libs/tdb/src/inc/tdb_inc.h → source/libs/tdb/src/inc/tdbUtil.h
View File

@ -13,46 +13,32 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_TDB_INC_H_
#define _TD_TDB_INC_H_
#include "os.h"
#include "tlist.h"
#include "tlockfree.h"
#ifndef _TDB_UTIL_H_
#define _TDB_UTIL_H_
#ifdef __cplusplus
extern "C" {
#endif
// pgno_t
typedef int32_t pgno_t;
#define TDB_IVLD_PGNO ((pgno_t)-1)
#if __STDC_VERSION__ >= 201112L
#define TDB_STATIC_ASSERT(op, info) static_assert(op, info)
#else
#define TDB_STATIC_ASSERT(op, info)
#endif
// fileid
#define TDB_FILE_ID_LEN 24
#define TDB_ROUND8(x) (((x) + 7) & ~7)
// pgid_t
typedef struct {
uint8_t fileid[TDB_FILE_ID_LEN];
pgno_t pgno;
} pgid_t;
#define TDB_IVLD_PGID (pgid_t){0, TDB_IVLD_PGNO};
int tdbGnrtFileID(const char *fname, uint8_t *fileid, bool unique);
// framd_id_t
typedef int32_t frame_id_t;
#define TDB_F_OK 0x1
#define TDB_R_OK 0x2
#define TDB_W_OK 0x4
int tdbCheckFileAccess(const char *pathname, int mode);
// pgsize_t
typedef int32_t pgsize_t;
#define TDB_MIN_PGSIZE 512
#define TDB_MAX_PGSIZE 16384
#define TDB_DEFAULT_PGSIZE 4096
#define TDB_IS_PGSIZE_VLD(s) (((s) >= TDB_MIN_PGSIZE) && ((s) <= TDB_MAX_PGSIZE))
// tdb_log
#define tdbError(var)
int tdbGetFileSize(const char *fname, pgsz_t pgSize, pgno_t *pSize);
#ifdef __cplusplus
}
#endif
#endif /*_TD_TDB_INC_H_*/
#endif /*_TDB_UTIL_H_*/

94
source/libs/tdb/src/inc/tdb_mpool.h
View File

@ -1,94 +0,0 @@
/*
* 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_TDB_MPOOL_H_
#define _TD_TDB_MPOOL_H_
#include "tdb_inc.h"
#ifdef __cplusplus
extern "C" {
#endif
// Exposed handle
typedef struct TDB_MPOOL TDB_MPOOL;
typedef struct TDB_MPFILE TDB_MPFILE;
typedef TD_DLIST_NODE(pg_t) pg_free_dlist_node_t, pg_hash_dlist_node_t;
typedef struct pg_t {
SRWLatch rwLatch;
frame_id_t frameid;
pgid_t pgid;
uint8_t dirty;
uint8_t rbit;
int32_t pinRef;
pg_free_dlist_node_t free;
pg_hash_dlist_node_t hash;
void * p;
} pg_t;
typedef TD_DLIST(pg_t) pg_list_t;
typedef struct {
SRWLatch latch;
TD_DLIST(TDB_MPFILE);
} mpf_bucket_t;
struct TDB_MPOOL {
int64_t cachesize;
pgsize_t pgsize;
int32_t npages;
pg_t * pages;
pg_list_t freeList;
frame_id_t clockHand;
struct {
int32_t nbucket;
pg_list_t *hashtab;
} pgtab; // page table, hash<pgid_t, pg_t>
struct {
#define MPF_HASH_BUCKETS 16
mpf_bucket_t buckets[MPF_HASH_BUCKETS];
} mpfht; // MPF hash table. MPFs using this MP will be put in this hash table
};
#define MP_PAGE_AT(mp, idx) (mp)->pages[idx]
typedef TD_DLIST_NODE(TDB_MPFILE) td_mpf_dlist_node_t;
struct TDB_MPFILE {
char * fname; // file name
int fd; // fd
uint8_t fileid[TDB_FILE_ID_LEN]; // file ID
TDB_MPOOL * mp; // underlying memory pool
td_mpf_dlist_node_t node;
};
/*=================================================== Exposed apis ==================================================*/
// TDB_MPOOL
int tdbMPoolOpen(TDB_MPOOL **mpp, uint64_t cachesize, pgsize_t pgsize);
int tdbMPoolClose(TDB_MPOOL *mp);
int tdbMPoolSync(TDB_MPOOL *mp);
// TDB_MPFILE
int tdbMPoolFileOpen(TDB_MPFILE **mpfp, const char *fname, TDB_MPOOL *mp);
int tdbMPoolFileClose(TDB_MPFILE *mpf);
int tdbMPoolFileNewPage(TDB_MPFILE *mpf, pgno_t *pgno, void *addr);
int tdbMPoolFileFreePage(TDB_MPOOL *mpf, pgno_t *pgno, void *addr);
int tdbMPoolFileGetPage(TDB_MPFILE *mpf, pgno_t pgno, void *addr);
int tdbMPoolFilePutPage(TDB_MPFILE *mpf, pgno_t pgno, void *addr);
int tdbMPoolFileSync(TDB_MPFILE *mpf);
#ifdef __cplusplus
}
#endif
#endif /*_TD_TDB_MPOOL_H_*/

4
source/libs/tdb/test/CMakeLists.txt
View File

@ -1,7 +1,3 @@
# tdbMPoolTest
add_executable(tdbMPoolTest "tdbMPoolTest.cpp")
target_link_libraries(tdbMPoolTest tdb gtest gtest_main)
# tdbTest
add_executable(tdbTest "tdbTest.cpp")
target_link_libraries(tdbTest tdb gtest gtest_main)

31
source/libs/tdb/test/tdbMPoolTest.cpp
View File

@ -1,31 +0,0 @@
#include "gtest/gtest.h"
#include <iostream>
#include "tdb_mpool.h"
TEST(tdb_mpool_test, test1) {
TDB_MPOOL * mp;
TDB_MPFILE *mpf;
pgno_t pgno;
void * pgdata;
// open mp
tdbMPoolOpen(&mp, 16384, 4096);
// open mpf
tdbMPoolFileOpen(&mpf, "test.db", mp);
#define TEST1_TOTAL_PAGES 100
for (int i = 0; i < TEST1_TOTAL_PAGES; i++) {
tdbMPoolFileNewPage(mpf, &pgno, pgdata);
*(pgno_t *)pgdata = i;
}
// close mpf
tdbMPoolFileClose(mpf);
// close mp
tdbMPoolClose(mp);
}

66
source/libs/tdb/test/tdbTest.cpp
View File

@ -2,13 +2,67 @@
#include "tdb.h"
TEST(tdb_api_test, tdb_create_open_close_db_test) {
// int ret;
// TDB *dbp;
TEST(tdb_test, simple_test) {
TENV * pEnv;
TDB * pDb1, *pDb2, *pDb3;
pgsz_t pgSize = 1024;
cachesz_t cacheSize = 10240;
// tdbCreateDB(&dbp, TDB_BTREE_T);
// ENV
GTEST_ASSERT_EQ(tdbEnvCreate(&pEnv, "./testtdb"), 0);
// tdbOpenDB(dbp, 0);
GTEST_ASSERT_EQ(tdbEnvSetCache(pEnv, pgSize, cacheSize), 0);
// tdbCloseDB(dbp, 0);
GTEST_ASSERT_EQ(tdbEnvGetCacheSize(pEnv), cacheSize);
GTEST_ASSERT_EQ(tdbEnvGetPageSize(pEnv), pgSize);
GTEST_ASSERT_EQ(tdbEnvOpen(pEnv), 0);
#if 1
// DB
GTEST_ASSERT_EQ(tdbCreate(&pDb1), 0);
// GTEST_ASSERT_EQ(tdbSetKeyLen(pDb1, 8), 0);
// GTEST_ASSERT_EQ(tdbGetKeyLen(pDb1), 8);
// GTEST_ASSERT_EQ(tdbSetValLen(pDb1, 3), 0);
// GTEST_ASSERT_EQ(tdbGetValLen(pDb1), 3);
// GTEST_ASSERT_EQ(tdbSetDup(pDb1, 1), 0);
// GTEST_ASSERT_EQ(tdbGetDup(pDb1), 1);
// GTEST_ASSERT_EQ(tdbSetCmprFunc(pDb1, NULL), 0);
tdbEnvBeginTxn(pEnv);
GTEST_ASSERT_EQ(tdbOpen(pDb1, "db.db", "db1", pEnv), 0);
// char *key = "key1";
// char *val = "value1";
// tdbInsert(pDb1, (void *)key, strlen(key), (void *)val, strlen(val));
tdbEnvCommit(pEnv);
#if 0
// Insert
// Query
// Delete
// Query
#endif
// GTEST_ASSERT_EQ(tdbOpen(&pDb2, "db.db", "db2", pEnv), 0);
// GTEST_ASSERT_EQ(tdbOpen(&pDb3, "index.db", NULL, pEnv), 0);
// tdbClose(pDb3);
// tdbClose(pDb2);
tdbClose(pDb1);
#endif
tdbEnvClose(pEnv);
}

2
source/util/CMakeLists.txt
View File

@ -10,7 +10,7 @@ target_link_libraries(
util
PRIVATE os
PUBLIC lz4_static
PUBLIC api
PUBLIC api cjson
)
if(${BUILD_TEST})

76
source/util/src/tjson.c Normal file
View File

@ -0,0 +1,76 @@
/*
* 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 "tjson.h"
#include "taoserror.h"
#include "cJSON.h"
SJson* tjsonCreateObject() {
return cJSON_CreateObject();
}
void tjsonDelete(SJson* pJson) {
cJSON_Delete((cJSON*)pJson);
}
int32_t tjsonAddIntegerToObject(SJson* pJson, const char* pName, const uint64_t number) {
char tmp[40] = {0};
snprintf(tmp, tListLen(tmp), "%"PRId64, number);
return tjsonAddStringToObject(pJson, pName, tmp);
}
int32_t tjsonAddStringToObject(SJson* pJson, const char* pName, const char* pVal) {
return (NULL == cJSON_AddStringToObject((cJSON*)pJson, pName, pVal) ? TSDB_CODE_FAILED : TSDB_CODE_SUCCESS);
}
SJson* tjsonAddArrayToObject(SJson* pJson, const char* pName) {
return cJSON_AddArrayToObject((cJSON*)pJson, pName);
}
int32_t tjsonAddItemToObject(SJson *pJson, const char* pName, SJson* pItem) {
return (cJSON_AddItemToObject((cJSON*)pJson, pName, pItem) ? TSDB_CODE_SUCCESS : TSDB_CODE_FAILED);
}
int32_t tjsonAddItemToArray(SJson* pJson, SJson* pItem) {
return (cJSON_AddItemToArray((cJSON*)pJson, pItem) ? TSDB_CODE_SUCCESS : TSDB_CODE_FAILED);
}
int32_t tjsonAddObject(SJson* pJson, const char* pName, FToJson func, const void* pObj) {
if (NULL == pObj) {
return TSDB_CODE_SUCCESS;
}
SJson* pJobj = tjsonCreateObject();
if (NULL == pJobj || TSDB_CODE_SUCCESS != func(pObj, pJobj)) {
printf("%s:%d code = %d\n", __FUNCTION__, __LINE__, TSDB_CODE_FAILED);
tjsonDelete(pJobj);
return TSDB_CODE_FAILED;
}
return tjsonAddItemToObject(pJson, pName, pJobj);
}
int32_t tjsonAddItem(SJson* pJson, FToJson func, const void* pObj) {
SJson* pJobj = tjsonCreateObject();
if (NULL == pJobj || TSDB_CODE_SUCCESS != func(pObj, pJobj)) {
tjsonDelete(pJobj);
return TSDB_CODE_FAILED;
}
return tjsonAddItemToArray(pJson, pJobj);
}
char* tjsonToString(const SJson* pJson) {
return cJSON_Print((cJSON*)pJson);
}