Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 10952778 编写于 作者: X Xiaoyu Wang
浏览文件
操作

Merge remote-tracking branch 'origin/feature/qnode' into feature/3.0_wxy

上级 a77a9c1f c1bbea8c
展开全部 隐藏空白更改
内联 并排
Showing with 2154 addition and 1671 deletion
include/common/tmsg.h
浏览文件 @ 10952778
......@@ -57,46 +57,6 @@ extern int tMsgDict[];
typedef uint16_t tmsg_t;
/* ------------------------ ENCODE/DECODE FUNCTIONS AND MACROS ------------------------ */
struct SMEListNode {
TD_SLIST_NODE(SMEListNode);
SEncoder coder;
};
typedef struct SMsgEncoder {
SEncoder coder;
TD_SLIST(SMEListNode) eStack; // encode stack
} SMsgEncoder;
struct SMDFreeListNode {
TD_SLIST_NODE(SMDFreeListNode);
char payload[];
};
struct SMDListNode {
TD_SLIST_NODE(SMDListNode);
SDecoder coder;
};
typedef struct SMsgDecoder {
SDecoder coder;
TD_SLIST(SMDListNode) dStack;
TD_SLIST(SMDFreeListNode) freeList;
} SMsgDecoder;
#define TMSG_MALLOC(SIZE, DECODER) \
({ \
void* ptr = malloc((SIZE) + sizeof(struct SMDFreeListNode)); \
if (ptr) { \
TD_SLIST_PUSH(&((DECODER)->freeList), (struct SMDFreeListNode*)ptr); \
ptr = POINTER_SHIFT(ptr, sizeof(struct SMDFreeListNode*)); \
} \
ptr; \
})
void tmsgInitMsgDecoder(SMsgDecoder* pMD, td_endian_t endian, uint8_t* data, int64_t size);
void tmsgClearMsgDecoder(SMsgDecoder* pMD);
/* ------------------------ OTHER DEFINITIONS ------------------------ */
// IE type
#define TSDB_IE_TYPE_SEC 1
......@@ -1090,6 +1050,7 @@ typedef struct SResFetchMsg {
} SResFetchMsg;
typedef struct SSchTasksStatusMsg {
SMsgHead header;
uint64_t sId;
} SSchTasksStatusMsg;
......@@ -1105,6 +1066,7 @@ typedef struct SSchedulerStatusRsp {
} SSchedulerStatusRsp;
typedef struct STaskCancelMsg {
SMsgHead header;
uint64_t sId;
uint64_t queryId;
uint64_t taskId;
......@@ -1115,6 +1077,7 @@ typedef struct STaskCancelRsp {
} STaskCancelRsp;
typedef struct STaskDropMsg {
SMsgHead header;
uint64_t sId;
uint64_t queryId;
uint64_t taskId;
......@@ -1283,8 +1246,6 @@ typedef struct {
SArray* pArray;
} SVCreateTbBatchReq;
int tmsgSVCreateTbReqEncode(SMsgEncoder* pCoder, SVCreateTbReq* pReq);
int tmsgSVCreateTbReqDecode(SMsgDecoder* pCoder, SVCreateTbReq* pReq);
int tSerializeSVCreateTbReq(void** buf, SVCreateTbReq* pReq);
void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq);
int tSVCreateTbBatchReqSerialize(void** buf, SVCreateTbBatchReq* pReq);
......
include/libs/planner/planner.h
浏览文件 @ 10952778
......@@ -120,11 +120,12 @@ typedef struct SSubplanId {
} SSubplanId;
typedef struct SSubplan {
SSubplanId id; // unique id of the subplan
SSubplanId id; // unique id of the subplan
int32_t type; // QUERY_TYPE_MERGE|QUERY_TYPE_PARTIAL|QUERY_TYPE_SCAN|QUERY_TYPE_MODIFY
int32_t level; // the execution level of current subplan, starting from 0.
int32_t msgType; // message type for subplan, used to denote the send message type to vnode.
int32_t level; // the execution level of current subplan, starting from 0 in a top-down manner.
SQueryNodeAddr execNode; // for the scan/modify subplan, the optional execution node
SArray *pChildern; // the datasource subplan,from which to fetch the result
SArray *pChildren; // the datasource subplan,from which to fetch the result
SArray *pParents; // the data destination subplan, get data from current subplan
SPhyNode *pNode; // physical plan of current subplan
SDataSink *pDataSink; // data of the subplan flow into the datasink
......@@ -147,7 +148,7 @@ int32_t qCreateQueryDag(const struct SQueryNode* pQueryInfo, struct SQueryDag**
// @subplan subplan to be schedule
// @templateId templateId of a group of datasource subplans of this @subplan
// @ep one execution location of this group of datasource subplans
int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SEpAddr* ep);
int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SQueryNodeAddr* ep);
int32_t qExplainQuery(const struct SQueryNode* pQueryInfo, struct SEpSet* pQnode, char** str);
......
include/util/encode.h
浏览文件 @ 10952778
此差异已折叠。
source/client/CMakeLists.txt
浏览文件 @ 10952778
......@@ -11,4 +11,6 @@ target_link_libraries(
PRIVATE os util common transport parser planner catalog scheduler function qcom
)
ADD_SUBDIRECTORY(test)
\ No newline at end of file
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
\ No newline at end of file
source/client/src/clientMain.c
浏览文件 @ 10952778
......@@ -29,7 +29,7 @@ int taos_options(TSDB_OPTION option, const void *arg, ...) {
// this function may be called by user or system, or by both simultaneously.
void taos_cleanup(void) {
tscDebug("start to cleanup client environment");
tscInfo("start to cleanup client environment");
if (atomic_val_compare_exchange_32(&sentinel, TSC_VAR_NOT_RELEASE, TSC_VAR_RELEASED) != TSC_VAR_NOT_RELEASE) {
return;
......@@ -47,6 +47,8 @@ void taos_cleanup(void) {
rpcCleanup();
taosCloseLog();
tscInfo("all local resources released");
}
TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port) {
......@@ -140,7 +142,9 @@ TAOS_ROW taos_fetch_row(TAOS_RES *pRes) {
SRequestObj *pRequest = (SRequestObj *) pRes;
if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
pRequest->type == TSDB_SQL_INSERT || pRequest->code != TSDB_CODE_SUCCESS) {
pRequest->type == TSDB_SQL_INSERT ||
pRequest->code != TSDB_CODE_SUCCESS ||
taos_num_fields(pRes) == 0) {
return NULL;
}
......
source/client/test/clientTests.cpp
浏览文件 @ 10952778
......@@ -279,7 +279,7 @@ TEST(testCase, connect_Test) {
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, create_table_Test) {
// // TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// // assert(pConn != NULL);
......@@ -470,9 +470,9 @@ TEST(testCase, create_multiple_tables) {
TAOS_RES* pRes = taos_query(pConn, "use abc1");
taos_free_result(pRes);
pRes = taos_query(pConn, "create table t_2 using st1 tags(1) t_3 using st1 tags(2)");
pRes = taos_query(pConn, "create table t_2 using st1 tags(1)");
if (taos_errno(pRes) != 0) {
printf("failed to show vgroups, reason:%s\n", taos_errstr(pRes));
printf("failed to create multiple tables, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
ASSERT_TRUE(false);
}
......
source/common/CMakeLists.txt
浏览文件 @ 10952778
......@@ -12,4 +12,6 @@ target_link_libraries(
INTERFACE api
)
ADD_SUBDIRECTORY(test)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
source/common/src/tmsg.c
浏览文件 @ 10952778
......@@ -27,77 +27,6 @@
#undef TD_MSG_SEG_CODE_
#include "tmsgdef.h"
static int tmsgStartEncode(SMsgEncoder *pME);
static void tmsgEndEncode(SMsgEncoder *pME);
static int tmsgStartDecode(SMsgDecoder *pMD);
static void tmsgEndDecode(SMsgDecoder *pMD);
/* ------------------------ ENCODE/DECODE FUNCTIONS ------------------------ */
void tmsgInitMsgEncoder(SMsgEncoder *pME, td_endian_t endian, uint8_t *data, int64_t size) {
tInitEncoder(&(pME->coder), endian, data, size);
TD_SLIST_INIT(&(pME->eStack));
}
void tmsgClearMsgEncoder(SMsgEncoder *pME) {
struct SMEListNode *pNode;
for (;;) {
pNode = TD_SLIST_HEAD(&(pME->eStack));
if (TD_IS_NULL(pNode)) break;
TD_SLIST_POP(&(pME->eStack));
free(pNode);
}
}
void tmsgInitMsgDecoder(SMsgDecoder *pMD, td_endian_t endian, uint8_t *data, int64_t size) {
tInitDecoder(&pMD->coder, endian, data, size);
TD_SLIST_INIT(&(pMD->dStack));
TD_SLIST_INIT(&(pMD->freeList));
}
void tmsgClearMsgDecoder(SMsgDecoder *pMD) {
{
struct SMDFreeListNode *pNode;
for (;;) {
pNode = TD_SLIST_HEAD(&(pMD->freeList));
if (TD_IS_NULL(pNode)) break;
TD_SLIST_POP(&(pMD->freeList));
free(pNode);
}
}
{
struct SMDListNode *pNode;
for (;;) {
pNode = TD_SLIST_HEAD(&(pMD->dStack));
if (TD_IS_NULL(pNode)) break;
TD_SLIST_POP(&(pMD->dStack));
free(pNode);
}
}
}
/* ------------------------ MESSAGE ENCODE/DECODE ------------------------ */
int tmsgSVCreateTbReqEncode(SMsgEncoder *pCoder, SVCreateTbReq *pReq) {
tmsgStartEncode(pCoder);
// TODO
tmsgEndEncode(pCoder);
return 0;
}
int tmsgSVCreateTbReqDecode(SMsgDecoder *pCoder, SVCreateTbReq *pReq) {
tmsgStartDecode(pCoder);
// TODO: decode
// Decode is not end
if (pCoder->coder.pos != pCoder->coder.size) {
// Continue decode
}
tmsgEndDecode(pCoder);
return 0;
}
int tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
int tlen = 0;
......@@ -218,64 +147,4 @@ void *tSVCreateTbBatchReqDeserialize(void *buf, SVCreateTbBatchReq *pReq) {
}
return buf;
}
/* ------------------------ STATIC METHODS ------------------------ */
static int tmsgStartEncode(SMsgEncoder *pME) {
struct SMEListNode *pNode = (struct SMEListNode *)malloc(sizeof(*pNode));
if (TD_IS_NULL(pNode)) return -1;
pNode->coder = pME->coder;
TD_SLIST_PUSH(&(pME->eStack), pNode);
TD_CODER_MOVE_POS(&(pME->coder), sizeof(int32_t));
return 0;
}
static void tmsgEndEncode(SMsgEncoder *pME) {
int32_t size;
struct SMEListNode *pNode;
pNode = TD_SLIST_HEAD(&(pME->eStack));
ASSERT(pNode);
TD_SLIST_POP(&(pME->eStack));
size = pME->coder.pos - pNode->coder.pos;
tEncodeI32(&(pNode->coder), size);
free(pNode);
}
static int tmsgStartDecode(SMsgDecoder *pMD) {
struct SMDListNode *pNode;
int32_t size;
pNode = (struct SMDListNode *)malloc(sizeof(*pNode));
if (pNode == NULL) return -1;
tDecodeI32(&(pMD->coder), &size);
pNode->coder = pMD->coder;
TD_SLIST_PUSH(&(pMD->dStack), pNode);
pMD->coder.pos = 0;
pMD->coder.size = size - sizeof(int32_t);
pMD->coder.data = TD_CODER_CURRENT(&(pNode->coder));
return 0;
}
static void tmsgEndDecode(SMsgDecoder *pMD) {
ASSERT(pMD->coder.pos == pMD->coder.size);
struct SMDListNode *pNode;
pNode = TD_SLIST_HEAD(&(pMD->dStack));
ASSERT(pNode);
TD_SLIST_POP(&(pMD->dStack));
pNode->coder.pos += pMD->coder.size;
pMD->coder = pNode->coder;
free(pNode);
}
\ No newline at end of file
source/libs/catalog/CMakeLists.txt
浏览文件 @ 10952778
......@@ -11,4 +11,6 @@ target_link_libraries(
PRIVATE os util transport qcom
)
ADD_SUBDIRECTORY(test)
\ No newline at end of file
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
\ No newline at end of file
source/libs/index/inc/index_fst_counting_writer.h
浏览文件 @ 10952778
......@@ -38,6 +38,7 @@ typedef struct WriterCtx {
int fd;
bool readOnly;
char buf[256];
int size;
} file;
struct {
int32_t capa;
......
source/libs/index/src/index.c
浏览文件 @ 10952778
......@@ -73,6 +73,7 @@ int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
#ifdef USE_INVERTED_INDEX
// sIdx->cache = (void*)indexCacheCreate(sIdx);
sIdx->tindex = indexTFileCreate(path);
if (sIdx->tindex == NULL) { goto END; }
sIdx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
sIdx->cVersion = 1;
sIdx->path = calloc(1, strlen(path) + 1);
......@@ -83,6 +84,8 @@ int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
return 0;
#endif
END:
if (sIdx != NULL) { indexClose(sIdx); }
*index = NULL;
return -1;
......@@ -135,7 +138,7 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) {
SIndexTerm* p = taosArrayGetP(fVals, i);
char buf[128] = {0};
ICacheKey key = {.suid = p->suid, .colName = p->colName};
ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName)};
int32_t sz = indexSerialCacheKey(&key, buf);
IndexCache** cache = taosHashGet(index->colObj, buf, sz);
......@@ -150,7 +153,7 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) {
SIndexTerm* p = taosArrayGetP(fVals, i);
char buf[128] = {0};
ICacheKey key = {.suid = p->suid, .colName = p->colName};
ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName)};
int32_t sz = indexSerialCacheKey(&key, buf);
IndexCache** cache = taosHashGet(index->colObj, buf, sz);
......@@ -212,7 +215,7 @@ int indexSearch(SIndex* index, SIndexMultiTermQuery* multiQuerys, SArray* result
indexInterResultsDestroy(interResults);
#endif
return 1;
return 0;
}
int indexDelete(SIndex* index, SIndexMultiTermQuery* query) {
......@@ -310,7 +313,7 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result
pthread_mutex_lock(&sIdx->mtx);
char buf[128] = {0};
ICacheKey key = {.suid = term->suid, .colName = term->colName};
ICacheKey key = {.suid = term->suid, .colName = term->colName, .nColName = strlen(term->colName)};
int32_t sz = indexSerialCacheKey(&key, buf);
IndexCache** pCache = taosHashGet(sIdx->colObj, buf, sz);
......
source/libs/index/src/index_cache.c
浏览文件 @ 10952778
......@@ -20,7 +20,7 @@
#define MAX_INDEX_KEY_LEN 256 // test only, change later
#define MEM_TERM_LIMIT 10 * 10000
#define MEM_TERM_LIMIT 5 * 10000
// ref index_cache.h:22
//#define CACHE_KEY_LEN(p) \
// (sizeof(int32_t) + sizeof(uint16_t) + sizeof(p->colType) + sizeof(p->nColVal) + p->nColVal + sizeof(uint64_t) +
......
source/libs/index/src/index_fst_counting_writer.c
浏览文件 @ 10952778
......@@ -72,9 +72,17 @@ WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int
if (readOnly == false) {
// ctx->file.fd = open(path, O_WRONLY | O_CREAT | O_APPEND, S_IRWXU | S_IRWXG | S_IRWXO);
ctx->file.fd = tfOpenCreateWriteAppend(path);
struct stat fstat;
stat(path, &fstat);
ctx->file.size = fstat.st_size;
} else {
// ctx->file.fd = open(path, O_RDONLY, S_IRWXU | S_IRWXG | S_IRWXO);
ctx->file.fd = tfOpenRead(path);
struct stat fstat;
stat(path, &fstat);
ctx->file.size = fstat.st_size;
}
memcpy(ctx->file.buf, path, strlen(path));
if (ctx->file.fd < 0) {
......@@ -104,6 +112,7 @@ void writerCtxDestroy(WriterCtx* ctx, bool remove) {
free(ctx->mem.buf);
} else {
tfClose(ctx->file.fd);
ctx->flush(ctx);
if (remove) { unlink(ctx->file.buf); }
}
free(ctx);
......
source/libs/index/src/index_tfile.c
浏览文件 @ 10952778
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
p *
* 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.
......@@ -45,12 +45,13 @@ static int tfileReaderLoadHeader(TFileReader* reader);
static int tfileReaderLoadFst(TFileReader* reader);
static int tfileReaderLoadTableIds(TFileReader* reader, int32_t offset, SArray* result);
static int tfileGetFileList(const char* path, SArray* result);
static int tfileRmExpireFile(SArray* result);
static void tfileDestroyFileName(void* elem);
static int tfileCompare(const void* a, const void* b);
static int tfileParseFileName(const char* filename, uint64_t* suid, int* colId, int* version);
static void tfileGenFileName(char* filename, uint64_t suid, int colId, int version);
static SArray* tfileGetFileList(const char* path);
static int tfileRmExpireFile(SArray* result);
static void tfileDestroyFileName(void* elem);
static int tfileCompare(const void* a, const void* b);
static int tfileParseFileName(const char* filename, uint64_t* suid, char* col, int* version);
static void tfileGenFileName(char* filename, uint64_t suid, const char* col, int version);
static void tfileGenFileFullName(char* fullname, const char* path, uint64_t suid, const char* col, int32_t version);
TFileCache* tfileCacheCreate(const char* path) {
TFileCache* tcache = calloc(1, sizeof(TFileCache));
......@@ -59,21 +60,24 @@ TFileCache* tfileCacheCreate(const char* path) {
tcache->tableCache = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
tcache->capacity = 64;
SArray* files = taosArrayInit(4, sizeof(void*));
tfileGetFileList(path, files);
taosArraySort(files, tfileCompare);
tfileRmExpireFile(files);
SArray* files = tfileGetFileList(path);
uint64_t suid;
int32_t colId, version;
for (size_t i = 0; i < taosArrayGetSize(files); i++) {
char* file = taosArrayGetP(files, i);
if (0 != tfileParseFileName(file, &suid, (int*)&colId, (int*)&version)) {
// refactor later, use colname and version info
char colName[256] = {0};
if (0 != tfileParseFileName(file, &suid, colName, (int*)&version)) {
indexInfo("try parse invalid file: %s, skip it", file);
continue;
}
WriterCtx* wc = writerCtxCreate(TFile, file, true, 1024 * 1024 * 64);
char fullName[256] = {0};
sprintf(fullName, "%s/%s", path, file);
WriterCtx* wc = writerCtxCreate(TFile, fullName, true, 1024 * 1024 * 64);
if (wc == NULL) {
indexError("failed to open index:%s", file);
goto End;
......@@ -200,12 +204,9 @@ int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* resul
}
TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int32_t version, const char* colName, uint8_t colType) {
char filename[128] = {0};
int32_t coldId = 1;
tfileGenFileName(filename, suid, coldId, version);
char fullname[256] = {0};
snprintf(fullname, sizeof(fullname), "%s/%s", path, filename);
tfileGenFileFullName(fullname, path, suid, colName, version);
// indexInfo("open write file name %s", fullname);
WriterCtx* wcx = writerCtxCreate(TFile, fullname, false, 1024 * 1024 * 64);
if (wcx == NULL) { return NULL; }
......@@ -218,13 +219,11 @@ TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int32_t version, const c
return tfileWriterCreate(wcx, &tfh);
}
TFileReader* tfileReaderOpen(char* path, uint64_t suid, int32_t version, const char* colName) {
char filename[128] = {0};
int32_t coldId = 1;
tfileGenFileName(filename, suid, coldId, version);
char fullname[256] = {0};
snprintf(fullname, sizeof(fullname), "%s/%s", path, filename);
tfileGenFileFullName(fullname, path, suid, colName, version);
WriterCtx* wc = writerCtxCreate(TFile, fullname, true, 1024 * 1024 * 1024);
// indexInfo("open read file name:%s, size: %d", wc->file.buf, wc->file.size);
if (wc == NULL) { return NULL; }
TFileReader* reader = tfileReaderCreate(wc);
......@@ -324,7 +323,6 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
}
// write data
indexError("--------Begin----------------");
for (size_t i = 0; i < sz; i++) {
// TODO, fst batch write later
TFileValue* v = taosArrayGetP((SArray*)data, i);
......@@ -332,11 +330,10 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
indexError("failed to write data: %s, offset: %d len: %d", v->colVal, v->offset,
(int)taosArrayGetSize(v->tableId));
} else {
indexInfo("success to write data: %s, offset: %d len: %d", v->colVal, v->offset,
(int)taosArrayGetSize(v->tableId));
// indexInfo("success to write data: %s, offset: %d len: %d", v->colVal, v->offset,
// (int)taosArrayGetSize(v->tableId));
}
}
indexError("--------End----------------");
fstBuilderFinish(tw->fb);
fstBuilderDestroy(tw->fb);
tw->fb = NULL;
......@@ -361,6 +358,7 @@ IndexTFile* indexTFileCreate(const char* path) {
return tfile;
}
void indexTFileDestroy(IndexTFile* tfile) {
if (tfile == NULL) { return; }
tfileCacheDestroy(tfile->cache);
free(tfile);
}
......@@ -550,6 +548,9 @@ static int tfileReaderLoadHeader(TFileReader* reader) {
//
indexError("actual Read: %d, to read: %d, errno: %d, filefd: %d, filename: %s", (int)(nread), (int)sizeof(buf),
errno, reader->ctx->file.fd, reader->ctx->file.buf);
} else {
indexError("actual Read: %d, to read: %d, errno: %d, filefd: %d, filename: %s", (int)(nread), (int)sizeof(buf),
errno, reader->ctx->file.fd, reader->ctx->file.buf);
}
// assert(nread == sizeof(buf));
memcpy(&reader->header, buf, sizeof(buf));
......@@ -558,13 +559,14 @@ static int tfileReaderLoadHeader(TFileReader* reader) {
}
static int tfileReaderLoadFst(TFileReader* reader) {
// current load fst into memory, refactor it later
static int FST_MAX_SIZE = 64 * 1024;
static int FST_MAX_SIZE = 64 * 1024 * 1024;
char* buf = calloc(1, sizeof(char) * FST_MAX_SIZE);
if (buf == NULL) { return -1; }
WriterCtx* ctx = reader->ctx;
int32_t nread = ctx->readFrom(ctx, buf, FST_MAX_SIZE, reader->header.fstOffset);
indexError("nread = %d, and fst offset=%d, filename: %s ", nread, reader->header.fstOffset, ctx->file.buf);
// we assuse fst size less than FST_MAX_SIZE
assert(nread > 0 && nread < FST_MAX_SIZE);
......@@ -608,19 +610,26 @@ void tfileReaderUnRef(TFileReader* reader) {
}
}
static int tfileGetFileList(const char* path, SArray* result) {
static SArray* tfileGetFileList(const char* path) {
SArray* files = taosArrayInit(4, sizeof(void*));
DIR* dir = opendir(path);
if (NULL == dir) { return -1; }
if (NULL == dir) { return NULL; }
struct dirent* entry;
while ((entry = readdir(dir)) != NULL) {
if (entry->d_type && DT_DIR) { continue; }
size_t len = strlen(entry->d_name);
char* buf = calloc(1, len + 1);
memcpy(buf, entry->d_name, len);
taosArrayPush(result, &buf);
taosArrayPush(files, &buf);
}
closedir(dir);
return 0;
taosArraySort(files, tfileCompare);
tfileRmExpireFile(files);
return files;
}
static int tfileRmExpireFile(SArray* result) {
// TODO(yihao): remove expire tindex after restart
......@@ -641,15 +650,21 @@ static int tfileCompare(const void* a, const void* b) {
if (ret == 0) { return ret; }
return ret < 0 ? -1 : 1;
}
// tfile name suid-colId-version.tindex
static void tfileGenFileName(char* filename, uint64_t suid, int colId, int version) {
sprintf(filename, "%" PRIu64 "-%d-%d.tindex", suid, colId, version);
return;
}
static int tfileParseFileName(const char* filename, uint64_t* suid, int* colId, int* version) {
if (3 == sscanf(filename, "%" PRIu64 "-%d-%d.tindex", suid, colId, version)) {
static int tfileParseFileName(const char* filename, uint64_t* suid, char* col, int* version) {
if (3 == sscanf(filename, "%" PRIu64 "-%[^-]-%d.tindex", suid, col, version)) {
// read suid & colid & version success
return 0;
}
return -1;
}
// tfile name suid-colId-version.tindex
static void tfileGenFileName(char* filename, uint64_t suid, const char* col, int version) {
sprintf(filename, "%" PRIu64 "-%s-%d.tindex", suid, col, version);
return;
}
static void tfileGenFileFullName(char* fullname, const char* path, uint64_t suid, const char* col, int32_t version) {
char filename[128] = {0};
tfileGenFileName(filename, suid, col, version);
sprintf(fullname, "%s/%s", path, filename);
}
source/libs/index/test/indexTests.cc
浏览文件 @ 10952778
......@@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
#include <algorithm>
#include <iostream>
#include <string>
#include <thread>
......@@ -638,7 +639,7 @@ class IndexObj {
indexInit();
}
int Init(const std::string& dir) {
taosRemoveDir(dir.c_str());
// taosRemoveDir(dir.c_str());
taosMkDir(dir.c_str());
int ret = indexOpen(&opts, dir.c_str(), &idx);
if (ret != 0) {
......@@ -663,10 +664,11 @@ class IndexObj {
int WriteMultiMillonData(const std::string& colName, const std::string& colVal = "Hello world",
size_t numOfTable = 100 * 10000) {
std::string tColVal = colVal;
size_t colValSize = tColVal.size();
for (int i = 0; i < numOfTable; i++) {
tColVal[tColVal.size() - 1] = 'a' + i % 26;
tColVal[i % colValSize] = 'a' + i % 26;
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
tColVal.c_str(), tColVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t i = 0; i < 10; i++) {
......@@ -695,7 +697,13 @@ class IndexObj {
indexMultiTermQueryAdd(mq, term, QUERY_TERM);
SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t));
if (Search(mq, result) == 0) { std::cout << "search one successfully" << std::endl; }
int64_t s = taosGetTimestampUs();
if (Search(mq, result) == 0) {
int64_t e = taosGetTimestampUs();
std::cout << "search one successfully and time cost:" << e - s << std::endl;
} else {
}
int sz = taosArrayGetSize(result);
indexMultiTermQueryDestroy(mq);
taosArrayDestroy(result);
......@@ -810,7 +818,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
}
TEST_F(IndexEnv2, testIndex_TrigeFlush) {
std::string path = "/tmp/test1";
std::string path = "/tmp/testxxx";
if (index->Init(path) != 0) {
// r
std::cout << "failed to init" << std::endl;
......@@ -826,6 +834,10 @@ static void write_and_search(IndexObj* idx) {
std::string colName("tag1"), colVal("Hello");
int target = idx->SearchOne("tag1", "Hello");
std::cout << "search: " << target << std::endl;
target = idx->SearchOne("tag2", "Test");
std::cout << "search: " << target << std::endl;
idx->PutOne(colName, colVal);
}
TEST_F(IndexEnv2, testIndex_serarch_cache_and_tfile) {
......@@ -833,7 +845,10 @@ TEST_F(IndexEnv2, testIndex_serarch_cache_and_tfile) {
if (index->Init(path) != 0) {
// opt
}
index->WriteMultiMillonData("tag1", "Hello", 200000);
index->PutOne("tag1", "Hello");
index->PutOne("tag2", "Test");
index->WriteMultiMillonData("tag1", "Hello", 50 * 10000);
index->WriteMultiMillonData("tag2", "Test", 50 * 10000);
std::thread threads[NUM_OF_THREAD];
for (int i = 0; i < NUM_OF_THREAD; i++) {
......@@ -847,15 +862,15 @@ TEST_F(IndexEnv2, testIndex_serarch_cache_and_tfile) {
}
TEST_F(IndexEnv2, testIndex_restart) {
std::string path = "/tmp";
std::string path = "/tmp/test1";
if (index->Init(path) != 0) {}
}
TEST_F(IndexEnv2, testIndex_performance) {
std::string path = "/tmp";
std::string path = "/tmp/test2";
if (index->Init(path) != 0) {}
}
TEST_F(IndexEnv2, testIndexMultiTag) {
std::string path = "/tmp";
std::string path = "/tmp/test3";
if (index->Init(path) != 0) {}
}
source/libs/parser/CMakeLists.txt
浏览文件 @ 10952778
......@@ -11,4 +11,6 @@ target_link_libraries(
PRIVATE os util catalog function transport qcom
)
ADD_SUBDIRECTORY(test)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
source/libs/parser/inc/sql.y
浏览文件 @ 10952778
......@@ -392,7 +392,7 @@ create_table_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) LP columnlist(X) RP. {
%type create_stable_args{SCreateTableSql*}
create_stable_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) LP columnlist(X) RP TAGS LP columnlist(Y) RP. {
A = tSetCreateTableInfo(X, Y, NULL, TSQL_CREATE_STABLE);
setSqlInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE);
setSqlInfo(pInfo, A, NULL, TSDB_SQL_CREATE_STABLE);
V.n += Z.n;
setCreatedTableName(pInfo, &V, &U);
......
source/libs/parser/src/astGenerator.c
浏览文件 @ 10952778
......@@ -785,7 +785,7 @@ void destroySqlInfo(SSqlInfo *pInfo) {
taosArrayDestroy(pInfo->funcs);
if (pInfo->type == TSDB_SQL_SELECT) {
destroyAllSqlNode(&pInfo->sub);
} else if (pInfo->type == TSDB_SQL_CREATE_TABLE) {
} else if (pInfo->type == TSDB_SQL_CREATE_STABLE) {
pInfo->pCreateTableInfo = destroyCreateTableSql(pInfo->pCreateTableInfo);
} else if (pInfo->type == TSDB_SQL_ALTER_TABLE) {
taosArrayDestroyEx(pInfo->pAlterInfo->varList, freeItem);
......
source/libs/parser/src/dCDAstProcess.c
浏览文件 @ 10952778
......@@ -761,8 +761,9 @@ SDclStmtInfo* qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, c
break;
}
case TSDB_SQL_CREATE_TABLE: {
case TSDB_SQL_CREATE_STABLE: {
SCreateTableSql* pCreateTable = pInfo->pCreateTableInfo;
assert(pCreateTable->type != TSQL_CREATE_CTABLE);
if (pCreateTable->type == TSQL_CREATE_TABLE || pCreateTable->type == TSQL_CREATE_STABLE) {
if ((code = doCheckForCreateTable(pInfo, pMsgBuf)) != TSDB_CODE_SUCCESS) {
......@@ -772,13 +773,6 @@ SDclStmtInfo* qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, c
pDcl->pMsg = (char*)buildCreateTableMsg(pCreateTable, &pDcl->msgLen, pCtx, pMsgBuf);
pDcl->msgType = (pCreateTable->type == TSQL_CREATE_TABLE) ? TDMT_VND_CREATE_TABLE : TDMT_MND_CREATE_STB;
} else if (pCreateTable->type == TSQL_CREATE_CTABLE) {
if ((code = doCheckForCreateCTable(pInfo, pCtx, pMsgBuf, &pDcl->pMsg, &pDcl->msgLen)) !=
TSDB_CODE_SUCCESS) {
goto _error;
}
pDcl->msgType = TDMT_VND_CREATE_TABLE;
} else if (pCreateTable->type == TSQL_CREATE_STREAM) {
// if ((code = doCheckForStream(pSql, pInfo)) != TSDB_CODE_SUCCESS) {
// return code;
......
source/libs/parser/src/sql.c
浏览文件 @ 10952778
此差异已折叠。
source/libs/planner/CMakeLists.txt
浏览文件 @ 10952778
......@@ -11,4 +11,6 @@ target_link_libraries(
PRIVATE os util catalog cjson parser transport function qcom
)
ADD_SUBDIRECTORY(test)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
source/libs/planner/inc/plannerInt.h
浏览文件 @ 10952778
......@@ -70,6 +70,11 @@ typedef struct SQueryPlanNode {
struct SQueryPlanNode *pParent;
} SQueryPlanNode;
typedef struct SDataPayloadInfo {
int32_t msgType;
SArray *payload;
} SDataPayloadInfo;
/**
* Optimize the query execution plan, currently not implement yet.
* @param pQueryNode
......@@ -101,7 +106,7 @@ int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str);
int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql);
int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag);
int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SEpAddr* ep);
int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SQueryNodeAddr* ep);
int32_t subPlanToString(const SSubplan *pPhyNode, char** str, int32_t* len);
int32_t stringToSubplan(const char* str, SSubplan** subplan);
......
source/libs/planner/src/logicPlan.c
浏览文件 @ 10952778
......@@ -37,18 +37,28 @@ int32_t optimizeQueryPlan(struct SQueryPlanNode* pQueryNode) {
return 0;
}
static int32_t createInsertPlan(const SQueryNode* pNode, SQueryPlanNode** pQueryPlan) {
static int32_t createModificationOpPlan(const SQueryNode* pNode, SQueryPlanNode** pQueryPlan) {
SInsertStmtInfo* pInsert = (SInsertStmtInfo*)pNode;
*pQueryPlan = calloc(1, sizeof(SQueryPlanNode));
SArray* blocks = taosArrayInit(taosArrayGetSize(pInsert->pDataBlocks), POINTER_BYTES);
if (NULL == *pQueryPlan || NULL == blocks) {
SDataPayloadInfo* pPayload = calloc(1, sizeof(SDataPayloadInfo));
if (NULL == *pQueryPlan || NULL == blocks || NULL == pPayload) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
(*pQueryPlan)->info.type = QNODE_MODIFY;
taosArrayAddAll(blocks, pInsert->pDataBlocks);
(*pQueryPlan)->pExtInfo = blocks;
if (pNode->type == TSDB_SQL_INSERT) {
pPayload->msgType = TDMT_VND_SUBMIT;
} else if (pNode->type == TSDB_SQL_CREATE_TABLE) {
pPayload->msgType = TDMT_VND_CREATE_TABLE;
}
pPayload->payload = blocks;
(*pQueryPlan)->pExtInfo = pPayload;
return TSDB_CODE_SUCCESS;
}
......@@ -69,7 +79,7 @@ int32_t createQueryPlan(const SQueryNode* pNode, SQueryPlanNode** pQueryPlan) {
case TSDB_SQL_INSERT:
case TSDB_SQL_CREATE_TABLE:
return createInsertPlan(pNode, pQueryPlan);
return createModificationOpPlan(pNode, pQueryPlan);
default:
return TSDB_CODE_FAILED;
......
source/libs/planner/src/physicalPlan.c
浏览文件 @ 10952778
......@@ -191,13 +191,15 @@ static SSubplan* initSubplan(SPlanContext* pCxt, int32_t type) {
subplan->level = 0;
if (NULL != pCxt->pCurrentSubplan) {
subplan->level = pCxt->pCurrentSubplan->level + 1;
if (NULL == pCxt->pCurrentSubplan->pChildern) {
pCxt->pCurrentSubplan->pChildern = validPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
if (NULL == pCxt->pCurrentSubplan->pChildren) {
pCxt->pCurrentSubplan->pChildren = validPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
}
taosArrayPush(pCxt->pCurrentSubplan->pChildern, &subplan);
taosArrayPush(pCxt->pCurrentSubplan->pChildren, &subplan);
subplan->pParents = validPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
taosArrayPush(subplan->pParents, &pCxt->pCurrentSubplan);
}
SArray* currentLevel;
if (subplan->level >= taosArrayGetSize(pCxt->pDag->pSubplans)) {
currentLevel = validPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
......@@ -205,6 +207,7 @@ static SSubplan* initSubplan(SPlanContext* pCxt, int32_t type) {
} else {
currentLevel = taosArrayGetP(pCxt->pDag->pSubplans, subplan->level);
}
taosArrayPush(currentLevel, &subplan);
pCxt->pCurrentSubplan = subplan;
++(pCxt->pDag->numOfSubplans);
......@@ -278,6 +281,7 @@ static SPhyNode* createPhyNode(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
default:
assert(false);
}
if (pPlanNode->pChildren != NULL && taosArrayGetSize(pPlanNode->pChildren) > 0) {
node->pChildren = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
size_t size = taosArrayGetSize(pPlanNode->pChildren);
......@@ -287,31 +291,38 @@ static SPhyNode* createPhyNode(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
taosArrayPush(node->pChildren, &child);
}
}
return node;
}
static void splitInsertSubplan(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
SArray* vgs = (SArray*)pPlanNode->pExtInfo;
size_t numOfVg = taosArrayGetSize(vgs);
for (int32_t i = 0; i < numOfVg; ++i) {
static void splitModificationOpSubPlan(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
SDataPayloadInfo* pPayload = (SDataPayloadInfo*) pPlanNode->pExtInfo;
size_t numOfVgroups = taosArrayGetSize(pPayload->payload);
for (int32_t i = 0; i < numOfVgroups; ++i) {
STORE_CURRENT_SUBPLAN(pCxt);
SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_MODIFY);
SVgDataBlocks* blocks = (SVgDataBlocks*)taosArrayGetP(vgs, i);
SVgDataBlocks* blocks = (SVgDataBlocks*)taosArrayGetP(pPayload->payload, i);
vgroupInfoToEpSet(&blocks->vg, &subplan->execNode);
subplan->pNode = NULL;
subplan->pDataSink = createDataInserter(pCxt, blocks);
subplan->type = QUERY_TYPE_MODIFY;
subplan->pNode = NULL;
subplan->type = QUERY_TYPE_MODIFY;
subplan->msgType = pPayload->msgType;
RECOVERY_CURRENT_SUBPLAN(pCxt);
}
}
static void createSubplanByLevel(SPlanContext* pCxt, SQueryPlanNode* pRoot) {
if (QNODE_MODIFY == pRoot->info.type) {
splitInsertSubplan(pCxt, pRoot);
splitModificationOpSubPlan(pCxt, pRoot);
} else {
SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_MERGE);
SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_MERGE);
++(pCxt->nextId.templateId);
subplan->pNode = createPhyNode(pCxt, pRoot);
subplan->msgType = TDMT_VND_QUERY;
subplan->pNode = createPhyNode(pCxt, pRoot);
subplan->pDataSink = createDataDispatcher(pCxt, pRoot);
}
// todo deal subquery
......@@ -325,6 +336,7 @@ int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryD
.pCurrentSubplan = NULL,
.nextId = {0} // todo queryid
};
*pDag = context.pDag;
context.pDag->pSubplans = validPointer(taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES));
createSubplanByLevel(&context, pQueryNode);
......@@ -336,6 +348,6 @@ int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryD
return TSDB_CODE_SUCCESS;
}
int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SEpAddr* ep) {
int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SQueryNodeAddr* ep) {
//todo
}
source/libs/planner/src/physicalPlanJson.c
浏览文件 @ 10952778
......@@ -793,7 +793,7 @@ static cJSON* subplanToJson(const SSubplan* subplan) {
return NULL;
}
// The 'type', 'level', 'execEpSet', 'pChildern' and 'pParents' fields do not need to be serialized.
// The 'type', 'level', 'execEpSet', 'pChildren' and 'pParents' fields do not need to be serialized.
bool res = addObject(jSubplan, jkSubplanId, subplanIdToJson, &subplan->id);
if (res) {
......@@ -835,7 +835,7 @@ int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
SDataInserter* insert = (SDataInserter*)(subplan->pDataSink);
*len = insert->size;
*str = insert->pData;
insert->pData == NULL;
insert->pData = NULL;
return TSDB_CODE_SUCCESS;
}
......@@ -844,6 +844,7 @@ int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
return TSDB_CODE_FAILED;
}
*str = cJSON_Print(json);
*len = strlen(*str) + 1;
return TSDB_CODE_SUCCESS;
......
source/libs/planner/src/planner.c
浏览文件 @ 10952778
......@@ -31,22 +31,25 @@ int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag)
destroyQueryPlan(logicPlan);
return code;
}
code = optimizeQueryPlan(logicPlan);
if (TSDB_CODE_SUCCESS != code) {
destroyQueryPlan(logicPlan);
return code;
}
code = createDag(logicPlan, NULL, pDag);
if (TSDB_CODE_SUCCESS != code) {
destroyQueryPlan(logicPlan);
qDestroyQueryDag(*pDag);
return code;
}
destroyQueryPlan(logicPlan);
return TSDB_CODE_SUCCESS;
}
int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SEpAddr* ep) {
int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SQueryNodeAddr* ep) {
return setSubplanExecutionNode(subplan, templateId, ep);
}
......
source/libs/qcom/CMakeLists.txt
浏览文件 @ 10952778
......@@ -7,8 +7,10 @@ target_include_directories(
)
target_link_libraries(
qcom
PRIVATE os util transport
qcom
PRIVATE os util transport
)
ADD_SUBDIRECTORY(test)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
source/libs/qworker/CMakeLists.txt
浏览文件 @ 10952778
......@@ -11,4 +11,6 @@ target_link_libraries(
PRIVATE os util transport planner qcom
)
ADD_SUBDIRECTORY(test)
\ No newline at end of file
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
\ No newline at end of file
source/libs/scheduler/CMakeLists.txt
浏览文件 @ 10952778
......@@ -12,4 +12,6 @@ target_link_libraries(
PRIVATE os util planner qcom common catalog transport
)
ADD_SUBDIRECTORY(test)
\ No newline at end of file
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
\ No newline at end of file
source/libs/scheduler/inc/schedulerInt.h
浏览文件 @ 10952778
......@@ -60,17 +60,19 @@ typedef struct SSchLevel {
typedef struct SSchTask {
uint64_t taskId; // task id
SSchLevel *level; // level
SSubplan *plan; // subplan
char *msg; // operator tree
int32_t msgLen; // msg length
int8_t status; // task status
SEpAddr execAddr; // task actual executed node address
SQueryProfileSummary summary; // task execution summary
int32_t childReady; // child task ready number
SArray *children; // the datasource tasks,from which to fetch the result, element is SQueryTask*
SArray *parents; // the data destination tasks, get data from current task, element is SQueryTask*
uint64_t taskId; // task id
SSchLevel *level; // level
SSubplan *plan; // subplan
char *msg; // operator tree
int32_t msgLen; // msg length
int8_t status; // task status
SQueryNodeAddr execAddr; // task actual executed node address
int8_t condidateIdx; // current try condidation index
SArray *condidateAddrs; // condidate node addresses, element is SQueryNodeAddr
SQueryProfileSummary summary; // task execution summary
int32_t childReady; // child task ready number
SArray *children; // the datasource tasks,from which to fetch the result, element is SQueryTask*
SArray *parents; // the data destination tasks, get data from current task, element is SQueryTask*
} SSchTask;
typedef struct SSchJobAttr {
......
source/libs/scheduler/src/scheduler.c
浏览文件 @ 10952778
......@@ -28,7 +28,7 @@ int32_t schBuildTaskRalation(SSchJob *job, SHashObj *planToTask) {
for (int32_t m = 0; m < level->taskNum; ++m) {
SSchTask *task = taosArrayGet(level->subTasks, m);
SSubplan *plan = task->plan;
int32_t childNum = plan->pChildern ? (int32_t)taosArrayGetSize(plan->pChildern) : 0;
int32_t childNum = plan->pChildren ? (int32_t)taosArrayGetSize(plan->pChildren) : 0;
int32_t parentNum = plan->pParents ? (int32_t)taosArrayGetSize(plan->pParents) : 0;
if (childNum > 0) {
......@@ -40,7 +40,7 @@ int32_t schBuildTaskRalation(SSchJob *job, SHashObj *planToTask) {
}
for (int32_t n = 0; n < childNum; ++n) {
SSubplan **child = taosArrayGet(plan->pChildern, n);
SSubplan **child = taosArrayGet(plan->pChildren, n);
SSchTask **childTask = taosHashGet(planToTask, child, POINTER_BYTES);
if (NULL == childTask || NULL == *childTask) {
qError("subplan relationship error, level:%d, taskIdx:%d, childIdx:%d", i, m, n);
......@@ -122,6 +122,7 @@ int32_t schValidateAndBuildJob(SQueryDag *dag, SSchJob *job) {
SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
//??
job->attr.needFetch = true;
job->levelNum = levelNum;
......@@ -215,28 +216,49 @@ _return:
SCH_RET(code);
}
int32_t schSetTaskExecEpSet(SSchJob *job, SEpSet *epSet) {
if (epSet->numOfEps >= SCH_MAX_CONDIDATE_EP_NUM) {
int32_t schSetTaskCondidateAddrs(SSchJob *job, SSchTask *task) {
if (task->condidateAddrs) {
return TSDB_CODE_SUCCESS;
}
task->condidateIdx = 0;
task->condidateAddrs = taosArrayInit(SCH_MAX_CONDIDATE_EP_NUM, sizeof(SQueryNodeAddr));
if (NULL == task->condidateAddrs) {
qError("taosArrayInit failed");
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
if (task->plan->execNode.numOfEps > 0) {
if (NULL == taosArrayPush(task->condidateAddrs, &task->plan->execNode)) {
qError("taosArrayPush failed");
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
return TSDB_CODE_SUCCESS;
}
int32_t addNum = 0;
int32_t nodeNum = taosArrayGetSize(job->nodeList);
for (int32_t i = 0; i < nodeNum && epSet->numOfEps < tListLen(epSet->port); ++i) {
SEpAddr *addr = taosArrayGet(job->nodeList, i);
for (int32_t i = 0; i < nodeNum && addNum < SCH_MAX_CONDIDATE_EP_NUM; ++i) {
SQueryNodeAddr *naddr = taosArrayGet(job->nodeList, i);
strncpy(epSet->fqdn[epSet->numOfEps], addr->fqdn, sizeof(addr->fqdn));
epSet->port[epSet->numOfEps] = addr->port;
if (NULL == taosArrayPush(task->condidateAddrs, &task->plan->execNode)) {
qError("taosArrayPush failed");
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
++epSet->numOfEps;
++addNum;
}
for (int32_t i = 0; i < job->dataSrcEps.numOfEps && epSet->numOfEps < tListLen(epSet->port); ++i) {
/*
for (int32_t i = 0; i < job->dataSrcEps.numOfEps && addNum < SCH_MAX_CONDIDATE_EP_NUM; ++i) {
strncpy(epSet->fqdn[epSet->numOfEps], job->dataSrcEps.fqdn[i], sizeof(job->dataSrcEps.fqdn[i]));
epSet->port[epSet->numOfEps] = job->dataSrcEps.port[i];
++epSet->numOfEps;
}
*/
return TSDB_CODE_SUCCESS;
}
......@@ -383,8 +405,8 @@ int32_t schProcessOnTaskSuccess(SSchJob *job, SSchTask *task) {
return TSDB_CODE_SUCCESS;
}
} else {
strncpy(job->resEp.fqdn, task->execAddr.fqdn, sizeof(job->resEp.fqdn));
job->resEp.port = task->execAddr.port;
strncpy(job->resEp.fqdn, task->execAddr.epAddr[task->execAddr.inUse].fqdn, sizeof(job->resEp.fqdn));
job->resEp.port = task->execAddr.epAddr[task->execAddr.inUse].port;
}
job->fetchTask = task;
......@@ -394,12 +416,14 @@ int32_t schProcessOnTaskSuccess(SSchJob *job, SSchTask *task) {
return TSDB_CODE_SUCCESS;
}
/*
if (SCH_IS_DATA_SRC_TASK(task) && job->dataSrcEps.numOfEps < SCH_MAX_CONDIDATE_EP_NUM) {
strncpy(job->dataSrcEps.fqdn[job->dataSrcEps.numOfEps], task->execAddr.fqdn, sizeof(task->execAddr.fqdn));
job->dataSrcEps.port[job->dataSrcEps.numOfEps] = task->execAddr.port;
++job->dataSrcEps.numOfEps;
}
*/
for (int32_t i = 0; i < parentNum; ++i) {
SSchTask *par = *(SSchTask **)taosArrayGet(task->parents, i);
......@@ -455,15 +479,27 @@ int32_t schProcessOnTaskFailure(SSchJob *job, SSchTask *task, int32_t errCode) {
int32_t schProcessRspMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *msg, int32_t msgSize, int32_t rspCode) {
int32_t code = 0;
switch (msgType) {
case TDMT_VND_CREATE_TABLE_RSP: {
if (rspCode != TSDB_CODE_SUCCESS) {
SCH_ERR_JRET(schProcessOnTaskFailure(job, task, rspCode));
} else {
// job->resNumOfRows += rsp->affectedRows;
code = schProcessOnTaskSuccess(job, task);
if (code) {
goto _task_error;
}
}
}
case TDMT_VND_SUBMIT_RSP: {
SShellSubmitRspMsg *rsp = (SShellSubmitRspMsg *)msg;
if (rsp->code != TSDB_CODE_SUCCESS) {
SCH_ERR_JRET(schProcessOnTaskFailure(job, task, rsp->code));
if (rspCode != TSDB_CODE_SUCCESS) {
SCH_ERR_JRET(schProcessOnTaskFailure(job, task, rspCode));
} else {
SShellSubmitRspMsg *rsp = (SShellSubmitRspMsg *)msg;
job->resNumOfRows += rsp->affectedRows;
code = schProcessOnTaskSuccess(job, task);
if (code) {
goto _task_error;
......@@ -547,22 +583,29 @@ int32_t schHandleCallback(void* param, const SDataBuf* pMsg, int32_t msgType, in
_return:
tfree(param);
SCH_RET(code);
}
int32_t schHandleSubmitCallback(void* param, const SDataBuf* pMsg, int32_t code) {
return schHandleCallback(param, pMsg, TDMT_VND_SUBMIT_RSP, code);
}
int32_t schHandleCreateTableCallback(void* param, const SDataBuf* pMsg, int32_t code) {
return schHandleCallback(param, pMsg, TDMT_VND_CREATE_TABLE_RSP, code);
}
int32_t schHandleQueryCallback(void* param, const SDataBuf* pMsg, int32_t code) {
return schHandleCallback(param, pMsg, TDMT_VND_QUERY_RSP, code);
}
int32_t schHandleFetchCallback(void* param, const SDataBuf* pMsg, int32_t code) {
return schHandleCallback(param, pMsg, TDMT_VND_FETCH_RSP, code);
}
int32_t schHandleReadyCallback(void* param, const SDataBuf* pMsg, int32_t code) {
return schHandleCallback(param, pMsg, TDMT_VND_RES_READY_RSP, code);
}
int32_t schHandleDropCallback(void* param, const SDataBuf* pMsg, int32_t code) {
SSchCallbackParam *pParam = (SSchCallbackParam *)param;
qDebug("drop task rsp received, queryId:%"PRIx64 ",taksId:%"PRIx64 ",code:%d", pParam->queryId, pParam->taskId, code);
......@@ -570,6 +613,9 @@ int32_t schHandleDropCallback(void* param, const SDataBuf* pMsg, int32_t code) {
int32_t schGetCallbackFp(int32_t msgType, __async_send_cb_fn_t *fp) {
switch (msgType) {
case TDMT_VND_CREATE_TABLE:
*fp = schHandleCreateTableCallback;
break;
case TDMT_VND_SUBMIT:
*fp = schHandleSubmitCallback;
break;
......@@ -633,6 +679,16 @@ _return:
SCH_RET(code);
}
void schConvertAddrToEpSet(SQueryNodeAddr *addr, SEpSet *epSet) {
epSet->inUse = addr->inUse;
epSet->numOfEps = addr->numOfEps;
for (int8_t i = 0; i < epSet->numOfEps; ++i) {
strncpy(epSet->fqdn[i], addr->epAddr[i].fqdn, sizeof(addr->epAddr[i].fqdn));
epSet->port[i] = addr->epAddr[i].port;
}
}
int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
uint32_t msgSize = 0;
......@@ -640,6 +696,7 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
int32_t code = 0;
switch (msgType) {
case TDMT_VND_CREATE_TABLE:
case TDMT_VND_SUBMIT: {
if (NULL == task->msg || task->msgLen <= 0) {
qError("submit msg is NULL");
......@@ -665,6 +722,7 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
SSubQueryMsg *pMsg = msg;
pMsg->header.vgId = htonl(task->plan->execNode.nodeId);
pMsg->sId = htobe64(schMgmt.sId);
pMsg->queryId = htobe64(job->queryId);
pMsg->taskId = htobe64(task->taskId);
......@@ -681,6 +739,8 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
}
SResReadyMsg *pMsg = msg;
pMsg->header.vgId = htonl(task->plan->execNode.nodeId);
pMsg->sId = htobe64(schMgmt.sId);
pMsg->queryId = htobe64(job->queryId);
pMsg->taskId = htobe64(task->taskId);
......@@ -698,6 +758,8 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
}
SResFetchMsg *pMsg = msg;
pMsg->header.vgId = htonl(task->plan->execNode.nodeId);
pMsg->sId = htobe64(schMgmt.sId);
pMsg->queryId = htobe64(job->queryId);
pMsg->taskId = htobe64(task->taskId);
......@@ -712,6 +774,8 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
}
STaskDropMsg *pMsg = msg;
pMsg->header.vgId = htonl(task->plan->execNode.nodeId);
pMsg->sId = htobe64(schMgmt.sId);
pMsg->queryId = htobe64(job->queryId);
pMsg->taskId = htobe64(task->taskId);
......@@ -723,7 +787,12 @@ int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, int32_t msgType) {
break;
}
SCH_ERR_JRET(schAsyncSendMsg(job->transport, &task->plan->execEpSet, job->queryId, task->taskId, msgType, msg, msgSize));
SEpSet epSet;
SQueryNodeAddr *addr = taosArrayGet(task->condidateAddrs, task->condidateIdx);
schConvertAddrToEpSet(addr, &epSet);
SCH_ERR_JRET(schAsyncSendMsg(job->transport, &epSet, job->queryId, task->taskId, msgType, msg, msgSize));
return TSDB_CODE_SUCCESS;
......@@ -737,28 +806,22 @@ _return:
int32_t schLaunchTask(SSchJob *job, SSchTask *task) {
SSubplan *plan = task->plan;
SCH_ERR_RET(qSubPlanToString(plan, &task->msg, &task->msgLen));
if (plan->execEpSet.numOfEps <= 0) {
SCH_ERR_RET(schSetTaskExecEpSet(job, &plan->execEpSet));
}
SCH_ERR_RET(schSetTaskCondidateAddrs(job, task));
if (plan->execEpSet.numOfEps <= 0) {
SCH_TASK_ERR_LOG("invalid execEpSet num:%d", plan->execEpSet.numOfEps);
if (NULL == task->condidateAddrs || taosArrayGetSize(task->condidateAddrs) <= 0) {
SCH_TASK_ERR_LOG("no valid condidate node for task:%"PRIx64, task->taskId);
SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
}
int32_t msgType = (plan->type == QUERY_TYPE_MODIFY) ? TDMT_VND_SUBMIT : TDMT_VND_QUERY;
// int32_t msgType = (plan->type == QUERY_TYPE_MODIFY)? TDMT_VND_SUBMIT : TDMT_VND_QUERY;
SCH_ERR_RET(schBuildAndSendMsg(job, task, msgType));
SCH_ERR_RET(schBuildAndSendMsg(job, task, plan->msgType));
SCH_ERR_RET(schPushTaskToExecList(job, task));
task->status = JOB_TASK_STATUS_EXECUTING;
return TSDB_CODE_SUCCESS;
}
int32_t schLaunchJob(SSchJob *job) {
SSchLevel *level = taosArrayGet(job->levels, job->levelIdx);
for (int32_t i = 0; i < level->taskNum; ++i) {
......
source/libs/scheduler/test/schedulerTests.cpp
浏览文件 @ 10952778
......@@ -54,11 +54,13 @@ void schtBuildQueryDag(SQueryDag *dag) {
scanPlan.id.templateId = 0x0000000000000002;
scanPlan.id.subplanId = 0x0000000000000003;
scanPlan.type = QUERY_TYPE_SCAN;
scanPlan.execNode.numOfEps = 1;
scanPlan.execNode.nodeId = 1;
scanPlan.execNode.inUse = 0;
scanPlan.execNode.epAddr[0].port = 6030;
strcpy(scanPlan.execNode.epAddr[0].fqdn, "ep0");
scanPlan.pChildren = NULL;
scanPlan.level = 1;
scanPlan.execEpSet.numOfEps = 1;
scanPlan.execEpSet.port[0] = 6030;
strcpy(scanPlan.execEpSet.fqdn[0], "ep0");
scanPlan.pChildern = NULL;
scanPlan.pParents = taosArrayInit(1, POINTER_BYTES);
scanPlan.pNode = (SPhyNode*)calloc(1, sizeof(SPhyNode));
......@@ -67,15 +69,15 @@ void schtBuildQueryDag(SQueryDag *dag) {
mergePlan.id.subplanId = 0x5555555555;
mergePlan.type = QUERY_TYPE_MERGE;
mergePlan.level = 0;
mergePlan.execEpSet.numOfEps = 0;
mergePlan.pChildern = taosArrayInit(1, POINTER_BYTES);
mergePlan.execNode.numOfEps = 0;
mergePlan.pChildren = taosArrayInit(1, POINTER_BYTES);
mergePlan.pParents = NULL;
mergePlan.pNode = (SPhyNode*)calloc(1, sizeof(SPhyNode));
SSubplan *mergePointer = (SSubplan *)taosArrayPush(merge, &mergePlan);
SSubplan *scanPointer = (SSubplan *)taosArrayPush(scan, &scanPlan);
taosArrayPush(mergePointer->pChildern, &scanPointer);
taosArrayPush(mergePointer->pChildren, &scanPointer);
taosArrayPush(scanPointer->pParents, &mergePointer);
taosArrayPush(dag->pSubplans, &merge);
......@@ -97,10 +99,12 @@ void schtBuildInsertDag(SQueryDag *dag) {
insertPlan[0].id.subplanId = 0x0000000000000004;
insertPlan[0].type = QUERY_TYPE_MODIFY;
insertPlan[0].level = 0;
insertPlan[0].execEpSet.numOfEps = 1;
insertPlan[0].execEpSet.port[0] = 6030;
strcpy(insertPlan[0].execEpSet.fqdn[0], "ep0");
insertPlan[0].pChildern = NULL;
insertPlan[0].execNode.numOfEps = 1;
insertPlan[0].execNode.nodeId = 1;
insertPlan[0].execNode.inUse = 0;
insertPlan[0].execNode.epAddr[0].port = 6030;
strcpy(insertPlan[0].execNode.epAddr[0].fqdn, "ep0");
insertPlan[0].pChildren = NULL;
insertPlan[0].pParents = NULL;
insertPlan[0].pNode = NULL;
insertPlan[0].pDataSink = (SDataSink*)calloc(1, sizeof(SDataSink));
......@@ -110,10 +114,12 @@ void schtBuildInsertDag(SQueryDag *dag) {
insertPlan[1].id.subplanId = 0x0000000000000005;
insertPlan[1].type = QUERY_TYPE_MODIFY;
insertPlan[1].level = 0;
insertPlan[1].execEpSet.numOfEps = 1;
insertPlan[1].execEpSet.port[0] = 6030;
strcpy(insertPlan[1].execEpSet.fqdn[0], "ep1");
insertPlan[1].pChildern = NULL;
insertPlan[1].execNode.numOfEps = 1;
insertPlan[1].execNode.nodeId = 1;
insertPlan[1].execNode.inUse = 1;
insertPlan[1].execNode.epAddr[0].port = 6030;
strcpy(insertPlan[1].execNode.epAddr[0].fqdn, "ep1");
insertPlan[1].pChildren = NULL;
insertPlan[1].pParents = NULL;
insertPlan[1].pNode = NULL;
insertPlan[1].pDataSink = (SDataSink*)calloc(1, sizeof(SDataSink));
......@@ -132,7 +138,7 @@ int32_t schtPlanToString(const SSubplan *subplan, char** str, int32_t* len) {
return 0;
}
int32_t schtExecNode(SSubplan* subplan, uint64_t templateId, SEpAddr* ep) {
int32_t schtExecNode(SSubplan* subplan, uint64_t templateId, SQueryNodeAddr* ep) {
return 0;
}
......
source/util/CMakeLists.txt
浏览文件 @ 10952778
......@@ -14,4 +14,7 @@ target_link_libraries(
PUBLIC api
)
ADD_SUBDIRECTORY(test)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})
source/util/src/encode.c 0 → 100644
浏览文件 @ 10952778
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "encode.h"
#if __STDC_VERSION__ >= 201112L
static_assert(sizeof(float) == sizeof(uint32_t), "sizeof(float) must equal to sizeof(uint32_t)");
static_assert(sizeof(double) == sizeof(uint64_t), "sizeof(double) must equal to sizeof(uint64_t)");
#endif
void tCoderInit(SCoder* pCoder, td_endian_t endian, uint8_t* data, int32_t size, td_coder_t type) {
if (type == TD_ENCODER) {
if (data == NULL) size = 0;
} else {
ASSERT(data && size > 0);
}
pCoder->type = type;
pCoder->endian = endian;
pCoder->data = data;
pCoder->size = size;
pCoder->pos = 0;
tFreeListInit(&(pCoder->fl));
TD_SLIST_INIT(&(pCoder->stack));
}
void tCoderClear(SCoder* pCoder) {
tFreeListClear(&(pCoder->fl));
struct SCoderNode* pNode;
for (;;) {
pNode = TD_SLIST_HEAD(&(pCoder->stack));
if (pNode == NULL) break;
TD_SLIST_POP(&(pCoder->stack));
free(pNode);
}
}
int tStartEncode(SCoder* pCoder) {
struct SCoderNode* pNode;
ASSERT(pCoder->type == TD_ENCODER);
if (pCoder->data) {
if (pCoder->size - pCoder->pos < sizeof(int32_t)) return -1;
pNode = malloc(sizeof(*pNode));
if (pNode == NULL) return -1;
pNode->data = pCoder->data;
pNode->pos = pCoder->pos;
pNode->size = pCoder->size;
pCoder->data = pNode->data + pNode->pos + sizeof(int32_t);
pCoder->pos = 0;
pCoder->size = pNode->size - pNode->pos - sizeof(int32_t);
TD_SLIST_PUSH(&(pCoder->stack), pNode);
} else {
pCoder->pos += sizeof(int32_t);
}
return 0;
}
void tEndEncode(SCoder* pCoder) {
struct SCoderNode* pNode;
int32_t len;
ASSERT(pCoder->type == TD_ENCODER);
if (pCoder->data) {
pNode = TD_SLIST_HEAD(&(pCoder->stack));
ASSERT(pNode);
TD_SLIST_POP(&(pCoder->stack));
len = pCoder->pos;
pCoder->data = pNode->data;
pCoder->size = pNode->size;
pCoder->pos = pNode->pos;
if (TD_RT_ENDIAN() == pCoder->endian) {
tPut(int32_t, pCoder->data + pCoder->pos, len);
} else {
tRPut32(pCoder->data + pCoder->pos, len);
}
TD_CODER_MOVE_POS(pCoder, len + sizeof(int32_t));
free(pNode);
}
}
int tStartDecode(SCoder* pCoder) {
int32_t len;
struct SCoderNode* pNode;
ASSERT(pCoder->type == TD_DECODER);
if (tDecodeI32(pCoder, &len) < 0) return -1;
pNode = malloc(sizeof(*pNode));
if (pNode == NULL) return -1;
pNode->data = pCoder->data;
pNode->pos = pCoder->pos;
pNode->size = pCoder->size;
pCoder->data = pNode->data + pNode->pos;
pCoder->size = len;
pCoder->pos = 0;
TD_SLIST_PUSH(&(pCoder->stack), pNode);
return 0;
}
void tEndDecode(SCoder* pCoder) {
struct SCoderNode* pNode;
ASSERT(pCoder->type == TD_DECODER);
ASSERT(tDecodeIsEnd(pCoder));
pNode = TD_SLIST_HEAD(&(pCoder->stack));
ASSERT(pNode);
TD_SLIST_POP(&(pCoder->stack));
pCoder->data = pNode->data;
pCoder->size = pNode->size;
pCoder->pos = pCoder->pos + pNode->pos;
free(pNode);
}
source/util/test/CMakeLists.txt
浏览文件 @ 10952778
......@@ -41,4 +41,8 @@ target_sources(freelistTest
)
target_link_libraries(freelistTest os util gtest gtest_main)
# encodeTest
add_executable(encodeTest "encodeTest.cpp")
target_link_libraries(encodeTest os util gtest gtest_main)
source/util/test/encodeTest.cpp 0 → 100644
浏览文件 @ 10952778
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或