Skip to content

T
TDengine

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

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 85a16137 编写于 作者: D dapan
浏览文件
操作

Merge remote-tracking branch 'origin/1.0' into feature/qnode

上级 30556c67 2942658d
展开全部 显示空白变更内容
内联 并排
Showing with 18869 addition and 1243 deletion
include/common/tdataformat.h
浏览文件 @ 85a16137
......@@ -38,7 +38,6 @@ extern "C" {
memcpy(varDataVal(x), (str), __len); \
} while (0);
#define STR_WITH_MAXSIZE_TO_VARSTR(x, str, _maxs) \
do { \
char *_e = stpncpy(varDataVal(x), (str), (_maxs)-VARSTR_HEADER_SIZE); \
......@@ -73,7 +72,8 @@ typedef struct {
typedef struct {
int version; // version
int numOfCols; // Number of columns appended
int tlen; // maximum length of a SDataRow without the header part (sizeof(VarDataOffsetT) + sizeof(VarDataLenT) + (bytes))
int tlen; // maximum length of a SDataRow without the header part (sizeof(VarDataOffsetT) + sizeof(VarDataLenT) +
// (bytes))
uint16_t flen; // First part length in a SDataRow after the header part
uint16_t vlen; // pure value part length, excluded the overhead (bytes only)
STColumn columns[];
......@@ -202,7 +202,6 @@ void tdFreeDataRow(SDataRow row);
void tdInitDataRow(SDataRow row, STSchema *pSchema);
SDataRow tdDataRowDup(SDataRow row);
// offset here not include dataRow header length
static FORCE_INLINE int tdAppendDataColVal(SDataRow row, const void *value, bool isCopyVarData, int8_t type,
int32_t offset) {
......@@ -228,7 +227,6 @@ static FORCE_INLINE int tdAppendDataColVal(SDataRow row, const void *value, bool
return 0;
}
// offset here not include dataRow header length
static FORCE_INLINE int tdAppendColVal(SDataRow row, const void *value, int8_t type, int32_t offset) {
return tdAppendDataColVal(row, value, true, type, offset);
......@@ -269,7 +267,8 @@ static FORCE_INLINE void tdSetColOfRowNullBySchema(SDataRow row, STSchema *pSche
setNull(tdGetRowDataOfCol(row, type, offset), type, bytes);
}
static FORCE_INLINE void tdCopyColOfRowBySchema(SDataRow dst, STSchema *pDstSchema, int dstIdx, SDataRow src, STSchema *pSrcSchema, int srcIdx) {
static FORCE_INLINE void tdCopyColOfRowBySchema(SDataRow dst, STSchema *pDstSchema, int dstIdx, SDataRow src,
STSchema *pSrcSchema, int srcIdx) {
int8_t type = pDstSchema->columns[dstIdx].type;
assert(type == pSrcSchema->columns[srcIdx].type);
void *pData = tdGetPtrToCol(dst, pDstSchema, dstIdx);
......@@ -319,7 +318,6 @@ static FORCE_INLINE void tdCopyColOfRowBySchema(SDataRow dst, STSchema *pDstSche
}
}
// ----------------- Data column structure
typedef struct SDataCol {
int8_t type; // column type
......@@ -454,6 +452,7 @@ typedef struct {
#define kvRowValLen(r) (kvRowLen(r) - TD_KV_ROW_HEAD_SIZE - sizeof(SColIdx) * kvRowNCols(r))
#define kvRowTKey(r) (*(TKEY *)(kvRowValues(r)))
#define kvRowKey(r) tdGetKey(kvRowTKey(r))
#define kvRowKeys(r) POINTER_SHIFT(r, *(uint16_t *)POINTER_SHIFT(r, TD_KV_ROW_HEAD_SIZE + sizeof(int16_t)))
#define kvRowDeleted(r) TKEY_IS_DELETED(kvRowTKey(r))
SKVRow tdKVRowDup(SKVRow row);
......@@ -547,7 +546,7 @@ SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder);
static FORCE_INLINE int tdAddColToKVRow(SKVRowBuilder *pBuilder, int16_t colId, int8_t type, const void *value) {
if (pBuilder->nCols >= pBuilder->tCols) {
pBuilder->tCols *= 2;
SColIdx* pColIdx = (SColIdx *)realloc((void *)(pBuilder->pColIdx), sizeof(SColIdx) * pBuilder->tCols);
SColIdx *pColIdx = (SColIdx *)realloc((void *)(pBuilder->pColIdx), sizeof(SColIdx) * pBuilder->tCols);
if (pColIdx == NULL) return -1;
pBuilder->pColIdx = pColIdx;
}
......@@ -562,7 +561,7 @@ static FORCE_INLINE int tdAddColToKVRow(SKVRowBuilder *pBuilder, int16_t colId,
while (tlen > pBuilder->alloc - pBuilder->size) {
pBuilder->alloc *= 2;
}
void* buf = realloc(pBuilder->buf, pBuilder->alloc);
void *buf = realloc(pBuilder->buf, pBuilder->alloc);
if (buf == NULL) return -1;
pBuilder->buf = buf;
}
......@@ -654,6 +653,7 @@ static FORCE_INLINE char *memRowEnd(SMemRow row) {
#define memRowTKey(r) (isDataRow(r) ? dataRowTKey(memRowDataBody(r)) : kvRowTKey(memRowKvBody(r)))
#define memRowKey(r) (isDataRow(r) ? dataRowKey(memRowDataBody(r)) : kvRowKey(memRowKvBody(r)))
#define memRowKeys(r) (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowKeys(memRowKvBody(r)))
#define memRowSetTKey(r, k) \
do { \
if (isDataRow(r)) { \
......@@ -750,10 +750,10 @@ static FORCE_INLINE void tdGetColAppendDeltaLen(const void *value, int8_t colTyp
typedef struct {
int16_t colId;
uint8_t colType;
char* colVal;
char * colVal;
} SColInfo;
static FORCE_INLINE void setSColInfo(SColInfo* colInfo, int16_t colId, uint8_t colType, char* colVal) {
static FORCE_INLINE void setSColInfo(SColInfo *colInfo, int16_t colId, uint8_t colType, char *colVal) {
colInfo->colId = colId;
colInfo->colType = colType;
colInfo->colVal = colVal;
......
include/common/tmsg.h
浏览文件 @ 85a16137
......@@ -244,11 +244,29 @@ typedef struct SSubmitBlk {
// Submit message for this TSDB
typedef struct SSubmitMsg {
SMsgHead header;
int64_t version;
int32_t length;
int32_t numOfBlocks;
char blocks[];
} SSubmitMsg;
typedef struct {
int32_t totalLen;
int32_t len;
SMemRow row;
} SSubmitBlkIter;
typedef struct {
int32_t totalLen;
int32_t len;
void* pMsg;
} SSubmitMsgIter;
int tInitSubmitMsgIter(SSubmitMsg* pMsg, SSubmitMsgIter* pIter);
int tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
int tInitSubmitBlkIter(SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
SMemRow tGetSubmitBlkNext(SSubmitBlkIter* pIter);
typedef struct {
int32_t index; // index of failed block in submit blocks
int32_t vnode; // vnode index of failed block
......@@ -917,13 +935,13 @@ typedef struct {
typedef struct {
int32_t dnodeId;
} SMCreateMnodeMsg, SMDropMnodeMsg, SDDropMnodeMsg;
} SMCreateMnodeReq, SMDropMnodeReq, SDDropMnodeReq;
typedef struct {
int32_t dnodeId;
int8_t replica;
SReplica replicas[TSDB_MAX_REPLICA];
} SDCreateMnodeMsg, SDAlterMnodeMsg;
} SDCreateMnodeReq, SDAlterMnodeReq;
typedef struct {
int32_t dnodeId;
......
include/dnode/vnode/tsdb/tsdb.h
浏览文件 @ 85a16137
......@@ -22,21 +22,45 @@
extern "C" {
#endif
typedef struct SDataStatis {
int16_t colId;
int64_t sum;
int64_t max;
int64_t min;
int16_t maxIndex;
int16_t minIndex;
int16_t numOfNull;
} SDataStatis;
typedef struct STable {
int32_t tid;
uint64_t uid;
STSchema *pSchema;
} STable;
#define TABLE_TID(t) (t)->tid
#define TABLE_UID(t) (t)->uid
// TYPES EXPOSED
typedef struct STsdb STsdb;
typedef struct STsdbCfg {
int8_t precision;
uint64_t lruCacheSize;
uint32_t keep0;
uint32_t keep1;
uint32_t keep2;
int32_t daysPerFile;
int32_t minRowsPerFileBlock;
int32_t maxRowsPerFileBlock;
int32_t keep;
int32_t keep1;
int32_t keep2;
int8_t update;
} STsdbCfg;
// STsdb
STsdb *tsdbOpen(const char *path, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF);
void tsdbClose(STsdb *);
void tsdbRemove(const char *path);
int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg);
int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg, SSubmitRsp *pRsp);
int tsdbPrepareCommit(STsdb *pTsdb);
int tsdbCommit(STsdb *pTsdb);
......
include/dnode/vnode/tsdb2/tsdb.h 0 → 100644
浏览文件 @ 85a16137
/*
* 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_TSDB_H_
#define _TD_TSDB_H_
#include <pthread.h>
#include <stdbool.h>
#include <stdint.h>
#include "common.h"
#include "taosdef.h"
#include "tarray.h"
#include "tdataformat.h"
#include "thash.h"
#include "tlist.h"
#include "tlockfree.h"
#include "tmsg.h"
#include "tname.h"
#ifdef __cplusplus
extern "C" {
#endif
#define TSDB_VERSION_MAJOR 1
#define TSDB_VERSION_MINOR 0
#define TSDB_INVALID_SUPER_TABLE_ID -1
#define TSDB_STATUS_COMMIT_START 1
#define TSDB_STATUS_COMMIT_OVER 2
#define TSDB_STATUS_COMMIT_NOBLOCK 3 // commit no block, need to be solved
// TSDB STATE DEFINITION
#define TSDB_STATE_OK 0x0
#define TSDB_STATE_BAD_META 0x1
#define TSDB_STATE_BAD_DATA 0x2
typedef struct SDataStatis {
int16_t colId;
int64_t sum;
int64_t max;
int64_t min;
int16_t maxIndex;
int16_t minIndex;
int16_t numOfNull;
} SDataStatis;
// --------- TSDB APPLICATION HANDLE DEFINITION
// --------- TSDB REPOSITORY CONFIGURATION DEFINITION
typedef struct {
int32_t tsdbId;
int32_t cacheBlockSize;
int32_t totalBlocks;
int32_t daysPerFile; // day per file sharding policy
int32_t keep; // day of data to keep
int32_t keep1;
int32_t keep2;
int32_t lruCacheSize;
int32_t minRowsPerFileBlock; // minimum rows per file block
int32_t maxRowsPerFileBlock; // maximum rows per file block
int8_t precision;
int8_t compression;
int8_t update;
int8_t cacheLastRow; // 0:no cache, 1: cache last row, 2: cache last NULL column 3: 1&2
} STsdbCfg;
#define CACHE_NO_LAST(c) ((c)->cacheLastRow == 0)
#define CACHE_LAST_ROW(c) (((c)->cacheLastRow & 1) > 0)
#define CACHE_LAST_NULL_COLUMN(c) (((c)->cacheLastRow & 2) > 0)
// --------- TSDB REPOSITORY USAGE STATISTICS
typedef struct {
int64_t totalStorage; // total bytes occupie
int64_t compStorage;
int64_t pointsWritten; // total data points written
} STsdbStat;
typedef struct STsdb STsdb;
STsdbCfg *tsdbGetCfg(const STsdb *repo);
// --------- TSDB REPOSITORY DEFINITION
// int32_t tsdbCreateRepo(int repoid);
// int32_t tsdbDropRepo(int repoid);
STsdb * tsdbOpen(STsdbCfg *pCfg, STsdbAppH *pAppH);
int tsdbClose(STsdb *repo, int toCommit);
int32_t tsdbConfigRepo(STsdb *repo, STsdbCfg *pCfg);
int tsdbGetState(STsdb *repo);
int8_t tsdbGetCompactState(STsdb *repo);
// --------- TSDB TABLE DEFINITION
typedef struct {
uint64_t uid; // the unique table ID
int32_t tid; // the table ID in the repository.
} STableId;
// --------- TSDB TABLE configuration
typedef struct {
ETableType type;
char * name;
STableId tableId;
int32_t sversion;
char * sname; // super table name
uint64_t superUid;
STSchema * schema;
STSchema * tagSchema;
SKVRow tagValues;
char * sql;
} STableCfg;
void tsdbClearTableCfg(STableCfg *config);
void *tsdbGetTableTagVal(const void *pTable, int32_t colId, int16_t type);
char *tsdbGetTableName(void *pTable);
#define TSDB_TABLEID(_table) ((STableId *)(_table))
#define TSDB_PREV_ROW 0x1
#define TSDB_NEXT_ROW 0x2
STableCfg *tsdbCreateTableCfgFromMsg(SMDCreateTableMsg *pMsg);
int tsdbCreateTable(STsdb *repo, STableCfg *pCfg);
int tsdbDropTable(STsdb *pRepo, STableId tableId);
int tsdbUpdateTableTagValue(STsdb *repo, SUpdateTableTagValMsg *pMsg);
uint32_t tsdbGetFileInfo(STsdb *repo, char *name, uint32_t *index, uint32_t eindex, int64_t *size);
// the TSDB repository info
typedef struct STsdbRepoInfo {
STsdbCfg tsdbCfg;
uint64_t version; // version of the repository
int64_t tsdbTotalDataSize; // the original inserted data size
int64_t tsdbTotalDiskSize; // the total disk size taken by this TSDB repository
// TODO: Other informations to add
} STsdbRepoInfo;
STsdbRepoInfo *tsdbGetStatus(STsdb *pRepo);
// the meter information report structure
typedef struct {
STableCfg tableCfg;
uint64_t version;
int64_t tableTotalDataSize; // In bytes
int64_t tableTotalDiskSize; // In bytes
} STableInfo;
// -- FOR INSERT DATA
/**
* Insert data to a table in a repository
* @param pRepo the TSDB repository handle
* @param pData the data to insert (will give a more specific description)
*
* @return the number of points inserted, -1 for failure and the error number is set
*/
int32_t tsdbInsertData(STsdb *repo, SSubmitMsg *pMsg, SShellSubmitRspMsg *pRsp);
// -- FOR QUERY TIME SERIES DATA
typedef void *TsdbQueryHandleT; // Use void to hide implementation details
#define BLOCK_LOAD_OFFSET_SEQ_ORDER 1
#define BLOCK_LOAD_TABLE_SEQ_ORDER 2
#define BLOCK_LOAD_TABLE_RR_ORDER 3
// query condition to build multi-table data block iterator
typedef struct STsdbQueryCond {
STimeWindow twindow;
int32_t order; // desc|asc order to iterate the data block
int64_t offset; // skip offset put down to tsdb
int32_t numOfCols;
SColumnInfo *colList;
bool loadExternalRows; // load external rows or not
int32_t type; // data block load type:
} STsdbQueryCond;
typedef struct STableData STableData;
typedef struct {
T_REF_DECLARE()
SRWLatch latch;
TSKEY keyFirst;
TSKEY keyLast;
int64_t numOfRows;
int32_t maxTables;
STableData **tData;
SList * actList;
SList * extraBuffList;
SList * bufBlockList;
int64_t pointsAdd; // TODO
int64_t storageAdd; // TODO
} SMemTable;
typedef struct {
SMemTable *mem;
SMemTable *imem;
SMemTable mtable;
SMemTable *omem;
} SMemSnapshot;
typedef struct SMemRef {
int32_t ref;
SMemSnapshot snapshot;
} SMemRef;
#if 0
typedef struct SFileBlockInfo {
int32_t numBlocksOfStep;
} SFileBlockInfo;
typedef struct {
void *pTable;
TSKEY lastKey;
} STableKeyInfo;
typedef struct {
uint32_t numOfTables;
SArray * pGroupList;
SHashObj *map; // speedup acquire the tableQueryInfo by table uid
} STableGroupInfo;
#define TSDB_BLOCK_DIST_STEP_ROWS 16
typedef struct {
uint16_t rowSize;
uint16_t numOfFiles;
uint32_t numOfTables;
uint64_t totalSize;
uint64_t totalRows;
int32_t maxRows;
int32_t minRows;
int32_t firstSeekTimeUs;
uint32_t numOfRowsInMemTable;
uint32_t numOfSmallBlocks;
SArray * dataBlockInfos;
} STableBlockDist;
/**
* Get the data block iterator, starting from position according to the query condition
*
* @param tsdb tsdb handle
* @param pCond query condition, including time window, result set order, and basic required columns for each block
* @param tableInfoGroup table object list in the form of set, grouped into different sets according to the
* group by condition
* @param qinfo query info handle from query processor
* @return
*/
TsdbQueryHandleT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, uint64_t qId,
SMemRef *pRef);
/**
* Get the last row of the given query time window for all the tables in STableGroupInfo object.
* Note that only one data block with only row will be returned while invoking retrieve data block function for
* all tables in this group.
*
* @param tsdb tsdb handle
* @param pCond query condition, including time window, result set order, and basic required columns for each block
* @param tableInfo table list.
* @return
*/
TsdbQueryHandleT tsdbQueryLastRow(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, uint64_t qId,
SMemRef *pRef);
TsdbQueryHandleT tsdbQueryCacheLast(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId,
SMemRef *pMemRef);
bool isTsdbCacheLastRow(TsdbQueryHandleT *pQueryHandle);
/**
* get the queried table object list
* @param pHandle
* @return
*/
SArray *tsdbGetQueriedTableList(TsdbQueryHandleT *pHandle);
/**
* get the group list according to table id from client
* @param tsdb
* @param pCond
* @param groupList
* @param qinfo
* @return
*/
TsdbQueryHandleT tsdbQueryRowsInExternalWindow(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList,
uint64_t qId, SMemRef *pRef);
/**
* get num of rows in mem table
*
* @param pHandle
* @return row size
*/
int64_t tsdbGetNumOfRowsInMemTable(TsdbQueryHandleT *pHandle);
/**
* move to next block if exists
*
* @param pQueryHandle
* @return
*/
bool tsdbNextDataBlock(TsdbQueryHandleT pQueryHandle);
/**
* Get current data block information
*
* @param pQueryHandle
* @param pBlockInfo
* @return
*/
void tsdbRetrieveDataBlockInfo(TsdbQueryHandleT *pQueryHandle, SDataBlockInfo *pBlockInfo);
/**
*
* Get the pre-calculated information w.r.t. current data block.
*
* In case of data block in cache, the pBlockStatis will always be NULL.
* If a block is not completed loaded from disk, the pBlockStatis will be NULL.
* @pBlockStatis the pre-calculated value for current data blocks. if the block is a cache block, always return 0
* @return
*/
int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT *pQueryHandle, SDataStatis **pBlockStatis);
/**
*
* The query condition with primary timestamp is passed to iterator during its constructor function,
* the returned data block must be satisfied with the time window condition in any cases,
* which means the SData data block is not actually the completed disk data blocks.
*
* @param pQueryHandle query handle
* @param pColumnIdList required data columns id list
* @return
*/
SArray *tsdbRetrieveDataBlock(TsdbQueryHandleT *pQueryHandle, SArray *pColumnIdList);
/**
* Get the qualified table id for a super table according to the tag query expression.
* @param stableid. super table sid
* @param pTagCond. tag query condition
*/
int32_t tsdbQuerySTableByTagCond(STsdb *tsdb, uint64_t uid, TSKEY key, const char *pTagCond, size_t len,
STableGroupInfo *pGroupList, SColIndex *pColIndex, int32_t numOfCols);
/**
* destroy the created table group list, which is generated by tag query
* @param pGroupList
*/
void tsdbDestroyTableGroup(STableGroupInfo *pGroupList);
/**
* create the table group result including only one table, used to handle the normal table query
*
* @param tsdb tsdbHandle
* @param uid table uid
* @param pGroupInfo the generated result
* @return
*/
int32_t tsdbGetOneTableGroup(STsdb *tsdb, uint64_t uid, TSKEY startKey, STableGroupInfo *pGroupInfo);
/**
*
* @param tsdb
* @param pTableIdList
* @param pGroupInfo
* @return
*/
int32_t tsdbGetTableGroupFromIdList(STsdb *tsdb, SArray *pTableIdList, STableGroupInfo *pGroupInfo);
/**
* clean up the query handle
* @param queryHandle
*/
void tsdbCleanupQueryHandle(TsdbQueryHandleT queryHandle);
void tsdbResetQueryHandle(TsdbQueryHandleT queryHandle, STsdbQueryCond *pCond);
void tsdbResetQueryHandleForNewTable(TsdbQueryHandleT queryHandle, STsdbQueryCond *pCond, STableGroupInfo *groupList);
int32_t tsdbGetFileBlocksDistInfo(TsdbQueryHandleT *queryHandle, STableBlockDist *pTableBlockInfo);
// obtain queryHandle attribute
int64_t tsdbSkipOffset(TsdbQueryHandleT queryHandle);
/**
* get the statistics of repo usage
* @param repo. point to the tsdbrepo
* @param totalPoints. total data point written
* @param totalStorage. total bytes took by the tsdb
* @param compStorage. total bytes took by the tsdb after compressed
*/
void tsdbReportStat(void *repo, int64_t *totalPoints, int64_t *totalStorage, int64_t *compStorage);
int tsdbInitCommitQueue();
void tsdbDestroyCommitQueue();
int tsdbSyncCommit(STsdb *repo);
void tsdbIncCommitRef(int vgId);
void tsdbDecCommitRef(int vgId);
void tsdbSwitchTable(TsdbQueryHandleT pQueryHandle);
// For TSDB file sync
int tsdbSyncSend(void *pRepo, SOCKET socketFd);
int tsdbSyncRecv(void *pRepo, SOCKET socketFd);
// // For TSDB Compact
// int tsdbCompact(STsdb *pRepo);
// For TSDB Health Monitor
// // no problem return true
// bool tsdbNoProblem(STsdb *pRepo);
// // unit of walSize: MB
// int tsdbCheckWal(STsdb *pRepo, uint32_t walSize);
// // for json tag
// void *getJsonTagValueElment(void *data, char *key, int32_t keyLen, char *out, int16_t bytes);
// void getJsonTagValueAll(void *data, void *dst, int16_t bytes);
// char *parseTagDatatoJson(void *p);
#endif
#ifdef __cplusplus
}
#endif
#endif // _TD_TSDB_H_
include/libs/catalog/catalog.h
浏览文件 @ 85a16137
......@@ -87,14 +87,14 @@ int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName,
/**
* Get a DB's all vgroup info.
* @param pCatalog (input, got with catalogGetHandle)
* @param pRpc (input, rpc object)
* @param pTransporter (input, rpc object)
* @param pMgmtEps (input, mnode EPs)
* @param pDBName (input, full db name)
* @param forceUpdate (input, force update db vgroup info from mnode)
* @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
* @return error code
*/
int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, bool forceUpdate, SArray** pVgroupList);
int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const char* pDBName, bool forceUpdate, SArray** pVgroupList);
int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo);
......@@ -149,13 +149,13 @@ int32_t catalogGetSTableMeta(struct SCatalog* pCatalog, void * pTransporter, con
/**
* Get a table's actual vgroup, for stable it's all possible vgroup list.
* @param pCatalog (input, got with catalogGetHandle)
* @param pRpc (input, rpc object)
* @param pTransporter (input, rpc object)
* @param pMgmtEps (input, mnode EPs)
* @param pTableName (input, table name, NOT including db name)
* @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
* @return error code
*/
int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList);
int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList);
/**
* Get a table's vgroup from its name's hash value.
......@@ -172,16 +172,16 @@ int32_t catalogGetTableHashVgroup(struct SCatalog* pCatalog, void * pTransporter
/**
* Get all meta data required in pReq.
* @param pCatalog (input, got with catalogGetHandle)
* @param pRpc (input, rpc object)
* @param pTransporter (input, rpc object)
* @param pMgmtEps (input, mnode EPs)
* @param pReq (input, reqest info)
* @param pRsp (output, response data)
* @return error code
*/
int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList);
int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, SArray* pQnodeList);
int32_t catalogGetExpiredSTables(struct SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num);
......
include/libs/parser/parsenodes.h
浏览文件 @ 85a16137
......@@ -135,9 +135,8 @@ typedef struct SQueryStmtInfo {
SArray *pUdfInfo;
struct SQueryStmtInfo *sibling; // sibling
struct SQueryStmtInfo *pDownstream;
SMultiFunctionsDesc info;
SArray *pUpstream; // SArray<struct SQueryStmtInfo>
SArray *pDownstream; // SArray<struct SQueryStmtInfo>
int32_t havingFieldNum;
int32_t exprListLevelIndex;
} SQueryStmtInfo;
......
include/libs/parser/parser.h
浏览文件 @ 85a16137
......@@ -24,7 +24,6 @@ extern "C" {
typedef struct SParseContext {
SParseBasicCtx ctx;
int8_t schemaAttached; // denote if submit block is built with table schema or not
const char *pSql; // sql string
size_t sqlLen; // length of the sql string
char *pMsg; // extended error message if exists to help identifying the problem in sql statement.
......@@ -41,8 +40,17 @@ typedef struct SParseContext {
*/
int32_t qParseQuerySql(SParseContext* pContext, SQueryNode** pQuery);
bool qIsDdlQuery(const SQueryNode* pQuery);
/**
* Return true if it is a ddl/dcl sql statement
* @param pQuery
* @return
*/
bool qIsDdlQuery(const SQueryNode* pQueryNode);
/**
* Destroy logic query plan
* @param pQueryNode
*/
void qDestroyQuery(SQueryNode* pQueryNode);
/**
......
include/libs/planner/planner.h
浏览文件 @ 85a16137
......@@ -52,6 +52,8 @@ typedef SSchema SSlotSchema;
typedef struct SDataBlockSchema {
SSlotSchema *pSchema;
int32_t numOfCols; // number of columns
int32_t resultRowSize;
int16_t precision;
} SDataBlockSchema;
typedef struct SQueryNodeBasicInfo {
......@@ -61,6 +63,7 @@ typedef struct SQueryNodeBasicInfo {
typedef struct SDataSink {
SQueryNodeBasicInfo info;
SDataBlockSchema schema;
} SDataSink;
typedef struct SDataDispatcher {
......@@ -139,8 +142,12 @@ typedef struct SQueryDag {
struct SQueryNode;
/**
/**
* Create the physical plan for the query, according to the AST.
* @param pQueryInfo
* @param pDag
* @param requestId
* @return
*/
int32_t qCreateQueryDag(const struct SQueryNode* pQueryInfo, struct SQueryDag** pDag, uint64_t requestId);
......
include/libs/scheduler/scheduler.h
浏览文件 @ 85a16137
......@@ -75,6 +75,12 @@ int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void
*/
int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void** pJob);
/**
* Fetch query result from the remote query executor
* @param pJob
* @param data
* @return
*/
int32_t scheduleFetchRows(void *pJob, void **data);
......@@ -85,6 +91,10 @@ int32_t scheduleFetchRows(void *pJob, void **data);
*/
int32_t scheduleCancelJob(void *pJob);
/**
* Free the query job
* @param pJob
*/
void scheduleFreeJob(void *pJob);
void schedulerDestroy(void);
......
include/libs/tfs/tfs.h 0 → 100644
浏览文件 @ 85a16137
/*
* 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_TFS_H
#define TD_TFS_H
#include "tglobal.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
int level;
int id;
} SDiskID;
#define TFS_UNDECIDED_LEVEL -1
#define TFS_UNDECIDED_ID -1
#define TFS_PRIMARY_LEVEL 0
#define TFS_PRIMARY_ID 0
#define TFS_MIN_LEVEL 0
#define TFS_MAX_LEVEL (TSDB_MAX_TIERS - 1)
// FS APIs ====================================
typedef struct {
int64_t tsize;
int64_t used;
int64_t avail;
} SFSMeta;
typedef struct {
int64_t size;
int64_t used;
int64_t free;
int16_t nAvailDisks; // # of Available disks
} STierMeta;
int tfsInit(SDiskCfg *pDiskCfg, int ndisk);
void tfsDestroy();
void tfsUpdateInfo(SFSMeta *pFSMeta, STierMeta *tierMetas, int8_t numLevels);
void tfsGetMeta(SFSMeta *pMeta);
void tfsAllocDisk(int expLevel, int *level, int *id);
const char *TFS_PRIMARY_PATH();
const char *TFS_DISK_PATH(int level, int id);
// TFILE APIs ====================================
typedef struct {
int level;
int id;
char rname[TSDB_FILENAME_LEN]; // REL name
char aname[TSDB_FILENAME_LEN]; // ABS name
} TFILE;
#define TFILE_LEVEL(pf) ((pf)->level)
#define TFILE_ID(pf) ((pf)->id)
#define TFILE_NAME(pf) ((pf)->aname)
#define TFILE_REL_NAME(pf) ((pf)->rname)
#define tfsopen(pf, flags) open(TFILE_NAME(pf), flags)
#define tfsclose(fd) close(fd)
#define tfsremove(pf) remove(TFILE_NAME(pf))
#define tfscopy(sf, df) taosCopyFile(TFILE_NAME(sf), TFILE_NAME(df))
#define tfsrename(sf, df) taosRename(TFILE_NAME(sf), TFILE_NAME(df))
void tfsInitFile(TFILE *pf, int level, int id, const char *bname);
bool tfsIsSameFile(const TFILE *pf1, const TFILE *pf2);
int tfsEncodeFile(void **buf, TFILE *pf);
void *tfsDecodeFile(void *buf, TFILE *pf);
void tfsbasename(const TFILE *pf, char *dest);
void tfsdirname(const TFILE *pf, char *dest);
// DIR APIs ====================================
int tfsMkdirAt(const char *rname, int level, int id);
int tfsMkdirRecurAt(const char *rname, int level, int id);
int tfsMkdir(const char *rname);
int tfsRmdir(const char *rname);
int tfsRename(char *orname, char *nrname);
typedef struct TDIR TDIR;
TDIR * tfsOpendir(const char *rname);
const TFILE *tfsReaddir(TDIR *tdir);
void tfsClosedir(TDIR *tdir);
#ifdef __cplusplus
}
#endif
#endif
include/os/os.h
浏览文件 @ 85a16137
......@@ -24,10 +24,13 @@ extern "C" {
#include <ctype.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <float.h>
#include <inttypes.h>
#include <limits.h>
#include <locale.h>
#include <math.h>
#include <regex.h>
#include <sched.h>
#include <setjmp.h>
#include <signal.h>
......@@ -44,6 +47,8 @@ extern "C" {
#include <unistd.h>
#include <wchar.h>
#include <wctype.h>
#include <wordexp.h>
#include <libgen.h>
#include <sys/mman.h>
......
include/os/osDef.h
浏览文件 @ 85a16137
......@@ -73,6 +73,12 @@ extern "C" {
#endif
#ifndef WINDOWS
#ifndef O_BINARY
#define O_BINARY 0
#endif
#endif
#define POINTER_SHIFT(p, b) ((void *)((char *)(p) + (b)))
#define POINTER_DISTANCE(p1, p2) ((char *)(p1) - (char *)(p2))
......
include/util/taoserror.h
浏览文件 @ 85a16137
......@@ -259,28 +259,24 @@ int32_t* taosGetErrno();
#define TSDB_CODE_DND_DNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0411)
#define TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0420)
#define TSDB_CODE_DND_MNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0421)
#define TSDB_CODE_DND_MNODE_ID_INVALID TAOS_DEF_ERROR_CODE(0, 0x0422)
#define TSDB_CODE_DND_MNODE_ID_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x0423)
#define TSDB_CODE_DND_MNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0424)
#define TSDB_CODE_DND_MNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0425)
#define TSDB_CODE_DND_MNODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0422)
#define TSDB_CODE_DND_MNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0423)
#define TSDB_CODE_DND_MNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0424)
#define TSDB_CODE_DND_QNODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0430)
#define TSDB_CODE_DND_QNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0431)
#define TSDB_CODE_DND_QNODE_ID_INVALID TAOS_DEF_ERROR_CODE(0, 0x0432)
#define TSDB_CODE_DND_QNODE_ID_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x0433)
#define TSDB_CODE_DND_QNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0434)
#define TSDB_CODE_DND_QNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0435)
#define TSDB_CODE_DND_QNODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0432)
#define TSDB_CODE_DND_QNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0433)
#define TSDB_CODE_DND_QNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0434)
#define TSDB_CODE_DND_SNODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0440)
#define TSDB_CODE_DND_SNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0441)
#define TSDB_CODE_DND_SNODE_ID_INVALID TAOS_DEF_ERROR_CODE(0, 0x0442)
#define TSDB_CODE_DND_SNODE_ID_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x0443)
#define TSDB_CODE_DND_SNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0444)
#define TSDB_CODE_DND_SNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0445)
#define TSDB_CODE_DND_SNODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0442)
#define TSDB_CODE_DND_SNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0443)
#define TSDB_CODE_DND_SNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0444)
#define TSDB_CODE_DND_BNODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0450)
#define TSDB_CODE_DND_BNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0451)
#define TSDB_CODE_DND_BNODE_ID_INVALID TAOS_DEF_ERROR_CODE(0, 0x0452)
#define TSDB_CODE_DND_BNODE_ID_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x0453)
#define TSDB_CODE_DND_BNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0454)
#define TSDB_CODE_DND_BNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0455)
#define TSDB_CODE_DND_BNODE_INVALID_OPTION TAOS_DEF_ERROR_CODE(0, 0x0452)
#define TSDB_CODE_DND_BNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0453)
#define TSDB_CODE_DND_BNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0454)
#define TSDB_CODE_DND_VNODE_TOO_MANY_VNODES TAOS_DEF_ERROR_CODE(0, 0x0460)
#define TSDB_CODE_DND_VNODE_READ_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0461)
#define TSDB_CODE_DND_VNODE_WRITE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0462)
......
source/client/inc/clientInt.h
浏览文件 @ 85a16137
......@@ -62,6 +62,7 @@ typedef struct SAppInstInfo {
SList *pConnList; // STscObj linked list
int64_t clusterId;
void *pTransporter;
SHeartBeatInfo hb;
} SAppInstInfo;
typedef struct SAppInfo {
......@@ -70,7 +71,7 @@ typedef struct SAppInfo {
char *ep;
int32_t pid;
int32_t numOfThreads;
SHeartBeatInfo hb;
SHashObj *pInstMap;
} SAppInfo;
......
source/client/src/clientImpl.c
浏览文件 @ 85a16137
......@@ -291,10 +291,10 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
nPrintTsc("%s", sql)
SRequestObj* pRequest = NULL;
SQueryNode* pQuery = NULL;
SQueryDag* pDag = NULL;
void* pJob = NULL;
SRequestObj *pRequest = NULL;
SQueryNode *pQuery = NULL;
SQueryDag *pDag = NULL;
void *pJob = NULL;
terrno = TSDB_CODE_SUCCESS;
CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
......
source/client/src/clientMain.c
浏览文件 @ 85a16137
#include "os.h"
#include "tref.h"
#include "trpc.h"
#include "clientInt.h"
#include "clientLog.h"
#include "query.h"
#include "tmsg.h"
#include "tglobal.h"
#include "tref.h"
#include "trpc.h"
#include "catalog.h"
#define TSC_VAR_NOT_RELEASE 1
#define TSC_VAR_RELEASED 0
......
source/client/test/clientTests.cpp
浏览文件 @ 85a16137
......@@ -485,6 +485,22 @@ TEST(testCase, drop_stable_Test) {
taos_close(pConn);
}
TEST(testCase, generated_request_id_test) {
SHashObj* phash = taosHashInit(10000, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
for (int32_t i = 0; i < 50000; ++i) {
uint64_t v = generateRequestId();
void* result = taosHashGet(phash, &v, sizeof(v));
if (result != nullptr) {
printf("0x%lx, index:%d\n", v, i);
}
assert(result == nullptr);
taosHashPut(phash, &v, sizeof(v), NULL, 0);
}
taosHashCleanup(phash);
}
// TEST(testCase, create_topic_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
......@@ -518,46 +534,55 @@ TEST(testCase, drop_stable_Test) {
// tmq_create_topic(pConn, "test_topic_1", sql, strlen(sql));
// taos_close(pConn);
//}
TEST(testCase, generated_request_id_test) {
SHashObj* phash = taosHashInit(10000, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
for (int32_t i = 0; i < 50000; ++i) {
uint64_t v = generateRequestId();
void* result = taosHashGet(phash, &v, sizeof(v));
if (result != nullptr) {
printf("0x%lx, index:%d\n", v, i);
}
assert(result == nullptr);
taosHashPut(phash, &v, sizeof(v), NULL, 0);
}
taosHashCleanup(phash);
}
// TEST(testCase, projection_query_tables) {
//TEST(testCase, insert_test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// ASSERT_EQ(pConn, nullptr);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "select * from t_2");
// pRes = taos_query(pConn, "insert into t_2 values(now, 1)");
// if (taos_errno(pRes) != 0) {
// printf("failed to create multiple tables, reason:%s\n", taos_errstr(pRes));
// taos_free_result(pRes);
// ASSERT_TRUE(false);
// }
//
// TAOS_ROW pRow = NULL;
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// int32_t numOfFields = taos_num_fields(pRes);
//
// char str[512] = {0};
// while((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//#endif
TEST(testCase, projection_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);
TAOS_RES* pRes = taos_query(pConn, "use test1");
if (taos_errno(pRes) != 0) {
printf("failed to use db, reason:%s", taos_errstr(pRes));
taos_free_result(pRes);
return;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "select * from tm0");
if (taos_errno(pRes) != 0) {
printf("failed to create multiple tables, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
ASSERT_TRUE(false);
}
TAOS_ROW pRow = NULL;
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
int32_t numOfFields = taos_num_fields(pRes);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
source/common/src/tmsg.c
浏览文件 @ 85a16137
......@@ -27,6 +27,64 @@
#undef TD_MSG_SEG_CODE_
#include "tmsgdef.h"
int tInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter) {
if (pMsg == NULL) {
terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
return -1;
}
pIter->totalLen = pMsg->length;
pIter->len = 0;
pIter->pMsg = pMsg;
if (pMsg->length <= sizeof(SSubmitMsg)) {
terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
return -1;
}
return 0;
}
int tGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
if (pIter->len == 0) {
pIter->len += sizeof(SSubmitMsg);
} else {
SSubmitBlk *pSubmitBlk = (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
pIter->len += (sizeof(SSubmitBlk) + pSubmitBlk->dataLen + pSubmitBlk->schemaLen);
}
if (pIter->len > pIter->totalLen) {
terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
*pPBlock = NULL;
return -1;
}
*pPBlock = (pIter->len == pIter->totalLen) ? NULL : (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
return 0;
}
int tInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter) {
if (pBlock->dataLen <= 0) return -1;
pIter->totalLen = pBlock->dataLen;
pIter->len = 0;
pIter->row = (SMemRow)(pBlock->data + pBlock->schemaLen);
return 0;
}
SMemRow tGetSubmitBlkNext(SSubmitBlkIter *pIter) {
SMemRow row = pIter->row;
if (pIter->len >= pIter->totalLen) {
return NULL;
} else {
pIter->len += memRowTLen(row);
if (pIter->len < pIter->totalLen) {
pIter->row = POINTER_SHIFT(row, memRowTLen(row));
}
return row;
}
}
int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) {
int tlen = 0;
tlen += taosEncodeSClientHbKey(buf, &pReq->connKey);
......
source/dnode/mgmt/impl/src/dndBnode.c
浏览文件 @ 85a16137
......@@ -268,7 +268,7 @@ int32_t dndProcessCreateBnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_BNODE_ID_INVALID;
terrno = TSDB_CODE_DND_BNODE_INVALID_OPTION;
dError("failed to create bnode since %s", terrstr());
return -1;
} else {
......@@ -281,7 +281,7 @@ int32_t dndProcessDropBnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_BNODE_ID_INVALID;
terrno = TSDB_CODE_DND_BNODE_INVALID_OPTION;
dError("failed to drop bnode since %s", terrstr());
return -1;
} else {
......
source/dnode/mgmt/impl/src/dndMnode.c
浏览文件 @ 85a16137
......@@ -305,25 +305,24 @@ static void dndBuildMnodeOpenOption(SDnode *pDnode, SMnodeOpt *pOption) {
memcpy(&pOption->replicas, pMgmt->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
}
static int32_t dndBuildMnodeOptionFromReq(SDnode *pDnode, SMnodeOpt *pOption, SDCreateMnodeMsg *pReq) {
static int32_t dndBuildMnodeOptionFromReq(SDnode *pDnode, SMnodeOpt *pOption, SDCreateMnodeReq *pCreate) {
dndInitMnodeOption(pDnode, pOption);
pOption->dnodeId = dndGetDnodeId(pDnode);
pOption->clusterId = dndGetClusterId(pDnode);
pOption->replica = pReq->replica;
pOption->replica = pCreate->replica;
pOption->selfIndex = -1;
for (int32_t i = 0; i < pReq->replica; ++i) {
for (int32_t i = 0; i < pCreate->replica; ++i) {
SReplica *pReplica = &pOption->replicas[i];
pReplica->id = pReq->replicas[i].id;
pReplica->port = pReq->replicas[i].port;
memcpy(pReplica->fqdn, pReq->replicas[i].fqdn, TSDB_FQDN_LEN);
pReplica->id = pCreate->replicas[i].id;
pReplica->port = pCreate->replicas[i].port;
memcpy(pReplica->fqdn, pCreate->replicas[i].fqdn, TSDB_FQDN_LEN);
if (pReplica->id == pOption->dnodeId) {
pOption->selfIndex = i;
}
}
if (pOption->selfIndex == -1) {
terrno = TSDB_CODE_DND_MNODE_ID_NOT_FOUND;
dError("failed to build mnode options since %s", terrstr());
return -1;
}
......@@ -423,8 +422,8 @@ static int32_t dndDropMnode(SDnode *pDnode) {
return 0;
}
static SDCreateMnodeMsg *dndParseCreateMnodeReq(SRpcMsg *pReq) {
SDCreateMnodeMsg *pCreate = pReq->pCont;
static SDCreateMnodeReq *dndParseCreateMnodeReq(SRpcMsg *pReq) {
SDCreateMnodeReq *pCreate = pReq->pCont;
pCreate->dnodeId = htonl(pCreate->dnodeId);
for (int32_t i = 0; i < pCreate->replica; ++i) {
pCreate->replicas[i].id = htonl(pCreate->replicas[i].id);
......@@ -435,51 +434,85 @@ static SDCreateMnodeMsg *dndParseCreateMnodeReq(SRpcMsg *pReq) {
}
int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDCreateMnodeMsg *pCreate = dndParseCreateMnodeReq(pReq);
SDCreateMnodeReq *pCreate = dndParseCreateMnodeReq(pReq);
if (pCreate->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_ID_INVALID;
if (pCreate->replica <= 1 || pCreate->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to create mnode since %s", terrstr());
return -1;
} else {
}
SMnodeOpt option = {0};
if (dndBuildMnodeOptionFromReq(pDnode, &option, pCreate) != 0) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to create mnode since %s", terrstr());
return -1;
}
return dndOpenMnode(pDnode, &option);
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) {
dndReleaseMnode(pDnode, pMnode);
terrno = TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED;
dError("failed to create mnode since %s", terrstr());
return -1;
}
dDebug("start to create mnode");
return dndOpenMnode(pDnode, &option);
}
int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDAlterMnodeMsg *pAlter = dndParseCreateMnodeReq(pReq);
SDAlterMnodeReq *pAlter = dndParseCreateMnodeReq(pReq);
if (pAlter->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_ID_INVALID;
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to alter mnode since %s", terrstr());
return -1;
}
SMnodeOpt option = {0};
if (dndBuildMnodeOptionFromReq(pDnode, &option, pAlter) != 0) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to alter mnode since %s", terrstr());
return -1;
}
if (dndAlterMnode(pDnode, &option) != 0) {
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) {
terrno = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
dError("failed to alter mnode since %s", terrstr());
return -1;
}
return dndWriteMnodeFile(pDnode);
dDebug("start to alter mnode");
int32_t code = dndAlterMnode(pDnode, &option);
dndReleaseMnode(pDnode, pMnode);
return code;
}
int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
SDDropMnodeMsg *pDrop = pReq->pCont;
SDDropMnodeReq *pDrop = pReq->pCont;
pDrop->dnodeId = htonl(pDrop->dnodeId);
if (pDrop->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_ID_INVALID;
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
dError("failed to drop mnode since %s", terrstr());
return -1;
}
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) {
terrno = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
dError("failed to drop mnode since %s", terrstr());
return -1;
} else {
return dndDropMnode(pDnode);
}
dDebug("start to drop mnode");
int32_t code = dndDropMnode(pDnode);
dndReleaseMnode(pDnode, pMnode);
return code;
}
static void dndProcessMnodeQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
......@@ -506,6 +539,7 @@ static void dndWriteMnodeMsgToWorker(SDnode *pDnode, SDnodeWorker *pWorker, SRpc
code = TSDB_CODE_OUT_OF_MEMORY;
} else {
code = dndWriteMsgToWorker(pWorker, pMsg, 0);
if (code != 0) code = terrno;
}
if (code != 0) {
......
source/dnode/mgmt/impl/src/dndQnode.c
浏览文件 @ 85a16137
......@@ -274,7 +274,7 @@ int32_t dndProcessCreateQnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_QNODE_ID_INVALID;
terrno = TSDB_CODE_DND_QNODE_INVALID_OPTION;
dError("failed to create qnode since %s", terrstr());
return -1;
} else {
......@@ -287,7 +287,7 @@ int32_t dndProcessDropQnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_QNODE_ID_INVALID;
terrno = TSDB_CODE_DND_QNODE_INVALID_OPTION;
dError("failed to drop qnode since %s", terrstr());
return -1;
} else {
......
source/dnode/mgmt/impl/src/dndSnode.c
浏览文件 @ 85a16137
......@@ -268,7 +268,7 @@ int32_t dndProcessCreateSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_SNODE_ID_INVALID;
terrno = TSDB_CODE_DND_SNODE_INVALID_OPTION;
dError("failed to create snode since %s", terrstr());
return -1;
} else {
......@@ -281,7 +281,7 @@ int32_t dndProcessDropSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
pMsg->dnodeId = htonl(pMsg->dnodeId);
if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_SNODE_ID_INVALID;
terrno = TSDB_CODE_DND_SNODE_INVALID_OPTION;
dError("failed to drop snode since %s", terrstr());
return -1;
} else {
......
source/dnode/mgmt/impl/src/dndVnodes.c
浏览文件 @ 85a16137
......@@ -197,6 +197,10 @@ static void dndCloseVnode(SDnode *pDnode, SVnodeObj *pVnode) {
dndFreeVnodeWriteQueue(pDnode, pVnode);
dndFreeVnodeApplyQueue(pDnode, pVnode);
dndFreeVnodeSyncQueue(pDnode, pVnode);
vnodeClose(pVnode->pImpl);
pVnode->pImpl = NULL;
free(pVnode->path);
free(pVnode->db);
free(pVnode);
......@@ -554,7 +558,7 @@ static void dndGenerateVnodeCfg(SCreateVnodeMsg *pCreate, SVnodeCfg *pCfg) {
pCfg->ttl = 4;
pCfg->keep = pCreate->daysToKeep0;
pCfg->isWeak = true;
pCfg->tsdbCfg.keep0 = pCreate->daysToKeep0;
pCfg->tsdbCfg.keep = pCreate->daysToKeep0;
pCfg->tsdbCfg.keep1 = pCreate->daysToKeep2;
pCfg->tsdbCfg.keep2 = pCreate->daysToKeep0;
pCfg->tsdbCfg.lruCacheSize = pCreate->cacheBlockSize;
......
source/dnode/mgmt/impl/test/bnode/dbnode.cpp
浏览文件 @ 85a16137
......@@ -34,7 +34,7 @@ TEST_F(DndTestBnode, 01_Create_Bnode) {
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_BNODE_ID_INVALID);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_BNODE_INVALID_OPTION);
}
{
......@@ -82,7 +82,7 @@ TEST_F(DndTestBnode, 01_Drop_Bnode) {
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_BNODE_ID_INVALID);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_BNODE_INVALID_OPTION);
}
{
......
source/dnode/mgmt/impl/test/mnode/qmnode.cpp → source/dnode/mgmt/impl/test/mnode/dmnode.cpp
浏览文件 @ 85a16137
......@@ -25,17 +25,20 @@ class DndTestMnode : public ::testing::Test {
Testbase DndTestMnode::test;
#if 0
TEST_F(DndTestMnode, 01_Create_Mnode) {
{
int32_t contLen = sizeof(SDCreateMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
pReq->replica = 1;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_ID_INVALID);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_INVALID_OPTION);
}
{
......@@ -43,10 +46,14 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 1;
pReq->replicas[0].id = htonl(2);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, 0);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_INVALID_OPTION);
}
{
......@@ -54,19 +61,13 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED);
}
test.Restart();
{
int32_t contLen = sizeof(SDCreateMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 2;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
pReq->replicas[1].id = htonl(1);
pReq->replicas[1].port = htonl(9114);
strcpy(pReq->replicas[1].fqdn, "localhost");
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -76,49 +77,48 @@ TEST_F(DndTestMnode, 01_Create_Mnode) {
TEST_F(DndTestMnode, 02_Alter_Mnode) {
{
int32_t contLen = sizeof(SDCreateMnodeReq);
int32_t contLen = sizeof(SDAlterMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
SDAlterMnodeReq* pReq = (SDAlterMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
pReq->replica = 1;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_ID_INVALID);
}
{
int32_t contLen = sizeof(SDCreateMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, 0);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_INVALID_OPTION);
}
{
int32_t contLen = sizeof(SDCreateMnodeReq);
int32_t contLen = sizeof(SDAlterMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
SDAlterMnodeReq* pReq = (SDAlterMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 1;
pReq->replicas[0].id = htonl(2);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_INVALID_OPTION);
}
test.Restart();
{
int32_t contLen = sizeof(SDCreateMnodeReq);
int32_t contLen = sizeof(SDAlterMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
SDAlterMnodeReq* pReq = (SDAlterMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 1;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED);
ASSERT_EQ(pRsp->code, 0);
}
}
......@@ -131,7 +131,7 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_ID_INVALID);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_INVALID_OPTION);
}
{
......@@ -156,28 +156,34 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_NOT_DEPLOYED);
}
test.Restart();
{
int32_t contLen = sizeof(SDDropMnodeReq);
int32_t contLen = sizeof(SDAlterMnodeReq);
SDDropMnodeReq* pReq = (SDDropMnodeReq*)rpcMallocCont(contLen);
SDAlterMnodeReq* pReq = (SDAlterMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 1;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_MNODE, pReq, contLen);
SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_MNODE_NOT_DEPLOYED);
}
{
int32_t contLen = sizeof(SDCreateMnodeReq);
SDCreateMnodeReq* pReq = (SDCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
pReq->replica = 2;
pReq->replicas[0].id = htonl(1);
pReq->replicas[0].port = htonl(9113);
strcpy(pReq->replicas[0].fqdn, "localhost");
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, 0);
}
}
\ No newline at end of file
#endif
\ No newline at end of file
source/dnode/mgmt/impl/test/qnode/dqnode.cpp
浏览文件 @ 85a16137
......@@ -34,7 +34,7 @@ TEST_F(DndTestQnode, 01_Create_Qnode) {
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_QNODE_ID_INVALID);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_QNODE_INVALID_OPTION);
}
{
......@@ -82,7 +82,7 @@ TEST_F(DndTestQnode, 02_Drop_Qnode) {
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_QNODE_ID_INVALID);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_QNODE_INVALID_OPTION);
}
{
......
source/dnode/mgmt/impl/test/snode/dsnode.cpp
浏览文件 @ 85a16137
......@@ -34,7 +34,7 @@ TEST_F(DndTestSnode, 01_Create_Snode) {
SRpcMsg* pRsp = test.SendReq(TDMT_DND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_SNODE_ID_INVALID);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_SNODE_INVALID_OPTION);
}
{
......@@ -82,7 +82,7 @@ TEST_F(DndTestSnode, 01_Drop_Snode) {
SRpcMsg* pRsp = test.SendReq(TDMT_DND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_SNODE_ID_INVALID);
ASSERT_EQ(pRsp->code, TSDB_CODE_DND_SNODE_INVALID_OPTION);
}
{
......
source/dnode/mgmt/impl/test/sut/src/sut.cpp
浏览文件 @ 85a16137
......@@ -16,7 +16,7 @@
#include "sut.h"
void Testbase::InitLog(const char* path) {
dDebugFlag = 0;
dDebugFlag = 143;
vDebugFlag = 0;
mDebugFlag = 143;
cDebugFlag = 0;
......
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @ 85a16137
......@@ -27,14 +27,14 @@ static SSdbRaw *mndMnodeActionEncode(SMnodeObj *pObj);
static SSdbRow *mndMnodeActionDecode(SSdbRaw *pRaw);
static int32_t mndMnodeActionInsert(SSdb *pSdb, SMnodeObj *pObj);
static int32_t mndMnodeActionDelete(SSdb *pSdb, SMnodeObj *pObj);
static int32_t mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOldMnode, SMnodeObj *pNewMnode);
static int32_t mndProcessCreateMnodeReq(SMnodeMsg *pMsg);
static int32_t mndProcessDropMnodeReq(SMnodeMsg *pMsg);
static int32_t mndProcessCreateMnodeRsp(SMnodeMsg *pMsg);
static int32_t mndProcessAlterMnodeRsp(SMnodeMsg *pMsg);
static int32_t mndProcessDropMnodeRsp(SMnodeMsg *pMsg);
static int32_t mndGetMnodeMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta);
static int32_t mndRetrieveMnodes(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOld, SMnodeObj *pNew);
static int32_t mndProcessCreateMnodeReq(SMnodeMsg *pReq);
static int32_t mndProcessDropMnodeReq(SMnodeMsg *pReq);
static int32_t mndProcessCreateMnodeRsp(SMnodeMsg *pRsp);
static int32_t mndProcessAlterMnodeRsp(SMnodeMsg *pRsp);
static int32_t mndProcessDropMnodeRsp(SMnodeMsg *pRsp);
static int32_t mndGetMnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaMsg *pMeta);
static int32_t mndRetrieveMnodes(SMnodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextMnode(SMnode *pMnode, void *pIter);
int32_t mndInitMnode(SMnode *pMnode) {
......@@ -65,7 +65,7 @@ void mndCleanupMnode(SMnode *pMnode) {}
static SMnodeObj *mndAcquireMnode(SMnode *pMnode, int32_t mnodeId) {
SSdb *pSdb = pMnode->pSdb;
SMnodeObj *pObj = sdbAcquire(pSdb, SDB_MNODE, &mnodeId);
if (pObj == NULL) {
if (pObj == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
terrno = TSDB_CODE_MND_MNODE_NOT_EXIST;
}
return pObj;
......@@ -207,9 +207,9 @@ static int32_t mndMnodeActionDelete(SSdb *pSdb, SMnodeObj *pObj) {
return 0;
}
static int32_t mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOldMnode, SMnodeObj *pNewMnode) {
mTrace("mnode:%d, perform update action, old_row:%p new_row:%p", pOldMnode->id, pOldMnode, pNewMnode);
pOldMnode->updateTime = pNewMnode->updateTime;
static int32_t mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOld, SMnodeObj *pNew) {
mTrace("mnode:%d, perform update action, old_row:%p new_row:%p", pOld->id, pOld, pNew);
pOld->updateTime = pNew->updateTime;
return 0;
}
......@@ -277,13 +277,13 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
void *pIter = NULL;
int32_t numOfReplicas = 0;
SDCreateMnodeMsg createMsg = {0};
SDCreateMnodeReq createReq = {0};
while (1) {
SMnodeObj *pMObj = NULL;
pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pMObj);
if (pIter == NULL) break;
SReplica *pReplica = &createMsg.replicas[numOfReplicas];
SReplica *pReplica = &createReq.replicas[numOfReplicas];
pReplica->id = htonl(pMObj->id);
pReplica->port = htons(pMObj->pDnode->port);
memcpy(pReplica->fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
......@@ -292,13 +292,13 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
sdbRelease(pSdb, pMObj);
}
SReplica *pReplica = &createMsg.replicas[numOfReplicas];
SReplica *pReplica = &createReq.replicas[numOfReplicas];
pReplica->id = htonl(pDnode->id);
pReplica->port = htons(pDnode->port);
memcpy(pReplica->fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
numOfReplicas++;
createMsg.replica = numOfReplicas;
createReq.replica = numOfReplicas;
while (1) {
SMnodeObj *pMObj = NULL;
......@@ -307,22 +307,23 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
STransAction action = {0};
SDAlterMnodeMsg *pMsg = malloc(sizeof(SDAlterMnodeMsg));
if (pMsg == NULL) {
SDAlterMnodeReq *pReq = malloc(sizeof(SDAlterMnodeReq));
if (pReq == NULL) {
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pMObj);
return -1;
}
memcpy(pMsg, &createMsg, sizeof(SDAlterMnodeMsg));
memcpy(pReq, &createReq, sizeof(SDAlterMnodeReq));
pMsg->dnodeId = htonl(pMObj->id);
pReq->dnodeId = htonl(pMObj->id);
action.epSet = mndGetDnodeEpset(pMObj->pDnode);
action.pCont = pMsg;
action.contLen = sizeof(SDAlterMnodeMsg);
action.pCont = pReq;
action.contLen = sizeof(SDAlterMnodeReq);
action.msgType = TDMT_DND_ALTER_MNODE;
action.acceptableCode = TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
free(pReq);
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pMObj);
return -1;
......@@ -335,17 +336,18 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
SDCreateMnodeMsg *pMsg = malloc(sizeof(SDCreateMnodeMsg));
if (pMsg == NULL) return -1;
memcpy(pMsg, &createMsg, sizeof(SDAlterMnodeMsg));
pMsg->dnodeId = htonl(pObj->id);
SDCreateMnodeReq *pReq = malloc(sizeof(SDCreateMnodeReq));
if (pReq == NULL) return -1;
memcpy(pReq, &createReq, sizeof(SDAlterMnodeReq));
pReq->dnodeId = htonl(pObj->id);
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pMsg;
action.contLen = sizeof(SDCreateMnodeMsg);
action.pCont = pReq;
action.contLen = sizeof(SDCreateMnodeReq);
action.msgType = TDMT_DND_CREATE_MNODE;
action.acceptableCode = TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
free(pReq);
return -1;
}
}
......@@ -353,39 +355,23 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
return 0;
}
static int32_t mndCreateMnode(SMnode *pMnode, SMnodeMsg *pMsg, SDnodeObj *pDnode, SMCreateMnodeMsg *pCreate) {
static int32_t mndCreateMnode(SMnode *pMnode, SMnodeMsg *pReq, SDnodeObj *pDnode, SMCreateMnodeReq *pCreate) {
int32_t code = -1;
SMnodeObj mnodeObj = {0};
mnodeObj.id = pDnode->id;
mnodeObj.createdTime = taosGetTimestampMs();
mnodeObj.updateTime = mnodeObj.createdTime;
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("mnode:%d, failed to create since %s", pCreate->dnodeId, terrstr());
goto CREATE_MNODE_OVER;
}
mDebug("trans:%d, used to create mnode:%d", pTrans->id, pCreate->dnodeId);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pReq->rpcMsg);
if (pTrans == NULL) goto CREATE_MNODE_OVER;
if (mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj) != 0) {
mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
goto CREATE_MNODE_OVER;
}
if (mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeObj) != 0) {
mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
goto CREATE_MNODE_OVER;
}
if (mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj) != 0) {
mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
goto CREATE_MNODE_OVER;
}
mDebug("trans:%d, used to create mnode:%d", pTrans->id, pCreate->dnodeId);
if (mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj) != 0) goto CREATE_MNODE_OVER;
if (mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeObj) != 0) goto CREATE_MNODE_OVER;
if (mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj) != 0) goto CREATE_MNODE_OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
goto CREATE_MNODE_OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) goto CREATE_MNODE_OVER;
code = 0;
......@@ -394,9 +380,9 @@ CREATE_MNODE_OVER:
return code;
}
static int32_t mndProcessCreateMnodeReq(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
SMCreateMnodeMsg *pCreate = pMsg->rpcMsg.pCont;
static int32_t mndProcessCreateMnodeReq(SMnodeMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
SMCreateMnodeReq *pCreate = pReq->rpcMsg.pCont;
pCreate->dnodeId = htonl(pCreate->dnodeId);
......@@ -408,6 +394,9 @@ static int32_t mndProcessCreateMnodeReq(SMnodeMsg *pMsg) {
mError("mnode:%d, mnode already exist", pObj->id);
terrno = TSDB_CODE_MND_MNODE_ALREADY_EXIST;
return -1;
} else if (terrno != TSDB_CODE_MND_MNODE_NOT_EXIST) {
mError("qnode:%d, failed to create mnode since %s", pCreate->dnodeId, terrstr());
return -1;
}
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pCreate->dnodeId);
......@@ -417,7 +406,7 @@ static int32_t mndProcessCreateMnodeReq(SMnodeMsg *pMsg) {
return -1;
}
int32_t code = mndCreateMnode(pMnode, pMsg, pDnode, pCreate);
int32_t code = mndCreateMnode(pMnode, pReq, pDnode, pCreate);
mndReleaseDnode(pMnode, pDnode);
if (code != 0) {
......@@ -449,14 +438,14 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
void *pIter = NULL;
int32_t numOfReplicas = 0;
SDAlterMnodeMsg alterMsg = {0};
SDAlterMnodeReq alterReq = {0};
while (1) {
SMnodeObj *pMObj = NULL;
pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pMObj);
if (pIter == NULL) break;
if (pMObj->id != pObj->id) {
SReplica *pReplica = &alterMsg.replicas[numOfReplicas];
SReplica *pReplica = &alterReq.replicas[numOfReplicas];
pReplica->id = htonl(pMObj->id);
pReplica->port = htons(pMObj->pDnode->port);
memcpy(pReplica->fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
......@@ -466,7 +455,7 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
sdbRelease(pSdb, pMObj);
}
alterMsg.replica = numOfReplicas;
alterReq.replica = numOfReplicas;
while (1) {
SMnodeObj *pMObj = NULL;
......@@ -475,22 +464,23 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
if (pMObj->id != pObj->id) {
STransAction action = {0};
SDAlterMnodeMsg *pMsg = malloc(sizeof(SDAlterMnodeMsg));
if (pMsg == NULL) {
SDAlterMnodeReq *pReq = malloc(sizeof(SDAlterMnodeReq));
if (pReq == NULL) {
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pMObj);
return -1;
}
memcpy(pMsg, &alterMsg, sizeof(SDAlterMnodeMsg));
memcpy(pReq, &alterReq, sizeof(SDAlterMnodeReq));
pMsg->dnodeId = htonl(pMObj->id);
pReq->dnodeId = htonl(pMObj->id);
action.epSet = mndGetDnodeEpset(pMObj->pDnode);
action.pCont = pMsg;
action.contLen = sizeof(SDAlterMnodeMsg);
action.pCont = pReq;
action.contLen = sizeof(SDAlterMnodeReq);
action.msgType = TDMT_DND_ALTER_MNODE;
action.acceptableCode = TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
free(pReq);
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pMObj);
return -1;
......@@ -504,19 +494,20 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
SDDropMnodeMsg *pMsg = malloc(sizeof(SDDropMnodeMsg));
if (pMsg == NULL) {
SDDropMnodeReq *pReq = malloc(sizeof(SDDropMnodeReq));
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pMsg->dnodeId = htonl(pObj->id);
pReq->dnodeId = htonl(pObj->id);
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pMsg;
action.contLen = sizeof(SDDropMnodeMsg);
action.pCont = pReq;
action.contLen = sizeof(SDDropMnodeReq);
action.msgType = TDMT_DND_DROP_MNODE;
action.acceptableCode = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
free(pReq);
return -1;
}
}
......@@ -524,35 +515,18 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
return 0;
}
static int32_t mndDropMnode(SMnode *pMnode, SMnodeMsg *pMsg, SMnodeObj *pObj) {
static int32_t mndDropMnode(SMnode *pMnode, SMnodeMsg *pReq, SMnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("mnode:%d, failed to drop since %s", pObj->id, terrstr());
goto DROP_MNODE_OVER;
}
mDebug("trans:%d, used to drop mnode:%d", pTrans->id, pObj->id);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pReq->rpcMsg);
if (pTrans == NULL) goto DROP_MNODE_OVER;
if (mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj) != 0) {
mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
goto DROP_MNODE_OVER;
}
if (mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj) != 0) {
mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
goto DROP_MNODE_OVER;
}
if (mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj) != 0) {
mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
goto DROP_MNODE_OVER;
}
mDebug("trans:%d, used to drop mnode:%d", pTrans->id, pObj->id);
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
goto DROP_MNODE_OVER;
}
if (mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj) != 0) goto DROP_MNODE_OVER;
if (mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj) != 0) goto DROP_MNODE_OVER;
if (mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj) != 0) goto DROP_MNODE_OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto DROP_MNODE_OVER;
code = 0;
......@@ -561,9 +535,9 @@ DROP_MNODE_OVER:
return code;
}
static int32_t mndProcessDropMnodeReq(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
SMDropMnodeMsg *pDrop = pMsg->rpcMsg.pCont;
static int32_t mndProcessDropMnodeReq(SMnodeMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
SMDropMnodeReq *pDrop = pReq->rpcMsg.pCont;
pDrop->dnodeId = htonl(pDrop->dnodeId);
mDebug("mnode:%d, start to drop", pDrop->dnodeId);
......@@ -577,12 +551,10 @@ static int32_t mndProcessDropMnodeReq(SMnodeMsg *pMsg) {
SMnodeObj *pObj = mndAcquireMnode(pMnode, pDrop->dnodeId);
if (pObj == NULL) {
mError("mnode:%d, not exist", pDrop->dnodeId);
terrno = TSDB_CODE_MND_DNODE_NOT_EXIST;
return -1;
}
int32_t code = mndDropMnode(pMnode, pMsg, pObj);
int32_t code = mndDropMnode(pMnode, pReq, pObj);
if (code != 0) {
mError("mnode:%d, failed to drop since %s", pMnode->dnodeId, terrstr());
return -1;
......@@ -592,23 +564,23 @@ static int32_t mndProcessDropMnodeReq(SMnodeMsg *pMsg) {
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}
static int32_t mndProcessCreateMnodeRsp(SMnodeMsg *pMsg) {
mndTransProcessRsp(pMsg);
static int32_t mndProcessCreateMnodeRsp(SMnodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessAlterMnodeRsp(SMnodeMsg *pMsg) {
mndTransProcessRsp(pMsg);
static int32_t mndProcessAlterMnodeRsp(SMnodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndProcessDropMnodeRsp(SMnodeMsg *pMsg) {
mndTransProcessRsp(pMsg);
static int32_t mndProcessDropMnodeRsp(SMnodeMsg *pRsp) {
mndTransProcessRsp(pRsp);
return 0;
}
static int32_t mndGetMnodeMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta) {
SMnode *pMnode = pMsg->pMnode;
static int32_t mndGetMnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaMsg *pMeta) {
SMnode *pMnode = pReq->pMnode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
......@@ -660,8 +632,8 @@ static int32_t mndGetMnodeMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *
return 0;
}
static int32_t mndRetrieveMnodes(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pMsg->pMnode;
static int32_t mndRetrieveMnodes(SMnodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pMnode;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
int32_t cols = 0;
......
source/dnode/mnode/impl/src/mndSync.c
浏览文件 @ 85a16137
......@@ -86,10 +86,6 @@ static int32_t mndRestoreWal(SMnode *pMnode) {
mndTransPullup(pMnode);
if (walBeginSnapshot(pWal, sdbVer) < 0) {
goto WAL_RESTORE_OVER;
}
if (sdbVer != lastSdbVer) {
mInfo("sdb restored from %" PRId64 " to %" PRId64 ", write file", lastSdbVer, sdbVer);
if (sdbWriteFile(pSdb) != 0) {
......@@ -97,6 +93,10 @@ static int32_t mndRestoreWal(SMnode *pMnode) {
}
}
if (walBeginSnapshot(pWal, sdbVer) < 0) {
goto WAL_RESTORE_OVER;
}
if (walEndSnapshot(pWal) < 0) {
goto WAL_RESTORE_OVER;
}
......
source/dnode/mnode/impl/test/mnode/mnode.cpp
浏览文件 @ 85a16137
......@@ -18,37 +18,25 @@ class MndTestMnode : public ::testing::Test {
public:
static void SetUpTestSuite() {
test.Init("/tmp/mnode_test_mnode1", 9031);
test.Init("/tmp/mnode_test_mnode1", 9028);
const char* fqdn = "localhost";
const char* firstEp = "localhost:9031";
const char* firstEp = "localhost:9028";
server2.Start("/tmp/mnode_test_mnode2", fqdn, 9032, firstEp);
server3.Start("/tmp/mnode_test_mnode3", fqdn, 9033, firstEp);
server4.Start("/tmp/mnode_test_mnode4", fqdn, 9034, firstEp);
server5.Start("/tmp/mnode_test_mnode5", fqdn, 9035, firstEp);
server2.Start("/tmp/mnode_test_mnode2", fqdn, 9029, firstEp);
taosMsleep(300);
}
static void TearDownTestSuite() {
server2.Stop();
server3.Stop();
server4.Stop();
server5.Stop();
test.Cleanup();
}
static Testbase test;
static TestServer server2;
static TestServer server3;
static TestServer server4;
static TestServer server5;
};
Testbase MndTestMnode::test;
TestServer MndTestMnode::server2;
TestServer MndTestMnode::server3;
TestServer MndTestMnode::server4;
TestServer MndTestMnode::server5;
TEST_F(MndTestMnode, 01_ShowDnode) {
test.SendShowMetaReq(TSDB_MGMT_TABLE_MNODE, "");
......@@ -64,7 +52,7 @@ TEST_F(MndTestMnode, 01_ShowDnode) {
EXPECT_EQ(test.GetShowRows(), 1);
CheckInt16(1);
CheckBinary("localhost:9031", TSDB_EP_LEN);
CheckBinary("localhost:9028", TSDB_EP_LEN);
CheckBinary("master", 12);
CheckInt64(0);
CheckTimestamp();
......@@ -72,9 +60,9 @@ TEST_F(MndTestMnode, 01_ShowDnode) {
TEST_F(MndTestMnode, 02_Create_Mnode_Invalid_Id) {
{
int32_t contLen = sizeof(SMCreateMnodeMsg);
int32_t contLen = sizeof(SMCreateMnodeReq);
SMCreateMnodeMsg* pReq = (SMCreateMnodeMsg*)rpcMallocCont(contLen);
SMCreateMnodeReq* pReq = (SMCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(1);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_MNODE, pReq, contLen);
......@@ -85,9 +73,9 @@ TEST_F(MndTestMnode, 02_Create_Mnode_Invalid_Id) {
TEST_F(MndTestMnode, 03_Create_Mnode_Invalid_Id) {
{
int32_t contLen = sizeof(SMCreateMnodeMsg);
int32_t contLen = sizeof(SMCreateMnodeReq);
SMCreateMnodeMsg* pReq = (SMCreateMnodeMsg*)rpcMallocCont(contLen);
SMCreateMnodeReq* pReq = (SMCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_MNODE, pReq, contLen);
......@@ -103,7 +91,7 @@ TEST_F(MndTestMnode, 04_Create_Mnode) {
SCreateDnodeReq* pReq = (SCreateDnodeReq*)rpcMallocCont(contLen);
strcpy(pReq->fqdn, "localhost");
pReq->port = htonl(9032);
pReq->port = htonl(9029);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_DNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
......@@ -117,9 +105,9 @@ TEST_F(MndTestMnode, 04_Create_Mnode) {
{
// create mnode
int32_t contLen = sizeof(SMCreateMnodeMsg);
int32_t contLen = sizeof(SMCreateMnodeReq);
SMCreateMnodeMsg* pReq = (SMCreateMnodeMsg*)rpcMallocCont(contLen);
SMCreateMnodeReq* pReq = (SMCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_MNODE, pReq, contLen);
......@@ -132,8 +120,8 @@ TEST_F(MndTestMnode, 04_Create_Mnode) {
CheckInt16(1);
CheckInt16(2);
CheckBinary("localhost:9031", TSDB_EP_LEN);
CheckBinary("localhost:9032", TSDB_EP_LEN);
CheckBinary("localhost:9028", TSDB_EP_LEN);
CheckBinary("localhost:9029", TSDB_EP_LEN);
CheckBinary("master", 12);
CheckBinary("slave", 12);
CheckInt64(0);
......@@ -144,9 +132,9 @@ TEST_F(MndTestMnode, 04_Create_Mnode) {
{
// drop mnode
int32_t contLen = sizeof(SMDropMnodeMsg);
int32_t contLen = sizeof(SMDropMnodeReq);
SMDropMnodeMsg* pReq = (SMDropMnodeMsg*)rpcMallocCont(contLen);
SMDropMnodeReq* pReq = (SMDropMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_MNODE, pReq, contLen);
......@@ -158,144 +146,145 @@ TEST_F(MndTestMnode, 04_Create_Mnode) {
EXPECT_EQ(test.GetShowRows(), 1);
CheckInt16(1);
CheckBinary("localhost:9031", TSDB_EP_LEN);
CheckBinary("localhost:9028", TSDB_EP_LEN);
CheckBinary("master", 12);
CheckInt64(0);
CheckTimestamp();
}
{
// drop mnode
int32_t contLen = sizeof(SMDropMnodeReq);
SMDropMnodeReq* pReq = (SMDropMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_MND_MNODE_NOT_EXIST);
}
}
TEST_F(MndTestMnode, 03_Create_Mnode_Rollback) {
{
// send message first, then dnode2 crash, result is returned, and rollback is started
int32_t contLen = sizeof(SMCreateMnodeReq);
SMCreateMnodeReq* pReq = (SMCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
{
// continue send message, mnode is creating
int32_t contLen = sizeof(SMCreateMnodeReq);
SMCreateMnodeReq* pReq = (SMCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_CREATING);
}
{
// continue send message, mnode is creating
int32_t contLen = sizeof(SMDropMnodeReq);
SMDropMnodeReq* pReq = (SMDropMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_CREATING);
}
{
// server start, wait until the rollback finished
server2.DoStart();
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateMnodeReq);
SMCreateMnodeReq* pReq = (SMCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
if (pRsp->code == TSDB_CODE_MND_MNODE_ALREADY_EXIST) break;
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
}
}
TEST_F(MndTestMnode, 04_Drop_Mnode_Rollback) {
{
// send message first, then dnode2 crash, result is returned, and rollback is started
int32_t contLen = sizeof(SMDropMnodeReq);
SMDropMnodeReq* pReq = (SMDropMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
{
// continue send message, mnode is dropping
int32_t contLen = sizeof(SMCreateMnodeReq);
SMCreateMnodeReq* pReq = (SMCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
// continue send message, mnode is dropping
int32_t contLen = sizeof(SMDropMnodeReq);
SMDropMnodeReq* pReq = (SMDropMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
// server start, wait until the rollback finished
server2.DoStart();
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 20;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateMnodeReq);
SMCreateMnodeReq* pReq = (SMCreateMnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_MNODE, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
if (pRsp->code == 0) break;
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
}
}
\ No newline at end of file
// {
// int32_t contLen = sizeof(SDropDnodeReq);
// SDropDnodeReq* pReq = (SDropDnodeReq*)rpcMallocCont(contLen);
// pReq->dnodeId = htonl(2);
// SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_DNODE, pReq, contLen);
// ASSERT_NE(pRsp, nullptr);
// ASSERT_EQ(pRsp->code, 0);
// }
// test.SendShowMetaReq(TSDB_MGMT_TABLE_DNODE, "");
// CHECK_META("show dnodes", 7);
// test.SendShowRetrieveReq();
// EXPECT_EQ(test.GetShowRows(), 1);
// CheckInt16(1);
// CheckBinary("localhost:9031", TSDB_EP_LEN);
// CheckInt16(0);
// CheckInt16(1);
// CheckBinary("ready", 10);
// CheckTimestamp();
// CheckBinary("", 24);
// {
// int32_t contLen = sizeof(SCreateDnodeReq);
// SCreateDnodeReq* pReq = (SCreateDnodeReq*)rpcMallocCont(contLen);
// strcpy(pReq->ep, "localhost:9033");
// SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_DNODE, pReq, contLen);
// ASSERT_NE(pRsp, nullptr);
// ASSERT_EQ(pRsp->code, 0);
// }
// {
// int32_t contLen = sizeof(SCreateDnodeReq);
// SCreateDnodeReq* pReq = (SCreateDnodeReq*)rpcMallocCont(contLen);
// strcpy(pReq->ep, "localhost:9034");
// SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_DNODE, pReq, contLen);
// ASSERT_NE(pRsp, nullptr);
// ASSERT_EQ(pRsp->code, 0);
// }
// {
// int32_t contLen = sizeof(SCreateDnodeReq);
// SCreateDnodeReq* pReq = (SCreateDnodeReq*)rpcMallocCont(contLen);
// strcpy(pReq->ep, "localhost:9035");
// SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_DNODE, pReq, contLen);
// ASSERT_NE(pRsp, nullptr);
// ASSERT_EQ(pRsp->code, 0);
// }
// taosMsleep(1300);
// test.SendShowMetaReq(TSDB_MGMT_TABLE_DNODE, "");
// CHECK_META("show dnodes", 7);
// test.SendShowRetrieveReq();
// EXPECT_EQ(test.GetShowRows(), 4);
// CheckInt16(1);
// CheckInt16(3);
// CheckInt16(4);
// CheckInt16(5);
// CheckBinary("localhost:9031", TSDB_EP_LEN);
// CheckBinary("localhost:9033", TSDB_EP_LEN);
// CheckBinary("localhost:9034", TSDB_EP_LEN);
// CheckBinary("localhost:9035", TSDB_EP_LEN);
// CheckInt16(0);
// CheckInt16(0);
// CheckInt16(0);
// CheckInt16(0);
// CheckInt16(1);
// CheckInt16(1);
// CheckInt16(1);
// CheckInt16(1);
// CheckBinary("ready", 10);
// CheckBinary("ready", 10);
// CheckBinary("ready", 10);
// CheckBinary("ready", 10);
// CheckTimestamp();
// CheckTimestamp();
// CheckTimestamp();
// CheckTimestamp();
// CheckBinary("", 24);
// CheckBinary("", 24);
// CheckBinary("", 24);
// CheckBinary("", 24);
// // restart
// uInfo("stop all server");
// test.Restart();
// server2.Restart();
// server3.Restart();
// server4.Restart();
// server5.Restart();
// taosMsleep(1300);
// test.SendShowMetaReq(TSDB_MGMT_TABLE_DNODE, "");
// CHECK_META("show dnodes", 7);
// test.SendShowRetrieveReq();
// EXPECT_EQ(test.GetShowRows(), 4);
// CheckInt16(1);
// CheckInt16(3);
// CheckInt16(4);
// CheckInt16(5);
// CheckBinary("localhost:9031", TSDB_EP_LEN);
// CheckBinary("localhost:9033", TSDB_EP_LEN);
// CheckBinary("localhost:9034", TSDB_EP_LEN);
// CheckBinary("localhost:9035", TSDB_EP_LEN);
// CheckInt16(0);
// CheckInt16(0);
// CheckInt16(0);
// CheckInt16(0);
// CheckInt16(1);
// CheckInt16(1);
// CheckInt16(1);
// CheckInt16(1);
// CheckBinary("ready", 10);
// CheckBinary("ready", 10);
// CheckBinary("ready", 10);
// CheckBinary("ready", 10);
// CheckTimestamp();
// CheckTimestamp();
// CheckTimestamp();
// CheckTimestamp();
// CheckBinary("", 24);
// CheckBinary("", 24);
// CheckBinary("", 24);
// CheckBinary("", 24);
// }
source/dnode/vnode/impl/inc/vnodeCommit.h
浏览文件 @ 85a16137
......@@ -23,8 +23,8 @@ extern "C" {
#endif
#define vnodeShouldCommit vnodeBufPoolIsFull
int vnodeSyncCommit(SVnode *pVnode);
int vnodeAsyncCommit(SVnode *pVnode);
int vnodeCommit(void *arg);
#ifdef __cplusplus
}
......
source/dnode/vnode/impl/src/vnodeCommit.c
浏览文件 @ 85a16137
......@@ -17,8 +17,12 @@
static int vnodeStartCommit(SVnode *pVnode);
static int vnodeEndCommit(SVnode *pVnode);
static int vnodeCommit(void *arg);
static void vnodeWaitCommit(SVnode *pVnode);
int vnodeAsyncCommit(SVnode *pVnode) {
vnodeWaitCommit(pVnode);
vnodeBufPoolSwitch(pVnode);
SVnodeTask *pTask = (SVnodeTask *)malloc(sizeof(*pTask));
......@@ -33,7 +37,13 @@ int vnodeAsyncCommit(SVnode *pVnode) {
return 0;
}
int vnodeCommit(void *arg) {
int vnodeSyncCommit(SVnode *pVnode) {
vnodeAsyncCommit(pVnode);
vnodeWaitCommit(pVnode);
return 0;
}
static int vnodeCommit(void *arg) {
SVnode *pVnode = (SVnode *)arg;
metaCommit(pVnode->pMeta);
......@@ -42,7 +52,6 @@ int vnodeCommit(void *arg) {
vnodeBufPoolRecycle(pVnode);
tsem_post(&(pVnode->canCommit));
// TODO
return 0;
}
......@@ -55,3 +64,5 @@ static int vnodeEndCommit(SVnode *pVnode) {
// TODO
return 0;
}
static FORCE_INLINE void vnodeWaitCommit(SVnode *pVnode) { tsem_wait(&pVnode->canCommit); }
\ No newline at end of file
source/dnode/vnode/impl/src/vnodeMain.c
浏览文件 @ 85a16137
......@@ -137,6 +137,7 @@ static int vnodeOpenImpl(SVnode *pVnode) {
}
static void vnodeCloseImpl(SVnode *pVnode) {
// vnodeSyncCommit(pVnode);
if (pVnode) {
vnodeCloseBufPool(pVnode);
metaClose(pVnode->pMeta);
......
source/dnode/vnode/impl/src/vnodeWrite.c
浏览文件 @ 85a16137
......@@ -92,7 +92,7 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
// }
break;
case TDMT_VND_SUBMIT:
if (tsdbInsertData(pVnode->pTsdb, (SSubmitMsg *)ptr) < 0) {
if (tsdbInsertData(pVnode->pTsdb, (SSubmitMsg *)ptr, NULL) < 0) {
// TODO: handle error
}
break;
......
source/dnode/vnode/tsdb/CMakeLists.txt
浏览文件 @ 85a16137
......@@ -10,6 +10,9 @@ else(0)
"src/tsdbMemTable.c"
"src/tsdbOptions.c"
"src/tsdbWrite.c"
"src/tsdbReadImpl.c"
"src/tsdbFile.c"
"src/tsdbFS.c"
)
endif(0)
......@@ -25,4 +28,5 @@ target_link_libraries(
PUBLIC util
PUBLIC common
PUBLIC tkv
PUBLIC tfs
)
\ No newline at end of file
source/dnode/vnode/tsdb/inc/tsdbCommit.h
浏览文件 @ 85a16137
......@@ -16,7 +16,6 @@
#ifndef _TD_TSDB_COMMIT_H_
#define _TD_TSDB_COMMIT_H_
#if 0
typedef struct {
int minFid;
int midFid;
......@@ -30,9 +29,19 @@ typedef struct {
int64_t size;
} SKVRecord;
void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn);
static FORCE_INLINE int TSDB_KEY_FID(TSKEY key, int32_t days, int8_t precision) {
if (key < 0) {
return (int)((key + 1) / tsTickPerDay[precision] / days - 1);
} else {
return (int)((key / tsTickPerDay[precision] / days));
}
}
#if 0
#define TSDB_DEFAULT_BLOCK_ROWS(maxRows) ((maxRows)*4 / 5)
void tsdbGetRtnSnap(STsdbRepo *pRepo, SRtn *pRtn);
int tsdbEncodeKVRecord(void **buf, SKVRecord *pRecord);
void *tsdbDecodeKVRecord(void *buf, SKVRecord *pRecord);
void *tsdbCommitData(STsdbRepo *pRepo);
......
source/dnode/vnode/tsdb/inc/tsdbDef.h
浏览文件 @ 85a16137
......@@ -17,27 +17,43 @@
#define _TD_TSDB_DEF_H_
#include "mallocator.h"
#include "tmsg.h"
#include "tlist.h"
#include "tcompression.h"
#include "tglobal.h"
#include "thash.h"
#include "tlist.h"
#include "tmsg.h"
#include "tskiplist.h"
#include "ttime.h"
#include "tsdb.h"
#include "tsdbCommit.h"
#include "tsdbFS.h"
#include "tsdbFile.h"
#include "tsdbLog.h"
#include "tsdbMemTable.h"
#include "tsdbMemory.h"
#include "tsdbOptions.h"
#include "tsdbReadImpl.h"
#ifdef __cplusplus
extern "C" {
#endif
struct STsdb {
int32_t vgId;
char * path;
STsdbCfg options;
STsdbCfg config;
STsdbMemTable * mem;
STsdbMemTable * imem;
SRtn rtn;
SMemAllocatorFactory *pmaf;
STsdbFS fs;
};
#define REPO_ID(r) 0
#define REPO_CFG(r) (&(r)->config)
#define REPO_FS(r) (&(r)->fs)
#ifdef __cplusplus
}
#endif
......
source/dnode/vnode/tsdb/inc/tsdbFS.h
浏览文件 @ 85a16137
......@@ -16,7 +16,7 @@
#ifndef _TD_TSDB_FS_H_
#define _TD_TSDB_FS_H_
#if 0
#include "tsdbFile.h"
#define TSDB_FS_VERSION 0
......@@ -39,19 +39,17 @@ typedef struct {
// ==================
typedef struct {
STsdbFSMeta meta; // FS meta
SMFile* pmf; // meta file pointer
SMFile mf; // meta file
SArray* df; // data file array
SArray * df; // data file array
} SFSStatus;
typedef struct {
pthread_rwlock_t lock;
SFSStatus* cstatus; // current status
SHashObj* metaCache; // meta cache
SHashObj* metaCacheComp; // meta cache for compact
SFSStatus *cstatus; // current status
SHashObj * metaCache; // meta cache
SHashObj * metaCacheComp; // meta cache for compact
bool intxn;
SFSStatus* nstatus; // new status
SFSStatus *nstatus; // new status
} STsdbFS;
#define FS_CURRENT_STATUS(pfs) ((pfs)->cstatus)
......@@ -63,10 +61,10 @@ typedef struct {
typedef struct {
int direction;
uint64_t version; // current FS version
STsdbFS* pfs;
STsdbFS * pfs;
int index; // used to position next fset when version the same
int fid; // used to seek when version is changed
SDFileSet* pSet;
SDFileSet *pSet;
} SFSIter;
#define TSDB_FS_ITER_FORWARD TSDB_ORDER_ASC
......@@ -74,21 +72,21 @@ typedef struct {
STsdbFS *tsdbNewFS(STsdbCfg *pCfg);
void * tsdbFreeFS(STsdbFS *pfs);
int tsdbOpenFS(STsdbRepo *pRepo);
void tsdbCloseFS(STsdbRepo *pRepo);
void tsdbStartFSTxn(STsdbRepo *pRepo, int64_t pointsAdd, int64_t storageAdd);
int tsdbEndFSTxn(STsdbRepo *pRepo);
int tsdbOpenFS(STsdb *pRepo);
void tsdbCloseFS(STsdb *pRepo);
void tsdbStartFSTxn(STsdb *pRepo, int64_t pointsAdd, int64_t storageAdd);
int tsdbEndFSTxn(STsdb *pRepo);
int tsdbEndFSTxnWithError(STsdbFS *pfs);
void tsdbUpdateFSTxnMeta(STsdbFS *pfs, STsdbFSMeta *pMeta);
void tsdbUpdateMFile(STsdbFS *pfs, const SMFile *pMFile);
// void tsdbUpdateMFile(STsdbFS *pfs, const SMFile *pMFile);
int tsdbUpdateDFileSet(STsdbFS *pfs, const SDFileSet *pSet);
void tsdbFSIterInit(SFSIter *pIter, STsdbFS *pfs, int direction);
void tsdbFSIterSeek(SFSIter *pIter, int fid);
SDFileSet *tsdbFSIterNext(SFSIter *pIter);
int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta);
int tsdbLoadMetaCache(STsdb *pRepo, bool recoverMeta);
static FORCE_INLINE int tsdbRLockFS(STsdbFS* pFs) {
static FORCE_INLINE int tsdbRLockFS(STsdbFS *pFs) {
int code = pthread_rwlock_rdlock(&(pFs->lock));
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(code);
......@@ -97,7 +95,7 @@ static FORCE_INLINE int tsdbRLockFS(STsdbFS* pFs) {
return 0;
}
static FORCE_INLINE int tsdbWLockFS(STsdbFS* pFs) {
static FORCE_INLINE int tsdbWLockFS(STsdbFS *pFs) {
int code = pthread_rwlock_wrlock(&(pFs->lock));
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(code);
......@@ -106,7 +104,7 @@ static FORCE_INLINE int tsdbWLockFS(STsdbFS* pFs) {
return 0;
}
static FORCE_INLINE int tsdbUnLockFS(STsdbFS* pFs) {
static FORCE_INLINE int tsdbUnLockFS(STsdbFS *pFs) {
int code = pthread_rwlock_unlock(&(pFs->lock));
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(code);
......@@ -115,6 +113,4 @@ static FORCE_INLINE int tsdbUnLockFS(STsdbFS* pFs) {
return 0;
}
#endif
#endif /* _TD_TSDB_FS_H_ */
source/dnode/vnode/tsdb/inc/tsdbFile.h
浏览文件 @ 85a16137
......@@ -16,7 +16,8 @@
#ifndef _TS_TSDB_FILE_H_
#define _TS_TSDB_FILE_H_
#if 0
#include "tchecksum.h"
#include "tfs.h"
#define TSDB_FILE_HEAD_SIZE 512
#define TSDB_FILE_DELIMITER 0xF00AFA0F
......@@ -34,7 +35,7 @@
#define TSDB_FILE_SET_CLOSED(f) (TSDB_FILE_FD(f) = -1)
#define TSDB_FILE_LEVEL(tf) TFILE_LEVEL(TSDB_FILE_F(tf))
#define TSDB_FILE_ID(tf) TFILE_ID(TSDB_FILE_F(tf))
#define TSDB_FILE_FSYNC(tf) taosFsync(TSDB_FILE_FD(tf))
#define TSDB_FILE_FSYNC(tf) taosFsyncFile(TSDB_FILE_FD(tf))
#define TSDB_FILE_STATE(tf) ((tf)->state)
#define TSDB_FILE_SET_STATE(tf, s) ((tf)->state = (s))
#define TSDB_FILE_IS_OK(tf) (TSDB_FILE_STATE(tf) == TSDB_FILE_STATE_OK)
......@@ -42,6 +43,7 @@
typedef enum { TSDB_FILE_HEAD = 0, TSDB_FILE_DATA, TSDB_FILE_LAST, TSDB_FILE_MAX, TSDB_FILE_META } TSDB_FILE_T;
#if 0
// =============== SMFile
typedef struct {
int64_t size;
......@@ -68,7 +70,7 @@ int tsdbApplyMFileChange(SMFile* from, SMFile* to);
int tsdbCreateMFile(SMFile* pMFile, bool updateHeader);
int tsdbUpdateMFileHeader(SMFile* pMFile);
int tsdbLoadMFileHeader(SMFile* pMFile, SMFInfo* pInfo);
int tsdbScanAndTryFixMFile(STsdbRepo* pRepo);
int tsdbScanAndTryFixMFile(STsdb* pRepo);
int tsdbEncodeMFInfo(void** buf, SMFInfo* pInfo);
void* tsdbDecodeMFInfo(void* buf, SMFInfo* pInfo);
......@@ -96,7 +98,7 @@ static FORCE_INLINE void tsdbCloseMFile(SMFile* pMFile) {
static FORCE_INLINE int64_t tsdbSeekMFile(SMFile* pMFile, int64_t offset, int whence) {
ASSERT(TSDB_FILE_OPENED(pMFile));
int64_t loffset = taosLSeek(TSDB_FILE_FD(pMFile), offset, whence);
int64_t loffset = taosLSeekFile(TSDB_FILE_FD(pMFile), offset, whence);
if (loffset < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
......@@ -108,7 +110,7 @@ static FORCE_INLINE int64_t tsdbSeekMFile(SMFile* pMFile, int64_t offset, int wh
static FORCE_INLINE int64_t tsdbWriteMFile(SMFile* pMFile, void* buf, int64_t nbyte) {
ASSERT(TSDB_FILE_OPENED(pMFile));
int64_t nwrite = taosWrite(pMFile->fd, buf, nbyte);
int64_t nwrite = taosWriteFile(pMFile->fd, buf, nbyte);
if (nwrite < nbyte) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
......@@ -150,7 +152,7 @@ static FORCE_INLINE int tsdbRemoveMFile(SMFile* pMFile) { return tfsremove(TSDB_
static FORCE_INLINE int64_t tsdbReadMFile(SMFile* pMFile, void* buf, int64_t nbyte) {
ASSERT(TSDB_FILE_OPENED(pMFile));
int64_t nread = taosRead(pMFile->fd, buf, nbyte);
int64_t nread = taosReadFile(pMFile->fd, buf, nbyte);
if (nread < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
......@@ -159,6 +161,8 @@ static FORCE_INLINE int64_t tsdbReadMFile(SMFile* pMFile, void* buf, int64_t nby
return nread;
}
#endif
// =============== SDFile
typedef struct {
uint32_t magic;
......@@ -210,7 +214,7 @@ static FORCE_INLINE void tsdbCloseDFile(SDFile* pDFile) {
static FORCE_INLINE int64_t tsdbSeekDFile(SDFile* pDFile, int64_t offset, int whence) {
ASSERT(TSDB_FILE_OPENED(pDFile));
int64_t loffset = taosLSeek(TSDB_FILE_FD(pDFile), offset, whence);
int64_t loffset = taosLSeekFile(TSDB_FILE_FD(pDFile), offset, whence);
if (loffset < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
......@@ -222,7 +226,7 @@ static FORCE_INLINE int64_t tsdbSeekDFile(SDFile* pDFile, int64_t offset, int wh
static FORCE_INLINE int64_t tsdbWriteDFile(SDFile* pDFile, void* buf, int64_t nbyte) {
ASSERT(TSDB_FILE_OPENED(pDFile));
int64_t nwrite = taosWrite(pDFile->fd, buf, nbyte);
int64_t nwrite = taosWriteFile(pDFile->fd, buf, nbyte);
if (nwrite < nbyte) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
......@@ -264,7 +268,7 @@ static FORCE_INLINE int tsdbRemoveDFile(SDFile* pDFile) { return tfsremove(TSDB_
static FORCE_INLINE int64_t tsdbReadDFile(SDFile* pDFile, void* buf, int64_t nbyte) {
ASSERT(TSDB_FILE_OPENED(pDFile));
int64_t nread = taosRead(pDFile->fd, buf, nbyte);
int64_t nread = taosReadFile(pDFile->fd, buf, nbyte);
if (nread < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
......@@ -316,7 +320,7 @@ void* tsdbDecodeDFileSetEx(void* buf, SDFileSet* pSet);
int tsdbApplyDFileSetChange(SDFileSet* from, SDFileSet* to);
int tsdbCreateDFileSet(SDFileSet* pSet, bool updateHeader);
int tsdbUpdateDFileSetHeader(SDFileSet* pSet);
int tsdbScanAndTryFixDFileSet(STsdbRepo *pRepo, SDFileSet* pSet);
int tsdbScanAndTryFixDFileSet(STsdb* pRepo, SDFileSet* pSet);
static FORCE_INLINE void tsdbCloseDFileSet(SDFileSet* pSet) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
......@@ -366,5 +370,4 @@ static FORCE_INLINE bool tsdbFSetIsOk(SDFileSet* pSet) {
return true;
}
#endif
#endif /* _TS_TSDB_FILE_H_ */
\ No newline at end of file
source/dnode/vnode/tsdb/inc/tsdbLog.h 0 → 100644
浏览文件 @ 85a16137
/*
* 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_TSDB_LOG_H_
#define _TD_TSDB_LOG_H_
#include "tlog.h"
extern int32_t tsdbDebugFlag;
#define tsdbFatal(...) do { if (tsdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", 255, __VA_ARGS__); }} while(0)
#define tsdbError(...) do { if (tsdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", 255, __VA_ARGS__); }} while(0)
#define tsdbWarn(...) do { if (tsdbDebugFlag & DEBUG_WARN) { taosPrintLog("TDB WARN ", 255, __VA_ARGS__); }} while(0)
#define tsdbInfo(...) do { if (tsdbDebugFlag & DEBUG_INFO) { taosPrintLog("TDB ", 255, __VA_ARGS__); }} while(0)
#define tsdbDebug(...) do { if (tsdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
#define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
#endif /* _TD_TSDB_LOG_H_ */
\ No newline at end of file
source/dnode/vnode/tsdb/inc/tsdbMemTable.h
浏览文件 @ 85a16137
......@@ -24,9 +24,9 @@ extern "C" {
typedef struct STsdbMemTable STsdbMemTable;
STsdbMemTable *tsdbNewMemTable(SMemAllocatorFactory *pMAF);
void tsdbFreeMemTable(SMemAllocatorFactory *pMAF, STsdbMemTable *pMemTable);
int tsdbInsertDataToMemTable(STsdbMemTable *pMemTable, SSubmitMsg *pMsg);
STsdbMemTable *tsdbNewMemTable(STsdb *pTsdb);
void tsdbFreeMemTable(STsdb *pTsdb, STsdbMemTable *pMemTable);
int tsdbMemTableInsert(STsdb *pTsdb, STsdbMemTable *pMemTable, SSubmitMsg *pMsg, SShellSubmitRspMsg *pRsp);
#ifdef __cplusplus
}
......
source/dnode/vnode/tsdb/inc/tsdbMemory.h 0 → 100644
浏览文件 @ 85a16137
/*
* 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_TSDB_MEMORY_H_
#define _TD_TSDB_MEMORY_H_
static void * taosTMalloc(size_t size);
static void * taosTCalloc(size_t nmemb, size_t size);
static void * taosTRealloc(void *ptr, size_t size);
static void * taosTZfree(void *ptr);
static size_t taosTSizeof(void *ptr);
static void taosTMemset(void *ptr, int c);
static FORCE_INLINE void *taosTMalloc(size_t size) {
if (size <= 0) return NULL;
void *ret = malloc(size + sizeof(size_t));
if (ret == NULL) return NULL;
*(size_t *)ret = size;
return (void *)((char *)ret + sizeof(size_t));
}
static FORCE_INLINE void *taosTCalloc(size_t nmemb, size_t size) {
size_t tsize = nmemb * size;
void * ret = taosTMalloc(tsize);
if (ret == NULL) return NULL;
taosTMemset(ret, 0);
return ret;
}
static FORCE_INLINE size_t taosTSizeof(void *ptr) { return (ptr) ? (*(size_t *)((char *)ptr - sizeof(size_t))) : 0; }
static FORCE_INLINE void taosTMemset(void *ptr, int c) { memset(ptr, c, taosTSizeof(ptr)); }
static FORCE_INLINE void * taosTRealloc(void *ptr, size_t size) {
if (ptr == NULL) return taosTMalloc(size);
if (size <= taosTSizeof(ptr)) return ptr;
void * tptr = (void *)((char *)ptr - sizeof(size_t));
size_t tsize = size + sizeof(size_t);
void* tptr1 = realloc(tptr, tsize);
if (tptr1 == NULL) return NULL;
tptr = tptr1;
*(size_t *)tptr = size;
return (void *)((char *)tptr + sizeof(size_t));
}
static FORCE_INLINE void* taosTZfree(void* ptr) {
if (ptr) {
free((void*)((char*)ptr - sizeof(size_t)));
}
return NULL;
}
#endif /* _TD_TSDB_MEMORY_H_ */
\ No newline at end of file
source/dnode/vnode/tsdb/inc/tsdbReadImpl.h
浏览文件 @ 85a16137
......@@ -15,14 +15,14 @@
#ifndef _TD_TSDB_READ_IMPL_H_
#define _TD_TSDB_READ_IMPL_H_
#if 0
#include "os.h"
#include "tfs.h"
#include "tsdb.h"
#include "os.h"
#include "tsdbFile.h"
#include "tskiplist.h"
#include "tsdbMeta.h"
#include "tsdbMemory.h"
#include "common.h"
typedef struct SReadH SReadH;
......@@ -91,7 +91,7 @@ typedef struct {
} SBlockData;
struct SReadH {
STsdbRepo * pRepo;
STsdb * pRepo;
SDFileSet rSet; // FSET to read
SArray * aBlkIdx; // SBlockIdx array
STable * pTable; // table to read
......@@ -116,7 +116,7 @@ struct SReadH {
#define TSDB_BLOCK_STATIS_SIZE(ncols) (sizeof(SBlockData) + sizeof(SBlockCol) * (ncols) + sizeof(TSCKSUM))
int tsdbInitReadH(SReadH *pReadh, STsdbRepo *pRepo);
int tsdbInitReadH(SReadH *pReadh, STsdb *pRepo);
void tsdbDestroyReadH(SReadH *pReadh);
int tsdbSetAndOpenReadFSet(SReadH *pReadh, SDFileSet *pSet);
void tsdbCloseAndUnsetFSet(SReadH *pReadh);
......@@ -151,6 +151,4 @@ static FORCE_INLINE int tsdbMakeRoom(void **ppBuf, size_t size) {
return 0;
}
#endif
#endif /*_TD_TSDB_READ_IMPL_H_*/
source/dnode/vnode/tsdb/src/tsdbCommit.c
浏览文件 @ 85a16137
......@@ -32,6 +32,23 @@ int tsdbCommit(STsdb *pTsdb) {
return 0;
}
void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) {
STsdbCfg *pCfg = REPO_CFG(pRepo);
TSKEY minKey, midKey, maxKey, now;
now = taosGetTimestamp(pCfg->precision);
minKey = now - pCfg->keep * tsTickPerDay[pCfg->precision];
midKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision];
maxKey = now - pCfg->keep1 * tsTickPerDay[pCfg->precision];
pRtn->minKey = minKey;
pRtn->minFid = (int)(TSDB_KEY_FID(minKey, pCfg->daysPerFile, pCfg->precision));
pRtn->midFid = (int)(TSDB_KEY_FID(midKey, pCfg->daysPerFile, pCfg->precision));
pRtn->maxFid = (int)(TSDB_KEY_FID(maxKey, pCfg->daysPerFile, pCfg->precision));
tsdbDebug("vgId:%d now:%" PRId64 " minKey:%" PRId64 " minFid:%d, midFid:%d, maxFid:%d", REPO_ID(pRepo), now, minKey,
pRtn->minFid, pRtn->midFid, pRtn->maxFid);
}
#if 0
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
......@@ -420,23 +437,6 @@ void *tsdbDecodeKVRecord(void *buf, SKVRecord *pRecord) {
return buf;
}
void tsdbGetRtnSnap(STsdbRepo *pRepo, SRtn *pRtn) {
STsdbCfg *pCfg = REPO_CFG(pRepo);
TSKEY minKey, midKey, maxKey, now;
now = taosGetTimestamp(pCfg->precision);
minKey = now - pCfg->keep * tsTickPerDay[pCfg->precision];
midKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision];
maxKey = now - pCfg->keep1 * tsTickPerDay[pCfg->precision];
pRtn->minKey = minKey;
pRtn->minFid = (int)(TSDB_KEY_FID(minKey, pCfg->daysPerFile, pCfg->precision));
pRtn->midFid = (int)(TSDB_KEY_FID(midKey, pCfg->daysPerFile, pCfg->precision));
pRtn->maxFid = (int)(TSDB_KEY_FID(maxKey, pCfg->daysPerFile, pCfg->precision));
tsdbDebug("vgId:%d now:%" PRId64 " minKey:%" PRId64 " minFid:%d, midFid:%d, maxFid:%d", REPO_ID(pRepo), now, minKey,
pRtn->minFid, pRtn->midFid, pRtn->maxFid);
}
static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void *cont, int contLen, bool compact) {
char buf[64] = "\0";
void * pBuf = buf;
......
source/dnode/vnode/tsdb/src/tsdbFile.c
浏览文件 @ 85a16137
......@@ -13,7 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tsdbint.h"
#include "tsdbDef.h"
static const char *TSDB_FNAME_SUFFIX[] = {
"head", // TSDB_FILE_HEAD
......@@ -24,11 +24,12 @@ static const char *TSDB_FNAME_SUFFIX[] = {
};
static void tsdbGetFilename(int vid, int fid, uint32_t ver, TSDB_FILE_T ftype, char *fname);
static int tsdbRollBackMFile(SMFile *pMFile);
// static int tsdbRollBackMFile(SMFile *pMFile);
static int tsdbEncodeDFInfo(void **buf, SDFInfo *pInfo);
static void *tsdbDecodeDFInfo(void *buf, SDFInfo *pInfo);
static int tsdbRollBackDFile(SDFile *pDFile);
#if 0
// ============== SMFile
void tsdbInitMFile(SMFile *pMFile, SDiskID did, int vid, uint32_t ver) {
char fname[TSDB_FILENAME_LEN];
......@@ -185,7 +186,7 @@ int tsdbLoadMFileHeader(SMFile *pMFile, SMFInfo *pInfo) {
return 0;
}
int tsdbScanAndTryFixMFile(STsdbRepo *pRepo) {
int tsdbScanAndTryFixMFile(STsdb *pRepo) {
SMFile * pMFile = pRepo->fs->cstatus->pmf;
struct stat mfstat;
SMFile mf;
......@@ -291,6 +292,8 @@ static int tsdbRollBackMFile(SMFile *pMFile) {
return 0;
}
#endif
// ============== Operations on SDFile
void tsdbInitDFile(SDFile *pDFile, SDiskID did, int vid, int fid, uint32_t ver, TSDB_FILE_T ftype) {
char fname[TSDB_FILENAME_LEN];
......@@ -397,7 +400,7 @@ int tsdbUpdateDFileHeader(SDFile *pDFile) {
}
void *ptr = buf;
taosEncodeFixedU32(&ptr, TSDB_FS_VERSION);
taosEncodeFixedU32(&ptr, 0);
tsdbEncodeDFInfo(&ptr, &(pDFile->info));
taosCalcChecksumAppend(0, (uint8_t *)buf, TSDB_FILE_HEAD_SIZE);
......@@ -433,7 +436,7 @@ int tsdbLoadDFileHeader(SDFile *pDFile, SDFInfo *pInfo) {
return 0;
}
static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
static int tsdbScanAndTryFixDFile(STsdb *pRepo, SDFile *pDFile) {
struct stat dfstat;
SDFile df;
......@@ -442,7 +445,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
if (access(TSDB_FILE_FULL_NAME(pDFile), F_OK) != 0) {
tsdbError("vgId:%d data file %s not exit, report to upper layer to fix it", REPO_ID(pRepo),
TSDB_FILE_FULL_NAME(pDFile));
pRepo->state |= TSDB_STATE_BAD_DATA;
// pRepo->state |= TSDB_STATE_BAD_DATA;
TSDB_FILE_SET_STATE(pDFile, TSDB_FILE_STATE_BAD);
return 0;
}
......@@ -457,7 +460,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
return -1;
}
if (taosFtruncate(df.fd, df.info.size) < 0) {
if (taosFtruncateFile(df.fd, df.info.size) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
tsdbCloseDFile(&df);
return -1;
......@@ -474,7 +477,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
} else if (pDFile->info.size > dfstat.st_size) {
tsdbError("vgId:%d data file %s has wrong size %" PRId64 " expected %" PRId64 ", report to upper layer to fix it",
REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), dfstat.st_size, pDFile->info.size);
pRepo->state |= TSDB_STATE_BAD_DATA;
// pRepo->state |= TSDB_STATE_BAD_DATA;
TSDB_FILE_SET_STATE(pDFile, TSDB_FILE_STATE_BAD);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return 0;
......@@ -538,7 +541,7 @@ static int tsdbRollBackDFile(SDFile *pDFile) {
return -1;
}
if (taosFtruncate(TSDB_FILE_FD(&df), pDFile->info.size) < 0) {
if (taosFtruncateFile(TSDB_FILE_FD(&df), pDFile->info.size) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
tsdbCloseDFile(&df);
return -1;
......@@ -651,7 +654,7 @@ int tsdbUpdateDFileSetHeader(SDFileSet *pSet) {
return 0;
}
int tsdbScanAndTryFixDFileSet(STsdbRepo *pRepo, SDFileSet *pSet) {
int tsdbScanAndTryFixDFileSet(STsdb *pRepo, SDFileSet *pSet) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
if (tsdbScanAndTryFixDFile(pRepo, TSDB_DFILE_IN_SET(pSet, ftype)) < 0) {
return -1;
......
source/dnode/vnode/tsdb/src/tsdbMain.c
浏览文件 @ 85a16137
......@@ -72,7 +72,7 @@ static STsdb *tsdbNew(const char *path, const STsdbCfg *pTsdbCfg, SMemAllocatorF
}
pTsdb->path = strdup(path);
tsdbOptionsCopy(&(pTsdb->options), pTsdbCfg);
tsdbOptionsCopy(&(pTsdb->config), pTsdbCfg);
pTsdb->pmaf = pMAF;
return pTsdb;
......
source/dnode/vnode/tsdb2/CMakeLists.txt 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbBuffer.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbCommit.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbCompact.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbFS.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbFile.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbLog.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbMemTable.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbMemory.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbMeta.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbReadImpl.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbRowMergeBuf.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/inc/tsdbint.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbBuffer.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbCommit.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbCompact.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbFS.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbFile.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbMain.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbMemTable.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbMeta.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbRead.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbReadImpl.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbRecover.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/src/tsdbRowMergeBuf.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/tests/CMakeLists.txt 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/dnode/vnode/tsdb2/tests/tsdbTests.cpp 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/libs/CMakeLists.txt
浏览文件 @ 85a16137
......@@ -12,3 +12,4 @@ add_subdirectory(planner)
add_subdirectory(function)
add_subdirectory(qcom)
add_subdirectory(qworker)
add_subdirectory(tfs)
source/libs/catalog/src/catalog.c
浏览文件 @ 85a16137
此差异已折叠。
source/libs/executor/CMakeLists.txt
浏览文件 @ 85a16137
此差异已折叠。
source/libs/executor/inc/dataSinkInt.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/libs/executor/inc/dataSinkMgt.h 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/libs/executor/src/dataDispatcher.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/libs/executor/src/dataSinkMgt.c 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/libs/parser/inc/parserInt.h
浏览文件 @ 85a16137
此差异已折叠。
source/libs/parser/src/parser.c
浏览文件 @ 85a16137
此差异已折叠。
source/libs/planner/src/logicPlan.c
浏览文件 @ 85a16137
此差异已折叠。
source/libs/planner/src/planner.c
浏览文件 @ 85a16137
此差异已折叠。
source/libs/tfs/CMakeLists.txt 0 → 100644
浏览文件 @ 85a16137
此差异已折叠。
source/util/src/terror.c
浏览文件 @ 85a16137
此差异已折叠。
src/tfs/CMakeLists.txt 已删除 100644 → 0
浏览文件 @ 30556c67
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或