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

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

This commit is contained in:
Shengliang Guan 2021-12-22 18:44:34 +08:00
parent 5b08678f30 cd8e91c96c
commit d96963a2ec
46 changed files with 2007 additions and 1588 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
include/common/taosmsg.h
View File

@ -51,7 +51,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_QUERY, "mq-query" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_CONNECT, "mq-connect" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_DISCONNECT, "mq-disconnect" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MQ_SET_CUR, "mq-set-cur" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_RSP_READY, "rsp-ready" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_RES_READY, "res-ready" )
// message from client to mnode
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CONNECT, "connect" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_ACCT, "create-acct" )
@ -1081,6 +1081,7 @@ typedef struct {
} SUpdateTagValRsp;
typedef struct SSchedulerQueryMsg {
uint64_t schedulerId;
uint64_t queryId;
uint64_t taskId;
uint32_t contentLen;
@ -1088,15 +1089,39 @@ typedef struct SSchedulerQueryMsg {
} SSchedulerQueryMsg;
typedef struct SSchedulerReadyMsg {
uint64_t schedulerId;
uint64_t queryId;
uint64_t taskId;
} SSchedulerReadyMsg;
typedef struct SSchedulerFetchMsg {
uint64_t schedulerId;
uint64_t queryId;
uint64_t taskId;
} SSchedulerFetchMsg;
typedef struct SSchedulerStatusMsg {
uint64_t schedulerId;
} SSchedulerStatusMsg;
typedef struct STaskStatus {
uint64_t queryId;
uint64_t taskId;
int8_t status;
} STaskStatus;
typedef struct SSchedulerStatusRsp {
uint32_t num;
STaskStatus status[];
} SSchedulerStatusRsp;
typedef struct SSchedulerCancelMsg {
uint64_t schedulerId;
uint64_t queryId;
uint64_t taskId;
} SSchedulerCancelMsg;
#pragma pack(pop)

44
include/dnode/qnode/qnode.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "trpc.h"
typedef struct {
uint64_t numOfStartTask;
@ -32,48 +32,6 @@ typedef struct {
uint64_t numOfErrors;
} SQnodeStat;
/* start Task msg */
typedef struct {
uint32_t schedulerIp;
uint16_t schedulerPort;
int64_t taskId;
int64_t queryId;
uint32_t srcIp;
uint16_t srcPort;
} SQnodeStartTaskMsg;
/* stop Task msg */
typedef struct {
int64_t taskId;
} SQnodeStopTaskMsg;
/* start/stop Task msg response */
typedef struct {
int64_t taskId;
int32_t code;
} SQnodeTaskRespMsg;
/* Task status msg */
typedef struct {
int64_t taskId;
int32_t status;
int64_t queryId;
} SQnodeTaskStatusMsg;
/* Qnode/Scheduler heartbeat msg */
typedef struct {
int32_t status;
int32_t load;
} SQnodeHeartbeatMsg;
/* Qnode sent/received msg */
typedef struct {
int8_t msgType;
int32_t msgLen;
char msg[];
} SQnodeMsg;
/**
* Start one Qnode in Dnode.

9
include/libs/index/index.h
View File

@ -76,8 +76,13 @@ void indexOptsDestroy(SIndexOpts *opts);
* @param:
*/
SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn operType, uint8_t colType, const char *colName,
int32_t nColName, const char *colVal, int32_t nColVal);
SIndexTerm* indexTermCreate(int64_t suid,
SIndexOperOnColumn operType,
uint8_t colType,
const char* colName,
int32_t nColName,
const char* colVal,
int32_t nColVal);
void indexTermDestroy(SIndexTerm* p);
#ifdef __cplusplus

1
include/libs/parser/parser.h
View File

@ -136,6 +136,7 @@ typedef struct SParseContext {
const char* pDbname;
void *pRpc;
const char* pClusterId;
struct SCatalog* pCatalog;
const SEpSet* pEpSet;
int64_t id; // query id, generated by uuid generator
int8_t schemaAttached; // denote if submit block is built with table schema or not

9
include/libs/qcom/query.h
View File

@ -24,6 +24,15 @@ extern "C" {
#include "thash.h"
#include "tlog.h"
enum {
JOB_TASK_STATUS_NOT_START = 1,
JOB_TASK_STATUS_EXECUTING,
JOB_TASK_STATUS_SUCCEED,
JOB_TASK_STATUS_FAILED,
JOB_TASK_STATUS_CANCELLING,
JOB_TASK_STATUS_CANCELLED
};
typedef struct STableComInfo {
uint8_t numOfTags; // the number of tags in schema
uint8_t precision; // the number of precision

62
include/libs/qworker/qworker.h Normal file
View File

@ -0,0 +1,62 @@
/*
* 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_QWORKER_H_
#define _TD_QWORKER_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "trpc.h"
typedef struct SQWorkerCfg {
uint32_t maxSchedulerNum;
uint32_t maxResCacheNum;
uint32_t maxSchTaskNum;
} SQWorkerCfg;
typedef struct {
uint64_t numOfStartTask;
uint64_t numOfStopTask;
uint64_t numOfRecvedFetch;
uint64_t numOfSentHb;
uint64_t numOfSentFetch;
uint64_t numOfTaskInQueue;
uint64_t numOfFetchInQueue;
uint64_t numOfErrors;
} SQWorkerStat;
int32_t qWorkerInit(SQWorkerCfg *cfg, void **qWorkerMgmt);
int32_t qWorkerProcessQueryMsg(void *qWorkerMgmt, SSchedulerQueryMsg *msg, SRpcMsg *rsp);
int32_t qWorkerProcessReadyMsg(void *qWorkerMgmt, SSchedulerReadyMsg *msg, SRpcMsg *rsp);
int32_t qWorkerProcessStatusMsg(void *qWorkerMgmt, SSchedulerStatusMsg *msg, SRpcMsg *rsp);
int32_t qWorkerProcessFetchMsg(void *qWorkerMgmt, SSchedulerFetchMsg *msg, SRpcMsg *rsp);
int32_t qWorkerProcessCancelMsg(void *qWorkerMgmt, SSchedulerCancelMsg *msg, SRpcMsg *rsp);
void qWorkerDestroy(void **qWorkerMgmt);
#ifdef __cplusplus
}
#endif
#endif /*_TD_QWORKER_H_*/

1
include/libs/scheduler/scheduler.h
View File

@ -25,6 +25,7 @@ extern "C" {
typedef struct SSchedulerCfg {
int32_t clusterType;
int32_t maxJobNum;
} SSchedulerCfg;
typedef struct SQueryProfileSummary {

3
include/util/thash.h
View File

@ -193,11 +193,10 @@ void taosHashCancelIterate(SHashObj *pHashObj, void *p);
/**
* Get the corresponding key information for a given data in hash table
* @param pHashObj
* @param data
* @return
*/
int32_t taosHashGetKey(SHashObj *pHashObj, void *data, void** key, size_t* keyLen);
int32_t taosHashGetKey(void *data, void** key, size_t* keyLen);
#ifdef __cplusplus
}

1
source/libs/CMakeLists.txt
View File

@ -11,3 +11,4 @@ add_subdirectory(executor)
add_subdirectory(planner)
add_subdirectory(function)
add_subdirectory(qcom)
add_subdirectory(qworker)

2
source/libs/executor/src/tfilter.c
View File

@ -1175,7 +1175,7 @@ int32_t filterAddGroupUnitFromNode(SFilterInfo *info, tExprNode* tree, SArray *g
void* key = NULL;
len = 0;
taosHashGetKey((SHashObj *)data, p, &key, &len);
taosHashGetKey(p, &key, &len);
void *fdata = NULL;
if (IS_VAR_DATA_TYPE(type)) {

3
source/libs/index/inc/index_cache.h
View File

@ -45,8 +45,7 @@ void indexCacheDestroy(void *cache);
int indexCachePut(void* cache, SIndexTerm* term, int16_t colId, int32_t version, uint64_t uid);
// int indexCacheGet(void *cache, uint64_t *rst);
int indexCacheSearch(
void *cache, SIndexTermQuery *query, int16_t colId, int32_t version, SArray *result, STermValueType *s);
int indexCacheSearch(void* cache, SIndexTermQuery* query, int16_t colId, int32_t version, SArray* result, STermValueType* s);
#ifdef __cplusplus
}

6
source/libs/index/inc/index_fst.h
View File

@ -142,8 +142,7 @@ uint64_t fstStateInputLen(FstState *state);
// end_addr
uint64_t fstStateEndAddrForOneTransNext(FstState* state, FstSlice* data);
uint64_t fstStateEndAddrForOneTrans(FstState* state, FstSlice* data, PackSizes sizes);
uint64_t fstStateEndAddrForAnyTrans(
FstState *state, uint64_t version, FstSlice *date, PackSizes sizes, uint64_t nTrans);
uint64_t fstStateEndAddrForAnyTrans(FstState* state, uint64_t version, FstSlice* date, PackSizes sizes, uint64_t nTrans);
// input
uint8_t fstStateInput(FstState* state, FstNode* node);
uint8_t fstStateInputForAnyTrans(FstState* state, FstNode* node, uint64_t i);
@ -311,8 +310,7 @@ StreamWithStateResult *swsResultCreate(FstSlice *data, FstOutput fOut, void *sta
void swsResultDestroy(StreamWithStateResult* result);
typedef void* (*StreamCallback)(void*);
StreamWithState *streamWithStateCreate(
Fst *fst, AutomationCtx *automation, FstBoundWithData *min, FstBoundWithData *max);
StreamWithState* streamWithStateCreate(Fst* fst, AutomationCtx* automation, FstBoundWithData* min, FstBoundWithData* max);
void streamWithStateDestroy(StreamWithState* sws);

2
source/libs/index/inc/index_fst_counting_writer.h
View File

@ -31,6 +31,7 @@ typedef struct WriterCtx {
int (*write)(struct WriterCtx* ctx, uint8_t* buf, int len);
int (*read)(struct WriterCtx* ctx, uint8_t* buf, int len);
int (*flush)(struct WriterCtx* ctx);
int (*readFrom)(struct WriterCtx* ctx, uint8_t* buf, int len, int32_t offset);
WriterType type;
union {
struct {
@ -48,6 +49,7 @@ typedef struct WriterCtx {
static int writeCtxDoWrite(WriterCtx* ctx, uint8_t* buf, int len);
static int writeCtxDoRead(WriterCtx* ctx, uint8_t* buf, int len);
static int writeCtxDoReadFrom(WriterCtx* ctx, uint8_t* buf, int len, int32_t offset);
static int writeCtxDoFlush(WriterCtx* ctx);
WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int32_t capacity);

35
source/libs/index/inc/index_tfile.h
View File

@ -26,25 +26,29 @@
extern "C" {
#endif
// tfile header
// |<---suid--->|<---version--->|<--colLen-->|<-colName->|<---type-->|
// |<-uint64_t->|<---int32_t--->|<--int32_t->|<-colLen-->|<-uint8_t->|
// tfile header content
// |<---suid--->|<---version--->|<-------colName------>|<---type-->|<--fstOffset->|
// |<-uint64_t->|<---int32_t--->|<--TSDB_COL_NAME_LEN-->|<-uint8_t->|<---int32_t-->|
typedef struct TFileReadHeader {
#pragma pack(push, 1)
typedef struct TFileHeader {
uint64_t suid;
int32_t version;
char colName[128]; //
char colName[TSDB_COL_NAME_LEN]; //
uint8_t colType;
} TFileReadHeader;
int32_t fstOffset;
} TFileHeader;
#pragma pack(pop)
#define TFILE_HEADER_SIZE (sizeof(TFILE_HEADER_SIZE) + sizeof(uint32_t));
#define TFILE_HADER_PRE_SIZE (sizeof(uint64_t) + sizeof(int32_t) + sizeof(int32_t))
#define TFILE_HEADER_SIZE (sizeof(TFileHeader))
#define TFILE_HEADER_NO_FST (TFILE_HEADER_SIZE - sizeof(int32_t))
//#define TFILE_HADER_PRE_SIZE (sizeof(uint64_t) + sizeof(int32_t) + sizeof(int32_t))
typedef struct TFileCacheKey {
uint64_t suid;
uint8_t colType;
int32_t version;
const char *colName;
char* colName;
int32_t nColName;
} TFileCacheKey;
@ -59,13 +63,15 @@ typedef struct TFileCache {
typedef struct TFileWriter {
FstBuilder* fb;
WriterCtx* ctx;
TFileHeader header;
uint32_t offset;
} TFileWriter;
typedef struct TFileReader {
T_REF_DECLARE()
Fst* fst;
WriterCtx* ctx;
TFileReadHeader header;
TFileHeader header;
} TFileReader;
typedef struct IndexTFile {
@ -94,11 +100,14 @@ void tfileCacheDestroy(TFileCache *tcache);
TFileReader* tfileCacheGet(TFileCache* tcache, TFileCacheKey* key);
void tfileCachePut(TFileCache* tcache, TFileCacheKey* key, TFileReader* reader);
TFileReader *tfileReaderCreate();
void TFileReaderDestroy(TFileReader *reader);
TFileReader* tfileReaderCreate(WriterCtx* ctx);
void tfileReaderDestroy(TFileReader* reader);
int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* result);
TFileWriter *tfileWriterCreate(const char *suid, const char *colName);
TFileWriter* tfileWriterCreate(WriterCtx* ctx, TFileHeader* header);
void tfileWriterDestroy(TFileWriter* tw);
int tfileWriterPut(TFileWriter* tw, void* data);
int tfileWriterFinish(TFileWriter* tw);
//
IndexTFile* indexTFileCreate(const char* path);

19
source/libs/index/src/index.c
View File

@ -58,6 +58,7 @@ int indexOpen(SIndexOpts *opts, const char *path, SIndex **index) {
sIdx->index = index;
#endif
#ifdef USE_INVERTED_INDEX
sIdx->cache = (void*)indexCacheCreate();
sIdx->tindex = NULL;
sIdx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
@ -67,6 +68,10 @@ int indexOpen(SIndexOpts *opts, const char *path, SIndex **index) {
*index = sIdx;
return 0;
#endif
*index = NULL;
return -1;
}
void indexClose(SIndex* sIdx) {
@ -237,8 +242,13 @@ int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm *term, EInde
return 0;
}
SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn oper, uint8_t colType, const char *colName,
int32_t nColName, const char *colVal, int32_t nColVal) {
SIndexTerm* indexTermCreate(int64_t suid,
SIndexOperOnColumn oper,
uint8_t colType,
const char* colName,
int32_t nColName,
const char* colVal,
int32_t nColVal) {
SIndexTerm* t = (SIndexTerm*)calloc(1, (sizeof(SIndexTerm)));
if (t == NULL) {
return NULL;
@ -263,7 +273,9 @@ void indexTermDestroy(SIndexTerm *p) {
free(p);
}
SIndexMultiTerm *indexMultiTermCreate() { return taosArrayInit(4, sizeof(SIndexTerm *)); }
SIndexMultiTerm* indexMultiTermCreate() {
return taosArrayInit(4, sizeof(SIndexTerm*));
}
int indexMultiTermAdd(SIndexMultiTerm* terms, SIndexTerm* term) {
taosArrayPush(terms, &term);
@ -336,6 +348,7 @@ static int indexMergeFinalResults(SArray *interResults, EIndexOperatorType oType
SArray* first = taosArrayGetP(interResults, 0);
taosArraySort(first, uidCompare);
taosArrayRemoveDuplicate(first, uidCompare, NULL);
if (oType == MUST) {
// just one column index, enhance later
taosArrayAddAll(fResults, first);

44
source/libs/index/src/index_cache.c
View File

@ -21,10 +21,11 @@
// 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) + \
sizeof(p->operType))
(sizeof(int32_t) + sizeof(uint16_t) + sizeof(p->colType) + sizeof(p->nColVal) + p->nColVal + sizeof(uint64_t) + sizeof(p->operType))
static char * getIndexKey(const void *pData) { return NULL; }
static char* getIndexKey(const void* pData) {
return NULL;
}
static int32_t compareKey(const void* l, const void* r) {
char* lp = (char*)l;
char* rp = (char*)r;
@ -40,9 +41,7 @@ static int32_t compareKey(const void *l, const void *r) {
int16_t lf, rf; // field id
memcpy(&lf, lp, sizeof(lf));
memcpy(&rf, rp, sizeof(rf));
if (lf != rf) {
return lf < rf ? -1 : 1;
}
if (lf != rf) { return lf < rf ? -1 : 1; }
lp += sizeof(lf);
rp += sizeof(rf);
@ -89,9 +88,8 @@ static int32_t compareKey(const void *l, const void *r) {
int32_t lv, rv;
memcpy(&lv, lp, sizeof(lv));
memcpy(&rv, rp, sizeof(rv));
if (lv != rv) {
return lv > rv ? -1 : 1;
}
if (lv != rv) { return lv > rv ? -1 : 1; }
lp += sizeof(lv);
rp += sizeof(rv);
// not care item type
@ -100,28 +98,29 @@ static int32_t compareKey(const void *l, const void *r) {
}
IndexCache* indexCacheCreate() {
IndexCache* cache = calloc(1, sizeof(IndexCache));
cache->skiplist = tSkipListCreate(
MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, MAX_INDEX_KEY_LEN, compareKey, SL_ALLOW_DUP_KEY, getIndexKey);
if (cache == NULL) {
indexError("failed to create index cache");
return NULL;
}
cache->skiplist =
tSkipListCreate(MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, MAX_INDEX_KEY_LEN, compareKey, SL_ALLOW_DUP_KEY, getIndexKey);
return cache;
}
void indexCacheDestroy(void* cache) {
IndexCache* pCache = cache;
if (pCache == NULL) {
return;
}
if (pCache == NULL) { return; }
tSkipListDestroy(pCache->skiplist);
free(pCache);
}
int indexCachePut(void* cache, SIndexTerm* term, int16_t colId, int32_t version, uint64_t uid) {
if (cache == NULL) {
return -1;
}
if (cache == NULL) { return -1; }
IndexCache* pCache = cache;
// encode data
int32_t total = CACHE_KEY_LEN(term);
char* buf = calloc(1, total);
char* p = buf;
@ -145,11 +144,8 @@ int indexCacheDel(void *cache, int32_t fieldId, const char *fieldValue, int32_t
IndexCache* pCache = cache;
return 0;
}
int indexCacheSearch(
void *cache, SIndexTermQuery *query, int16_t colId, int32_t version, SArray *result, STermValueType *s) {
if (cache == NULL) {
return -1;
}
int indexCacheSearch(void* cache, SIndexTermQuery* query, int16_t colId, int32_t version, SArray* result, STermValueType* s) {
if (cache == NULL) { return -1; }
IndexCache* pCache = cache;
SIndexTerm* term = query->term;
EIndexQueryType qtype = query->qType;
@ -158,9 +154,13 @@ int indexCacheSearch(
char* buf = calloc(1, keyLen);
if (qtype == QUERY_TERM) {
//
} else if (qtype == QUERY_PREFIX) {
//
} else if (qtype == QUERY_SUFFIX) {
//
} else if (qtype == QUERY_REGEX) {
//
}
return 0;

240
source/libs/index/src/index_fst.c
View File

@ -31,9 +31,7 @@ static uint8_t fstPackDetla(FstCountingWriter *wrt, CompiledAddr nodeAddr, Compi
FstUnFinishedNodes* fstUnFinishedNodesCreate() {
FstUnFinishedNodes* nodes = malloc(sizeof(FstUnFinishedNodes));
if (nodes == NULL) {
return NULL;
}
if (nodes == NULL) { return NULL; }
nodes->stack = (SArray*)taosArrayInit(64, sizeof(FstBuilderNodeUnfinished));
fstUnFinishedNodesPushEmpty(nodes, false);
@ -46,9 +44,7 @@ void unFinishedNodeDestroyElem(void *elem) {
b->last = NULL;
}
void fstUnFinishedNodesDestroy(FstUnFinishedNodes* nodes) {
if (nodes == NULL) {
return;
}
if (nodes == NULL) { return; }
taosArrayDestroyEx(nodes->stack, unFinishedNodeDestroyElem);
free(nodes);
@ -91,15 +87,12 @@ void fstUnFinishedNodesSetRootOutput(FstUnFinishedNodes *nodes, Output out) {
// un->node->trans = NULL;
}
void fstUnFinishedNodesTopLastFreeze(FstUnFinishedNodes* nodes, CompiledAddr addr) {
size_t sz = taosArrayGetSize(nodes->stack) - 1;
FstBuilderNodeUnfinished *un = taosArrayGet(nodes->stack, sz);
FstBuilderNodeUnfinished* un = taosArrayGet(nodes->stack, taosArrayGetSize(nodes->stack) - 1);
fstBuilderNodeUnfinishedLastCompiled(un, addr);
}
void fstUnFinishedNodesAddSuffix(FstUnFinishedNodes* nodes, FstSlice bs, Output out) {
FstSlice* s = &bs;
if (fstSliceIsEmpty(s)) {
return;
}
if (fstSliceIsEmpty(s)) { return; }
size_t sz = taosArrayGetSize(nodes->stack) - 1;
FstBuilderNodeUnfinished* un = taosArrayGet(nodes->stack, sz);
assert(un->last == NULL);
@ -179,9 +172,7 @@ uint64_t fstUnFinishedNodesFindCommPrefixAndSetOutput(FstUnFinishedNodes *node,
FstState fstStateCreateFrom(FstSlice* slice, CompiledAddr addr) {
FstState fs = {.state = EmptyFinal, .val = 0};
if (addr == EMPTY_ADDRESS) {
return fs;
}
if (addr == EMPTY_ADDRESS) { return fs; }
uint8_t* data = fstSliceData(slice, NULL);
uint8_t v = data[addr];
@ -197,8 +188,10 @@ FstState fstStateCreateFrom(FstSlice *slice, CompiledAddr addr) {
return fs;
}
static FstState fstStateDict[] = {{.state = OneTransNext, .val = 0b11000000}, {.state = OneTrans, .val = 0b10000000},
{.state = AnyTrans, .val = 0b00000000}, {.state = EmptyFinal, .val = 0b00000000}};
static FstState fstStateDict[] = {{.state = OneTransNext, .val = 0b11000000},
{.state = OneTrans, .val = 0b10000000},
{.state = AnyTrans, .val = 0b00000000},
{.state = EmptyFinal, .val = 0b00000000}};
// debug
static const char* fstStateStr[] = {"ONE_TRANS_NEXT", "ONE_TRANS", "ANY_TRANS", "EMPTY_FINAL"};
@ -236,9 +229,7 @@ void fstStateCompileForOneTrans(FstCountingWriter *w, CompiledAddr addr, FstTran
fstStateSetCommInput(&st, trn->inp);
bool null = false;
uint8_t inp = fstStateCommInput(&st, &null);
if (null == true) {
fstCountingWriterWrite(w, (char *)&trn->inp, sizeof(trn->inp));
}
if (null == true) { fstCountingWriterWrite(w, (char*)&trn->inp, sizeof(trn->inp)); }
fstCountingWriterWrite(w, (char*)(&(st.val)), sizeof(st.val));
return;
}
@ -272,9 +263,7 @@ void fstStateCompileForAnyTrans(FstCountingWriter *w, CompiledAddr addr, FstBuil
fstStateSetStateNtrans(&st, (uint8_t)sz);
if (anyOuts) {
if (FST_BUILDER_NODE_IS_FINAL(node)) {
fstCountingWriterPackUintIn(w, node->finalOutput, oSize);
}
if (FST_BUILDER_NODE_IS_FINAL(node)) { fstCountingWriterPackUintIn(w, node->finalOutput, oSize); }
for (int32_t i = sz - 1; i >= 0; i--) {
FstTransition* t = taosArrayGet(node->trans, i);
fstCountingWriterPackUintIn(w, t->out, oSize);
@ -365,8 +354,7 @@ uint64_t fstStateEndAddrForOneTrans(FstState *s, FstSlice *data, PackSizes sizes
return FST_SLICE_LEN(data) - 1 - fstStateInputLen(s) - 1 // pack size
- FST_GET_TRANSITION_PACK_SIZE(sizes) - FST_GET_OUTPUT_PACK_SIZE(sizes);
}
uint64_t fstStateEndAddrForAnyTrans(
FstState *state, uint64_t version, FstSlice *date, PackSizes sizes, uint64_t nTrans) {
uint64_t fstStateEndAddrForAnyTrans(FstState* state, uint64_t version, FstSlice* date, PackSizes sizes, uint64_t nTrans) {
uint8_t oSizes = FST_GET_OUTPUT_PACK_SIZE(sizes);
uint8_t finalOsize = !fstStateIsFinalState(state) ? 0 : oSizes;
return FST_SLICE_LEN(date) - 1 - fstStateNtransLen(state) - 1 // pack size
@ -415,8 +403,8 @@ CompiledAddr fstStateTransAddrForAnyTrans(FstState *s, FstNode *node, uint64_t i
FstSlice* slice = &node->data;
uint8_t tSizes = FST_GET_TRANSITION_PACK_SIZE(node->sizes);
uint64_t at = node->start - fstStateNtransLen(s) - 1 - fstStateTransIndexSize(s, node->version, node->nTrans) -
node->nTrans - (i * tSizes) - tSizes;
uint64_t at = node->start - fstStateNtransLen(s) - 1 - fstStateTransIndexSize(s, node->version, node->nTrans) - node->nTrans -
(i * tSizes) - tSizes;
uint8_t* data = fstSliceData(slice, NULL);
return unpackDelta(data + at, tSizes, node->end);
}
@ -439,9 +427,7 @@ Output fstStateOutput(FstState *s, FstNode *node) {
assert(s->state == OneTrans);
uint8_t oSizes = FST_GET_OUTPUT_PACK_SIZE(node->sizes);
if (oSizes == 0) {
return 0;
}
if (oSizes == 0) { return 0; }
FstSlice* slice = &node->data;
uint8_t tSizes = FST_GET_TRANSITION_PACK_SIZE(node->sizes);
@ -453,9 +439,7 @@ Output fstStateOutputForAnyTrans(FstState *s, FstNode *node, uint64_t i) {
assert(s->state == AnyTrans);
uint8_t oSizes = FST_GET_OUTPUT_PACK_SIZE(node->sizes);
if (oSizes == 0) {
return 0;
}
if (oSizes == 0) { return 0; }
FstSlice* slice = &node->data;
uint8_t* data = fstSliceData(slice, NULL);
uint64_t at = node->start - fstStateNtransLen(s) - 1 // pack size
@ -468,9 +452,7 @@ Output fstStateOutputForAnyTrans(FstState *s, FstNode *node, uint64_t i) {
void fstStateSetFinalState(FstState* s, bool yes) {
assert(s->state == AnyTrans);
if (yes) {
s->val |= 0b01000000;
}
if (yes) { s->val |= 0b01000000; }
return;
}
bool fstStateIsFinalState(FstState* s) {
@ -480,9 +462,7 @@ bool fstStateIsFinalState(FstState *s) {
void fstStateSetStateNtrans(FstState* s, uint8_t n) {
assert(s->state == AnyTrans);
if (n <= 0b00111111) {
s->val = (s->val & 0b11000000) | n;
}
if (n <= 0b00111111) { s->val = (s->val & 0b11000000) | n; }
return;
}
// state_ntrans
@ -514,9 +494,7 @@ uint64_t fstStateNtransLen(FstState *s) {
uint64_t fstStateNtrans(FstState* s, FstSlice* slice) {
bool null = false;
uint8_t n = fstStateStateNtrans(s, &null);
if (null != true) {
return n;
}
if (null != true) { return n; }
int32_t len;
uint8_t* data = fstSliceData(slice, &len);
n = data[len - 2];
@ -526,9 +504,7 @@ uint64_t fstStateNtrans(FstState *s, FstSlice *slice) {
}
Output fstStateFinalOutput(FstState* s, uint64_t version, FstSlice* slice, PackSizes sizes, uint64_t nTrans) {
uint8_t oSizes = FST_GET_OUTPUT_PACK_SIZE(sizes);
if (oSizes == 0 || !fstStateIsFinalState(s)) {
return 0;
}
if (oSizes == 0 || !fstStateIsFinalState(s)) { return 0; }
uint64_t at = FST_SLICE_LEN(slice) - 1 - fstStateNtransLen(s) - 1 // pack size
- fstStateTotalTransSize(s, version, sizes, nTrans) - (nTrans * oSizes) - oSizes;
@ -545,9 +521,7 @@ uint64_t fstStateFindInput(FstState *s, FstNode *node, uint8_t b, bool *null) {
uint8_t* data = fstSliceData(slice, &dlen);
uint64_t i = data[at + b];
// uint64_t i = slice->data[slice->start + at + b];
if (i >= node->nTrans) {
*null = true;
}
if (i >= node->nTrans) { *null = true; }
return i;
} else {
uint64_t start = node->start - fstStateNtransLen(s) - 1 // pack size
@ -564,9 +538,7 @@ uint64_t fstStateFindInput(FstState *s, FstNode *node, uint8_t b, bool *null) {
return node->nTrans - i - 1; // bug
}
}
if (i == len) {
*null = true;
}
if (i == len) { *null = true; }
fstSliceDestroy(&t);
}
}
@ -575,9 +547,7 @@ uint64_t fstStateFindInput(FstState *s, FstNode *node, uint8_t b, bool *null) {
FstNode* fstNodeCreate(int64_t version, CompiledAddr addr, FstSlice* slice) {
FstNode* n = (FstNode*)malloc(sizeof(FstNode));
if (n == NULL) {
return NULL;
}
if (n == NULL) { return NULL; }
FstState st = fstStateCreateFrom(slice, addr);
@ -625,8 +595,7 @@ FstNode *fstNodeCreate(int64_t version, CompiledAddr addr, FstSlice *slice) {
n->isFinal = fstStateIsFinalState(&st); // s.is_final_state();
n->nTrans = nTrans;
n->sizes = sz;
n->finalOutput =
fstStateFinalOutput(&st, version, &data, sz, nTrans); // s.final_output(version, data, sz, ntrans);
n->finalOutput = fstStateFinalOutput(&st, version, &data, sz, nTrans); // s.final_output(version, data, sz, ntrans);
}
return n;
}
@ -643,9 +612,7 @@ void fstNodeDestroy(FstNode *node) {
}
FstTransitions* fstNodeTransitions(FstNode* node) {
FstTransitions* t = malloc(sizeof(FstTransitions));
if (NULL == t) {
return NULL;
}
if (NULL == t) { return NULL; }
FstRange range = {.start = 0, .end = FST_NODE_LEN(node)};
t->range = range;
t->node = node;
@ -752,9 +719,7 @@ bool fstBuilderNodeCompileTo(FstBuilderNode *b, FstCountingWriter *wrt, Compiled
FstBuilder* fstBuilderCreate(void* w, FstType ty) {
FstBuilder* b = malloc(sizeof(FstBuilder));
if (NULL == b) {
return b;
}
if (NULL == b) { return b; }
b->wrt = fstCountingWriterCreate(w);
b->unfinished = fstUnFinishedNodesCreate();
@ -776,9 +741,7 @@ FstBuilder *fstBuilderCreate(void *w, FstType ty) {
return b;
}
void fstBuilderDestroy(FstBuilder* b) {
if (b == NULL) {
return;
}
if (b == NULL) { return; }
fstCountingWriterDestroy(b->wrt);
fstUnFinishedNodesDestroy(b->unfinished);
@ -794,7 +757,7 @@ bool fstBuilderInsert(FstBuilder *b, FstSlice bs, Output in) {
fstBuilderInsertOutput(b, bs, in);
return true;
}
indexInfo("key must be ordered");
indexInfo("fst write key must be ordered");
return false;
}
@ -879,9 +842,7 @@ CompiledAddr fstBuilderCompile(FstBuilder *b, FstBuilderNode *bn) {
fstBuilderNodeCompileTo(bn, b->wrt, b->lastAddr, startAddr);
b->lastAddr = (CompiledAddr)(FST_WRITER_COUNT(b->wrt) - 1);
if (entry->state == NOTFOUND) {
FST_REGISTRY_CELL_INSERT(entry->cell, b->lastAddr);
}
if (entry->state == NOTFOUND) { FST_REGISTRY_CELL_INSERT(entry->cell, b->lastAddr); }
fstRegistryEntryDestroy(entry);
return b->lastAddr;
@ -914,7 +875,9 @@ void *fstBuilderInsertInner(FstBuilder *b) {
// b->wrt = NULL;
return b->wrt;
}
void fstBuilderFinish(FstBuilder *b) { fstBuilderInsertInner(b); }
void fstBuilderFinish(FstBuilder* b) {
fstBuilderInsertInner(b);
}
FstSlice fstNodeAsSlice(FstNode* node) {
FstSlice* slice = &node->data;
@ -924,21 +887,19 @@ FstSlice fstNodeAsSlice(FstNode *node) {
FstLastTransition* fstLastTransitionCreate(uint8_t inp, Output out) {
FstLastTransition* trn = malloc(sizeof(FstLastTransition));
if (trn == NULL) {
return NULL;
}
if (trn == NULL) { return NULL; }
trn->inp = inp;
trn->out = out;
return trn;
}
void fstLastTransitionDestroy(FstLastTransition *trn) { free(trn); }
void fstLastTransitionDestroy(FstLastTransition* trn) {
free(trn);
}
void fstBuilderNodeUnfinishedLastCompiled(FstBuilderNodeUnfinished* unNode, CompiledAddr addr) {
FstLastTransition* trn = unNode->last;
if (trn == NULL) {
return;
}
if (trn == NULL) { return; }
FstTransition t = {.inp = trn->inp, .out = trn->out, .addr = addr};
taosArrayPush(unNode->node->trans, &t);
fstLastTransitionDestroy(trn);
@ -947,35 +908,27 @@ void fstBuilderNodeUnfinishedLastCompiled(FstBuilderNodeUnfinished *unNode, Comp
}
void fstBuilderNodeUnfinishedAddOutputPrefix(FstBuilderNodeUnfinished* unNode, Output out) {
if (FST_BUILDER_NODE_IS_FINAL(unNode->node)) {
unNode->node->finalOutput += out;
}
if (FST_BUILDER_NODE_IS_FINAL(unNode->node)) { unNode->node->finalOutput += out; }
size_t sz = taosArrayGetSize(unNode->node->trans);
for (size_t i = 0; i < sz; i++) {
FstTransition* trn = taosArrayGet(unNode->node->trans, i);
trn->out += out;
}
if (unNode->last) {
unNode->last->out += out;
}
if (unNode->last) { unNode->last->out += out; }
return;
}
Fst* fstCreate(FstSlice* slice) {
int32_t slen;
char* buf = fstSliceData(slice, &slen);
if (slen < 36) {
return NULL;
}
if (slen < 36) { return NULL; }
uint64_t len = slen;
uint64_t skip = 0;
uint64_t version;
taosDecodeFixedU64(buf, &version);
skip += sizeof(version);
if (version == 0 || version > VERSION) {
return NULL;
}
if (version == 0 || version > VERSION) { return NULL; }
uint64_t type;
taosDecodeFixedU64(buf + skip, &type);
@ -994,14 +947,10 @@ Fst *fstCreate(FstSlice *slice) {
taosDecodeFixedU64(buf + len, &fstLen);
// TODO(validate root addr)
Fst* fst = (Fst*)calloc(1, sizeof(Fst));
if (fst == NULL) {
return NULL;
}
if (fst == NULL) { return NULL; }
fst->meta = (FstMeta*)malloc(sizeof(FstMeta));
if (NULL == fst->meta) {
goto FST_CREAT_FAILED;
}
if (NULL == fst->meta) { goto FST_CREAT_FAILED; }
fst->meta->version = version;
fst->meta->rootAddr = rootAddr;
@ -1039,9 +988,7 @@ bool fstGet(Fst *fst, FstSlice *b, Output *out) {
for (uint32_t i = 0; i < len; i++) {
uint8_t inp = data[i];
Output res = 0;
if (false == fstNodeFindInput(root, inp, &res)) {
return false;
}
if (false == fstNodeFindInput(root, inp, &res)) { return false; }
FstTransition trn;
fstNodeGetTransitionAt(root, res, &trn);
@ -1066,26 +1013,32 @@ bool fstGet(Fst *fst, FstSlice *b, Output *out) {
return true;
}
FstStreamBuilder *fstSearch(Fst *fst, AutomationCtx *ctx) { return fstStreamBuilderCreate(fst, ctx); }
StreamWithState * streamBuilderIntoStream(FstStreamBuilder *sb) {
if (sb == NULL) {
return NULL;
FstStreamBuilder* fstSearch(Fst* fst, AutomationCtx* ctx) {
return fstStreamBuilderCreate(fst, ctx);
}
StreamWithState* streamBuilderIntoStream(FstStreamBuilder* sb) {
if (sb == NULL) { return NULL; }
return streamWithStateCreate(sb->fst, sb->aut, sb->min, sb->max);
}
FstStreamWithStateBuilder *fstSearchWithState(Fst *fst, AutomationCtx *ctx) { return fstStreamBuilderCreate(fst, ctx); }
FstStreamWithStateBuilder* fstSearchWithState(Fst* fst, AutomationCtx* ctx) {
return fstStreamBuilderCreate(fst, ctx);
}
FstNode* fstGetRoot(Fst* fst) {
if (fst->root != NULL) {
return fst->root;
}
if (fst->root != NULL) { return fst->root; }
CompiledAddr rAddr = fstGetRootAddr(fst);
fst->root = fstGetNode(fst, rAddr);
return fst->root;
}
FstNode * fstGetNode(Fst *fst, CompiledAddr addr) { return fstNodeCreate(fst->meta->version, addr, fst->data); }
FstType fstGetType(Fst *fst) { return fst->meta->ty; }
CompiledAddr fstGetRootAddr(Fst *fst) { return fst->meta->rootAddr; }
FstNode* fstGetNode(Fst* fst, CompiledAddr addr) {
return fstNodeCreate(fst->meta->version, addr, fst->data);
}
FstType fstGetType(Fst* fst) {
return fst->meta->ty;
}
CompiledAddr fstGetRootAddr(Fst* fst) {
return fst->meta->rootAddr;
}
Output fstEmptyFinalOutput(Fst* fst, bool* null) {
Output res = 0;
@ -1104,18 +1057,14 @@ bool fstVerify(Fst *fst) {
int32_t len;
uint8_t* data = fstSliceData(fst->data, &len);
TSCKSUM initSum = 0;
if (!taosCheckChecksumWhole(data, len)) {
return false;
}
if (!taosCheckChecksumWhole(data, len)) { return false; }
return true;
}
// data bound function
FstBoundWithData* fstBoundStateCreate(FstBound type, FstSlice* data) {
FstBoundWithData* b = calloc(1, sizeof(FstBoundWithData));
if (b == NULL) {
return NULL;
}
if (b == NULL) { return NULL; }
if (data != NULL) {
b->data = fstSliceCopy(data, data->start, data->end);
@ -1145,17 +1094,18 @@ bool fstBoundWithDataIsEmpty(FstBoundWithData *bound) {
}
}
bool fstBoundWithDataIsIncluded(FstBoundWithData *bound) { return bound->type == Excluded ? false : true; }
void fstBoundDestroy(FstBoundWithData *bound) { free(bound); }
StreamWithState *streamWithStateCreate(
Fst *fst, AutomationCtx *automation, FstBoundWithData *min, FstBoundWithData *max) {
StreamWithState *sws = calloc(1, sizeof(StreamWithState));
if (sws == NULL) {
return NULL;
bool fstBoundWithDataIsIncluded(FstBoundWithData* bound) {
return bound->type == Excluded ? false : true;
}
void fstBoundDestroy(FstBoundWithData* bound) {
free(bound);
}
StreamWithState* streamWithStateCreate(Fst* fst, AutomationCtx* automation, FstBoundWithData* min, FstBoundWithData* max) {
StreamWithState* sws = calloc(1, sizeof(StreamWithState));
if (sws == NULL) { return NULL; }
sws->fst = fst;
sws->aut = automation;
sws->inp = (SArray*)taosArrayInit(256, sizeof(uint8_t));
@ -1170,9 +1120,7 @@ StreamWithState *streamWithStateCreate(
return sws;
}
void streamWithStateDestroy(StreamWithState* sws) {
if (sws == NULL) {
return;
}
if (sws == NULL) { return; }
taosArrayDestroy(sws->inp);
taosArrayDestroyEx(sws->stack, streamStateDestroy);
@ -1183,9 +1131,7 @@ void streamWithStateDestroy(StreamWithState *sws) {
bool streamWithStateSeekMin(StreamWithState* sws, FstBoundWithData* min) {
AutomationCtx* aut = sws->aut;
if (fstBoundWithDataIsEmpty(min)) {
if (fstBoundWithDataIsIncluded(min)) {
sws->emptyOutput.out = fstEmptyFinalOutput(sws->fst, &(sws->emptyOutput.null));
}
if (fstBoundWithDataIsIncluded(min)) { sws->emptyOutput.out = fstEmptyFinalOutput(sws->fst, &(sws->emptyOutput.null)); }
StreamState s = {.node = fstGetRoot(sws->fst),
.trans = 0,
.out = {.null = false, .out = 0},
@ -1195,7 +1141,6 @@ bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min) {
}
FstSlice* key = NULL;
bool inclusize = false;
;
if (min->type == Included) {
key = &min->data;
@ -1239,9 +1184,7 @@ bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min) {
uint64_t i = 0;
for (i = trans->range.start; i < trans->range.end; i++) {
FstTransition trn;
if (fstNodeGetTransitionAt(node, i, &trn) && trn.inp > b) {
break;
}
if (fstNodeGetTransitionAt(node, i, &trn) && trn.inp > b) { break; }
}
StreamState s = {.node = node, .trans = i, .out = {.null = false, .out = out}, .autState = autState};
@ -1260,8 +1203,7 @@ bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min) {
uint64_t trans = s->trans;
FstTransition trn;
fstNodeGetTransitionAt(n, trans - 1, &trn);
StreamState s = {
.node = fstGetNode(sws->fst, trn.addr), .trans = 0, .out = {.null = false, .out = out}, .autState = autState};
StreamState s = {.node = fstGetNode(sws->fst, trn.addr), .trans = 0, .out = {.null = false, .out = out}, .autState = autState};
taosArrayPush(sws->stack, &s);
return true;
}
@ -1289,9 +1231,7 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb
while (taosArrayGetSize(sws->stack) > 0) {
StreamState* p = (StreamState*)taosArrayPop(sws->stack);
if (p->trans >= FST_NODE_LEN(p->node) || !automFuncs[aut->type].canMatch(aut, p->autState)) {
if (FST_NODE_ADDR(p->node) != fstGetRootAddr(sws->fst)) {
taosArrayPop(sws->inp);
}
if (FST_NODE_ADDR(p->node) != fstGetRootAddr(sws->fst)) { taosArrayPop(sws->inp); }
streamStateDestroy(p);
continue;
}
@ -1308,9 +1248,7 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb
if (FST_NODE_IS_FINAL(nextNode)) {
// void *eofState = sws->aut->acceptEof(nextState);
void* eofState = automFuncs[aut->type].acceptEof(aut, nextState);
if (eofState != NULL) {
isMatch = automFuncs[aut->type].isMatch(aut, eofState);
}
if (eofState != NULL) { isMatch = automFuncs[aut->type].isMatch(aut, eofState); }
}
StreamState s1 = {.node = p->node, .trans = p->trans + 1, .out = p->out, .autState = p->autState};
taosArrayPush(sws->stack, &s1);
@ -1351,9 +1289,7 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb
StreamWithStateResult* swsResultCreate(FstSlice* data, FstOutput fOut, void* state) {
StreamWithStateResult* result = calloc(1, sizeof(StreamWithStateResult));
if (result == NULL) {
return NULL;
}
if (result == NULL) { return NULL; }
result->data = fstSliceCopy(data, 0, FST_SLICE_LEN(data) - 1);
result->out = fOut;
@ -1362,9 +1298,7 @@ StreamWithStateResult *swsResultCreate(FstSlice *data, FstOutput fOut, void *sta
return result;
}
void swsResultDestroy(StreamWithStateResult* result) {
if (NULL == result) {
return;
}
if (NULL == result) { return; }
fstSliceDestroy(&result->data);
startWithStateValueDestroy(result->state);
@ -1372,9 +1306,7 @@ void swsResultDestroy(StreamWithStateResult *result) {
}
void streamStateDestroy(void* s) {
if (NULL == s) {
return;
}
if (NULL == s) { return; }
StreamState* ss = (StreamState*)s;
fstNodeDestroy(ss->node);
@ -1383,9 +1315,7 @@ void streamStateDestroy(void *s) {
FstStreamBuilder* fstStreamBuilderCreate(Fst* fst, AutomationCtx* aut) {
FstStreamBuilder* b = calloc(1, sizeof(FstStreamBuilder));
if (NULL == b) {
return NULL;
}
if (NULL == b) { return NULL; }
b->fst = fst;
b->aut = aut;
@ -1401,9 +1331,7 @@ void fstStreamBuilderDestroy(FstStreamBuilder *b) {
free(b);
}
FstStreamBuilder* fstStreamBuilderRange(FstStreamBuilder* b, FstSlice* val, RangeType type) {
if (b == NULL) {
return NULL;
}
if (b == NULL) { return NULL; }
if (type == GE) {
b->min->type = Included;

57
source/libs/index/src/index_fst_automation.c
View File

@ -17,9 +17,7 @@
StartWithStateValue* startWithStateValueCreate(StartWithStateKind kind, ValueType ty, void* val) {
StartWithStateValue* nsv = calloc(1, sizeof(StartWithStateValue));
if (nsv == NULL) {
return NULL;
}
if (nsv == NULL) { return NULL; }
nsv->kind = kind;
nsv->type = ty;
@ -37,9 +35,7 @@ StartWithStateValue *startWithStateValueCreate(StartWithStateKind kind, ValueTyp
}
void startWithStateValueDestroy(void* val) {
StartWithStateValue* sv = (StartWithStateValue*)val;
if (sv == NULL) {
return;
}
if (sv == NULL) { return; }
if (sv->type == FST_INT) {
//
@ -52,9 +48,7 @@ void startWithStateValueDestroy(void *val) {
}
StartWithStateValue* startWithStateValueDump(StartWithStateValue* sv) {
StartWithStateValue* nsv = calloc(1, sizeof(StartWithStateValue));
if (nsv == NULL) {
return NULL;
}
if (nsv == NULL) { return NULL; }
nsv->kind = sv->kind;
nsv->type = sv->type;
@ -65,6 +59,7 @@ StartWithStateValue *startWithStateValueDump(StartWithStateValue *sv) {
nsv->ptr = (char*)calloc(1, len + 1);
memcpy(nsv->ptr, sv->ptr, len);
} else if (nsv->type == FST_ARRAY) {
//
}
return nsv;
}
@ -83,17 +78,15 @@ static bool prefixCanMatch(AutomationCtx *ctx, void *sv) {
StartWithStateValue* ssv = (StartWithStateValue*)sv;
return ssv->val >= 0;
}
static bool prefixWillAlwaysMatch(AutomationCtx *ctx, void *state) { return true; }
static bool prefixWillAlwaysMatch(AutomationCtx* ctx, void* state) {
return true;
}
static void* prefixAccept(AutomationCtx* ctx, void* state, uint8_t byte) {
StartWithStateValue* ssv = (StartWithStateValue*)state;
if (ssv == NULL || ctx == NULL) {
return NULL;
}
if (ssv == NULL || ctx == NULL) { return NULL; }
char* data = ctx->data;
if (ssv->kind == Done) {
return startWithStateValueCreate(Done, FST_INT, &ssv->val);
}
if (ssv->kind == Done) { return startWithStateValueCreate(Done, FST_INT, &ssv->val); }
if ((strlen(data) > ssv->val) && data[ssv->val] == byte) {
int val = ssv->val + 1;
StartWithStateValue* nsv = startWithStateValueCreate(Running, FST_INT, &val);
@ -106,18 +99,32 @@ static void *prefixAccept(AutomationCtx *ctx, void *state, uint8_t byte) {
}
return NULL;
}
static void *prefixAcceptEof(AutomationCtx *ctx, void *state) { return NULL; }
static void* prefixAcceptEof(AutomationCtx* ctx, void* state) {
return NULL;
}
// pattern query, impl later
static void *patternStart(AutomationCtx *ctx) { return NULL; }
static bool patternIsMatch(AutomationCtx *ctx, void *data) { return true; }
static bool patternCanMatch(AutomationCtx *ctx, void *data) { return true; }
static bool patternWillAlwaysMatch(AutomationCtx *ctx, void *state) { return true; }
static void* patternStart(AutomationCtx* ctx) {
return NULL;
}
static bool patternIsMatch(AutomationCtx* ctx, void* data) {
return true;
}
static bool patternCanMatch(AutomationCtx* ctx, void* data) {
return true;
}
static bool patternWillAlwaysMatch(AutomationCtx* ctx, void* state) {
return true;
}
static void *patternAccept(AutomationCtx *ctx, void *state, uint8_t byte) { return NULL; }
static void* patternAccept(AutomationCtx* ctx, void* state, uint8_t byte) {
return NULL;
}
static void *patternAcceptEof(AutomationCtx *ctx, void *state) { return NULL; }
static void* patternAcceptEof(AutomationCtx* ctx, void* state) {
return NULL;
}
AutomationFunc automFuncs[] = {
{prefixStart, prefixIsMatch, prefixCanMatch, prefixWillAlwaysMatch, prefixAccept, prefixAcceptEof},
@ -127,9 +134,7 @@ AutomationFunc automFuncs[] = {
AutomationCtx* automCtxCreate(void* data, AutomationType atype) {
AutomationCtx* ctx = calloc(1, sizeof(AutomationCtx));
if (ctx == NULL) {
return NULL;
}
if (ctx == NULL) { return NULL; }
StartWithStateValue* sv = NULL;
if (atype == AUTOMATION_PREFIX) {

272
source/libs/index/src/index_fst_common.c
View File

@ -274,260 +274,20 @@ const uint8_t COMMON_INPUTS[] = {
};
const char COMMON_INPUTS_INV[] = {
't',
'e',
'/',
'o',
'a',
's',
'r',
'i',
'p',
'c',
'n',
'w',
'.',
'h',
'l',
'm',
'-',
'd',
'u',
'0',
'1',
'2',
'g',
'=',
':',
'b',
'f',
'3',
'y',
'5',
'&',
'_',
'4',
'v',
'9',
'6',
'7',
'8',
'k',
'%',
'?',
'x',
'C',
'D',
'A',
'S',
'F',
'I',
'B',
'E',
'j',
'P',
'T',
'z',
'R',
'N',
'M',
'+',
'L',
'O',
'q',
'H',
'G',
'W',
'U',
'V',
',',
'Y',
'K',
'J',
'Z',
'X',
'Q',
';',
')',
'(',
'~',
'[',
']',
'$',
'!',
'\'',
'*',
'@',
'\x00',
'\x01',
'\x02',
'\x03',
'\x04',
'\x05',
'\x06',
'\x07',
'\x08',
'\t',
'\n',
'\x0b',
'\x0c',
'\r',
'\x0e',
'\x0f',
'\x10',
'\x11',
'\x12',
'\x13',
'\x14',
'\x15',
'\x16',
'\x17',
'\x18',
'\x19',
'\x1a',
'\x1b',
'\x1c',
'\x1d',
'\x1e',
'\x1f',
' ',
'"',
'#',
'<',
'>',
'\\',
'^',
'`',
'{',
'|',
'}',
'\x7f',
'\x80',
'\x81',
'\x82',
'\x83',
'\x84',
'\x85',
'\x86',
'\x87',
'\x88',
'\x89',
'\x8a',
'\x8b',
'\x8c',
'\x8d',
'\x8e',
'\x8f',
'\x90',
'\x91',
'\x92',
'\x93',
'\x94',
'\x95',
'\x96',
'\x97',
'\x98',
'\x99',
'\x9a',
'\x9b',
'\x9c',
'\x9d',
'\x9e',
'\x9f',
'\xa0',
'\xa1',
'\xa2',
'\xa3',
'\xa4',
'\xa5',
'\xa6',
'\xa7',
'\xa8',
'\xa9',
'\xaa',
'\xab',
'\xac',
'\xad',
'\xae',
'\xaf',
'\xb0',
'\xb1',
'\xb2',
'\xb3',
'\xb4',
'\xb5',
'\xb6',
'\xb7',
'\xb8',
'\xb9',
'\xba',
'\xbb',
'\xbc',
'\xbd',
'\xbe',
'\xbf',
'\xc0',
'\xc1',
'\xc2',
'\xc3',
'\xc4',
'\xc5',
'\xc6',
'\xc7',
'\xc8',
'\xc9',
'\xca',
'\xcb',
'\xcc',
'\xcd',
'\xce',
'\xcf',
'\xd0',
'\xd1',
'\xd2',
'\xd3',
'\xd4',
'\xd5',
'\xd6',
'\xd7',
'\xd8',
'\xd9',
'\xda',
'\xdb',
'\xdc',
'\xdd',
'\xde',
'\xdf',
'\xe0',
'\xe1',
'\xe2',
'\xe3',
'\xe4',
'\xe5',
'\xe6',
'\xe7',
'\xe8',
'\xe9',
'\xea',
'\xeb',
'\xec',
'\xed',
'\xee',
'\xef',
'\xf0',
'\xf1',
'\xf2',
'\xf3',
'\xf4',
'\xf5',
'\xf6',
'\xf7',
'\xf8',
'\xf9',
'\xfa',
'\xfb',
'\xfc',
'\xfd',
'\xfe',
'\xff',
't', 'e', '/', 'o', 'a', 's', 'r', 'i', 'p', 'c', 'n', 'w', '.', 'h', 'l', 'm',
'-', 'd', 'u', '0', '1', '2', 'g', '=', ':', 'b', 'f', '3', 'y', '5', '&', '_',
'4', 'v', '9', '6', '7', '8', 'k', '%', '?', 'x', 'C', 'D', 'A', 'S', 'F', 'I',
'B', 'E', 'j', 'P', 'T', 'z', 'R', 'N', 'M', '+', 'L', 'O', 'q', 'H', 'G', 'W',
'U', 'V', ',', 'Y', 'K', 'J', 'Z', 'X', 'Q', ';', ')', '(', '~', '[', ']', '$',
'!', '\'', '*', '@', '\x00', '\x01', '\x02', '\x03', '\x04', '\x05', '\x06', '\x07', '\x08', '\t', '\n', '\x0b',
'\x0c', '\r', '\x0e', '\x0f', '\x10', '\x11', '\x12', '\x13', '\x14', '\x15', '\x16', '\x17', '\x18', '\x19', '\x1a', '\x1b',
'\x1c', '\x1d', '\x1e', '\x1f', ' ', '"', '#', '<', '>', '\\', '^', '`', '{', '|', '}', '\x7f',
'\x80', '\x81', '\x82', '\x83', '\x84', '\x85', '\x86', '\x87', '\x88', '\x89', '\x8a', '\x8b', '\x8c', '\x8d', '\x8e', '\x8f',
'\x90', '\x91', '\x92', '\x93', '\x94', '\x95', '\x96', '\x97', '\x98', '\x99', '\x9a', '\x9b', '\x9c', '\x9d', '\x9e', '\x9f',
'\xa0', '\xa1', '\xa2', '\xa3', '\xa4', '\xa5', '\xa6', '\xa7', '\xa8', '\xa9', '\xaa', '\xab', '\xac', '\xad', '\xae', '\xaf',
'\xb0', '\xb1', '\xb2', '\xb3', '\xb4', '\xb5', '\xb6', '\xb7', '\xb8', '\xb9', '\xba', '\xbb', '\xbc', '\xbd', '\xbe', '\xbf',
'\xc0', '\xc1', '\xc2', '\xc3', '\xc4', '\xc5', '\xc6', '\xc7', '\xc8', '\xc9', '\xca', '\xcb', '\xcc', '\xcd', '\xce', '\xcf',
'\xd0', '\xd1', '\xd2', '\xd3', '\xd4', '\xd5', '\xd6', '\xd7', '\xd8', '\xd9', '\xda', '\xdb', '\xdc', '\xdd', '\xde', '\xdf',
'\xe0', '\xe1', '\xe2', '\xe3', '\xe4', '\xe5', '\xe6', '\xe7', '\xe8', '\xe9', '\xea', '\xeb', '\xec', '\xed', '\xee', '\xef',
'\xf0', '\xf1', '\xf2', '\xf3', '\xf4', '\xf5', '\xf6', '\xf7', '\xf8', '\xf9', '\xfa', '\xfb', '\xfc', '\xfd', '\xfe', '\xff',
};

42
source/libs/index/src/index_fst_counting_writer.c
View File

@ -18,9 +18,7 @@
#include "tutil.h"
static int writeCtxDoWrite(WriterCtx* ctx, uint8_t* buf, int len) {
if (ctx->offset + len > ctx->limit) {
return -1;
}
if (ctx->offset + len > ctx->limit) { return -1; }
if (ctx->type == TFile) {
assert(len == tfWrite(ctx->file.fd, buf, len));
@ -41,6 +39,17 @@ static int writeCtxDoRead(WriterCtx *ctx, uint8_t *buf, int len) {
return nRead;
}
static int writeCtxDoReadFrom(WriterCtx* ctx, uint8_t* buf, int len, int32_t offset) {
int nRead = 0;
if (ctx->type == TFile) {
tfLseek(ctx->file.fd, offset, 0);
nRead = tfRead(ctx->file.fd, buf, len);
} else {
// refactor later
assert(0);
}
return nRead;
}
static int writeCtxDoFlush(WriterCtx* ctx) {
if (ctx->type == TFile) {
// tfFsync(ctx->fd);
@ -53,9 +62,7 @@ static int writeCtxDoFlush(WriterCtx *ctx) {
WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int32_t capacity) {
WriterCtx* ctx = calloc(1, sizeof(WriterCtx));
if (ctx == NULL) {
return NULL;
}
if (ctx == NULL) { return NULL; }
ctx->type = type;
if (ctx->type == TFile) {
@ -67,8 +74,8 @@ WriterCtx *writerCtxCreate(WriterType type, const char *path, bool readOnly, int
ctx->file.fd = tfOpenReadWrite(tmpFile);
}
if (ctx->file.fd < 0) {
goto END;
indexError("open file error %d", errno);
goto END;
}
} else if (ctx->type == TMemory) {
ctx->mem.buf = calloc(1, sizeof(char) * capacity);
@ -77,15 +84,14 @@ WriterCtx *writerCtxCreate(WriterType type, const char *path, bool readOnly, int
ctx->write = writeCtxDoWrite;
ctx->read = writeCtxDoRead;
ctx->flush = writeCtxDoFlush;
ctx->readFrom = writeCtxDoReadFrom;
ctx->offset = 0;
ctx->limit = capacity;
return ctx;
END:
if (ctx->type == TMemory) {
free(ctx->mem.buf);
}
if (ctx->type == TMemory) { free(ctx->mem.buf); }
free(ctx);
}
void writerCtxDestroy(WriterCtx* ctx) {
@ -99,9 +105,7 @@ void writerCtxDestroy(WriterCtx *ctx) {
FstCountingWriter* fstCountingWriterCreate(void* wrt) {
FstCountingWriter* cw = calloc(1, sizeof(FstCountingWriter));
if (cw == NULL) {
return NULL;
}
if (cw == NULL) { return NULL; }
cw->wrt = wrt;
//(void *)(writerCtxCreate(TFile, readOnly));
@ -115,9 +119,7 @@ void fstCountingWriterDestroy(FstCountingWriter *cw) {
}
int fstCountingWriterWrite(FstCountingWriter* write, uint8_t* buf, uint32_t len) {
if (write == NULL) {
return 0;
}
if (write == NULL) { return 0; }
// update checksum
// write data to file/socket or mem
WriterCtx* ctx = write->wrt;
@ -128,16 +130,16 @@ int fstCountingWriterWrite(FstCountingWriter *write, uint8_t *buf, uint32_t len)
return len;
}
int fstCountingWriterRead(FstCountingWriter* write, uint8_t* buf, uint32_t len) {
if (write == NULL) {
return 0;
}
if (write == NULL) { return 0; }
WriterCtx* ctx = write->wrt;
int nRead = ctx->read(ctx, buf, len);
// assert(nRead == len);
return nRead;
}
uint32_t fstCountingWriterMaskedCheckSum(FstCountingWriter *write) { return 0; }
uint32_t fstCountingWriterMaskedCheckSum(FstCountingWriter* write) {
return 0;
}
int fstCountingWriterFlush(FstCountingWriter* write) {
WriterCtx* ctx = write->wrt;
ctx->flush(ctx);

32
source/libs/index/src/index_fst_node.c
View File

@ -22,45 +22,31 @@ FstBuilderNode *fstBuilderNodeDefault() {
return bn;
}
void fstBuilderNodeDestroy(FstBuilderNode* node) {
if (node == NULL) {
return;
}
if (node == NULL) { return; }
taosArrayDestroy(node->trans);
free(node);
}
bool fstBuilderNodeEqual(FstBuilderNode* n1, FstBuilderNode* n2) {
if (n1 == n2) {
return true;
}
if (n1 == NULL || n2 == NULL) {
return false;
}
if (n1 == n2) { return true; }
if (n1 == NULL || n2 == NULL) { return false; }
if (n1->isFinal != n2->isFinal || n1->finalOutput != n2->finalOutput) {
return false;
}
if (n1->isFinal != n2->isFinal || n1->finalOutput != n2->finalOutput) { return false; }
size_t s1 = n1->trans ? taosArrayGetSize(n1->trans) : 0;
size_t s2 = n2->trans ? taosArrayGetSize(n2->trans) : 0;
if (s1 != s2) {
return false;
}
if (s1 != s2) { return false; }
for (size_t i = 0; i < s1; i++) {
FstTransition* t1 = taosArrayGet(n1->trans, i);
FstTransition* t2 = taosArrayGet(n2->trans, i);
if (t1->inp != t2->inp || t1->out != t2->out || t1->addr != t2->addr) {
return false;
}
if (t1->inp != t2->inp || t1->out != t2->out || t1->addr != t2->addr) { return false; }
}
return true;
}
FstBuilderNode* fstBuilderNodeClone(FstBuilderNode* src) {
FstBuilderNode* node = malloc(sizeof(FstBuilderNode));
if (node == NULL) {
return NULL;
}
if (node == NULL) { return NULL; }
//
size_t sz = taosArrayGetSize(src->trans);
@ -78,9 +64,7 @@ FstBuilderNode *fstBuilderNodeClone(FstBuilderNode *src) {
}
// not destroy src, User's bussiness
void fstBuilderNodeCloneFrom(FstBuilderNode* dst, FstBuilderNode* src) {
if (dst == NULL || src == NULL) {
return;
}
if (dst == NULL || src == NULL) { return; }
dst->isFinal = src->isFinal;
dst->finalOutput = src->finalOutput;

24
source/libs/index/src/index_fst_registry.c
View File

@ -34,9 +34,7 @@ uint64_t fstRegistryHash(FstRegistry *registry, FstBuilderNode *bNode) {
}
static void fstRegistryCellSwap(SArray* arr, uint32_t a, uint32_t b) {
size_t sz = taosArrayGetSize(arr);
if (a >= sz || b >= sz) {
return;
}
if (a >= sz || b >= sz) { return; }
FstRegistryCell* cell1 = (FstRegistryCell*)taosArrayGet(arr, a);
FstRegistryCell* cell2 = (FstRegistryCell*)taosArrayGet(arr, b);
@ -53,9 +51,7 @@ static void fstRegistryCellSwap(SArray *arr, uint32_t a, uint32_t b) {
static void fstRegistryCellPromote(SArray* arr, uint32_t start, uint32_t end) {
size_t sz = taosArrayGetSize(arr);
if (start >= sz && end >= sz) {
return;
}
if (start >= sz && end >= sz) { return; }
assert(start >= end);
@ -69,9 +65,7 @@ static void fstRegistryCellPromote(SArray *arr, uint32_t start, uint32_t end) {
FstRegistry* fstRegistryCreate(uint64_t tableSize, uint64_t mruSize) {
FstRegistry* registry = malloc(sizeof(FstRegistry));
if (registry == NULL) {
return NULL;
}
if (registry == NULL) { return NULL; }
uint64_t nCells = tableSize * mruSize;
SArray* tb = (SArray*)taosArrayInit(nCells, sizeof(FstRegistryCell));
@ -92,9 +86,7 @@ FstRegistry *fstRegistryCreate(uint64_t tableSize, uint64_t mruSize) {
}
void fstRegistryDestroy(FstRegistry* registry) {
if (registry == NULL) {
return;
}
if (registry == NULL) { return; }
SArray* tb = registry->table;
size_t sz = taosArrayGetSize(tb);
@ -107,9 +99,7 @@ void fstRegistryDestroy(FstRegistry *registry) {
}
FstRegistryEntry* fstRegistryGetEntry(FstRegistry* registry, FstBuilderNode* bNode) {
if (taosArrayGetSize(registry->table) <= 0) {
return NULL;
}
if (taosArrayGetSize(registry->table) <= 0) { return NULL; }
uint64_t bucket = fstRegistryHash(registry, bNode);
uint64_t start = registry->mruSize * bucket;
uint64_t end = start + registry->mruSize;
@ -174,4 +164,6 @@ FstRegistryEntry *fstRegistryGetEntry(FstRegistry *registry, FstBuilderNode *bNo
}
return entry;
}
void fstRegistryEntryDestroy(FstRegistryEntry *entry) { free(entry); }
void fstRegistryEntryDestroy(FstRegistryEntry* entry) {
free(entry);
}

9
source/libs/index/src/index_fst_util.c
View File

@ -64,6 +64,7 @@ uint8_t packSize(uint64_t n) {
uint64_t unpackUint64(uint8_t* ch, uint8_t sz) {
uint64_t n = 0;
for (uint8_t i = 0; i < sz; i++) {
//
n = n | (ch[i] << (8 * i));
}
return n;
@ -129,13 +130,13 @@ FstSlice fstSliceDeepCopy(FstSlice *s, int32_t start, int32_t end) {
ans.end = tlen - 1;
return ans;
}
bool fstSliceIsEmpty(FstSlice *s) { return s->str == NULL || s->str->len == 0 || s->start < 0 || s->end < 0; }
bool fstSliceIsEmpty(FstSlice* s) {
return s->str == NULL || s->str->len == 0 || s->start < 0 || s->end < 0;
}
uint8_t* fstSliceData(FstSlice* s, int32_t* size) {
FstString* str = s->str;
if (size != NULL) {
*size = s->end - s->start + 1;
}
if (size != NULL) { *size = s->end - s->start + 1; }
return str->data + s->start;
}
void fstSliceDestroy(FstSlice* s) {

412
source/libs/index/src/index_tfile.c
View File

@ -21,80 +21,36 @@
#include "index_fst_counting_writer.h"
#include "index_util.h"
#include "taosdef.h"
#include "tcompare.h"
static FORCE_INLINE int tfileReadLoadHeader(TFileReader *reader) {
// TODO simple tfile header later
char buf[TFILE_HADER_PRE_SIZE];
char * p = buf;
TFileReadHeader *header = &reader->header;
int64_t nread = reader->ctx->read(reader->ctx, buf, TFILE_HADER_PRE_SIZE);
assert(nread == TFILE_HADER_PRE_SIZE);
#define TF_TABLE_TATOAL_SIZE(sz) (sizeof(sz) + sz * sizeof(uint64_t))
memcpy(&header->suid, p, sizeof(header->suid));
p += sizeof(header->suid);
typedef struct TFileValue {
char* colVal; // null terminated
SArray* tableId;
int32_t offset;
} TFileValue;
memcpy(&header->version, p, sizeof(header->version));
p += sizeof(header->version);
static int tfileValueCompare(const void* a, const void* b, const void* param);
static void tfileSerialTableIdsToBuf(char* buf, SArray* tableIds);
int32_t colLen = 0;
memcpy(&colLen, p, sizeof(colLen));
assert(colLen < sizeof(header->colName));
nread = reader->ctx->read(reader->ctx, header->colName, colLen);
assert(nread == colLen);
static int tfileWriteHeader(TFileWriter* writer);
static int tfileWriteFstOffset(TFileWriter* tw, int32_t offset);
static int tfileWriteData(TFileWriter* write, TFileValue* tval);
nread = reader->ctx->read(reader->ctx, &header->colType, sizeof(header->colType));
return 0;
};
static int tfileGetFileList(const char *path, SArray *result) {
DIR *dir = opendir(path);
if (NULL == dir) {
return -1;
}
static int tfileReadLoadHeader(TFileReader* reader);
static int tfileReadLoadFst(TFileReader* reader);
static int tfileReadLoadTableIds(TFileReader* reader, int32_t offset, SArray* result);
struct dirent *entry;
while ((entry = readdir(dir)) != NULL) {
size_t len = strlen(entry->d_name);
char * buf = calloc(1, len + 1);
memcpy(buf, entry->d_name, len);
taosArrayPush(result, &buf);
}
closedir(dir);
return 0;
}
static void tfileDestroyFileName(void *elem) {
char *p = *(char **)elem;
free(p);
}
static int tfileCompare(const void *a, const void *b) {
const char *aName = *(char **)a;
const char *bName = *(char **)b;
size_t aLen = strlen(aName);
size_t bLen = strlen(bName);
return strncmp(aName, bName, aLen > bLen ? aLen : bLen);
}
// tfile name suid-colId-version.tindex
static int tfileParseFileName(const char *filename, uint64_t *suid, int *colId, int *version) {
if (3 == sscanf(filename, "%" PRIu64 "-%d-%d.tindex", suid, colId, version)) {
// read suid & colid & version success
return 0;
}
return -1;
}
static void tfileSerialCacheKey(TFileCacheKey *key, char *buf) {
SERIALIZE_MEM_TO_BUF(buf, key, suid);
SERIALIZE_VAR_TO_BUF(buf, '_', char);
SERIALIZE_MEM_TO_BUF(buf, key, colType);
SERIALIZE_VAR_TO_BUF(buf, '_', char);
SERIALIZE_MEM_TO_BUF(buf, key, version);
SERIALIZE_VAR_TO_BUF(buf, '_', char);
SERIALIZE_STR_MEM_TO_BUF(buf, key, colName, key->nColName);
}
static int tfileGetFileList(const char* path, 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 tfileSerialCacheKey(TFileCacheKey* key, char* buf);
TFileCache* tfileCacheCreate(const char* path) {
TFileCache* tcache = calloc(1, sizeof(TFileCache));
if (tcache == NULL) {
return NULL;
}
if (tcache == NULL) { return NULL; }
tcache->tableCache = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
tcache->capacity = 64;
@ -102,15 +58,14 @@ TFileCache *tfileCacheCreate(const char *path) {
SArray* files = taosArrayInit(4, sizeof(void*));
tfileGetFileList(path, files);
taosArraySort(files, tfileCompare);
uint64_t suid;
int32_t colId, version;
for (size_t i = 0; i < taosArrayGetSize(files); i++) {
char* file = taosArrayGetP(files, i);
uint64_t suid;
int colId, version;
if (0 != tfileParseFileName(file, &suid, &colId, &version)) {
goto End;
if (0 != tfileParseFileName(file, &suid, (int*)&colId, (int*)&version)) {
indexInfo("try parse invalid file: %s, skip it", file);
continue;
}
WriterCtx* wc = writerCtxCreate(TFile, file, true, 1024 * 64);
if (wc == NULL) {
indexError("failed to open index: %s", file);
@ -118,10 +73,26 @@ TFileCache *tfileCacheCreate(const char *path) {
}
TFileReader* reader = tfileReaderCreate(wc);
if (0 != tfileReadLoadHeader(reader)) {
TFileReaderDestroy(reader);
indexError("failed to load index header, index Id: %s", file);
tfileReaderDestroy(reader);
indexError("failed to load index header, index file: %s", file);
goto End;
}
if (0 != tfileReadLoadFst(reader)) {
tfileReaderDestroy(reader);
indexError("failed to load index fst, index file: %s", file);
}
// loader fst and validate it
TFileHeader* header = &reader->header;
TFileCacheKey key = {.suid = header->suid,
.version = header->version,
.colName = header->colName,
.nColName = strlen(header->colName),
.colType = header->colType};
char buf[128] = {0};
tfileSerialCacheKey(&key, buf);
taosHashPut(tcache->tableCache, buf, strlen(buf), &reader, sizeof(void*));
}
taosArrayDestroyEx(files, tfileDestroyFileName);
return tcache;
@ -131,17 +102,15 @@ End:
return NULL;
}
void tfileCacheDestroy(TFileCache* tcache) {
if (tcache == NULL) {
return;
}
if (tcache == NULL) { return; }
// free table cache
TFileReader** reader = taosHashIterate(tcache->tableCache, NULL);
while (reader) {
TFileReader* p = *reader;
indexInfo("drop table cache suid: %" PRIu64 ", colName: %s, colType: %d", p->header.suid, p->header.colName,
p->header.colType);
TFileReaderDestroy(p);
indexInfo("drop table cache suid: %" PRIu64 ", colName: %s, colType: %d", p->header.suid, p->header.colName, p->header.colType);
tfileReaderDestroy(p);
reader = taosHashIterate(tcache->tableCache, reader);
}
taosHashCleanup(tcache->tableCache);
@ -163,45 +132,298 @@ void tfileCachePut(TFileCache *tcache, TFileCacheKey *key, TFileReader *reader)
TFileReader* tfileReaderCreate(WriterCtx* ctx) {
TFileReader* reader = calloc(1, sizeof(TFileReader));
if (reader == NULL) {
return NULL;
}
reader->ctx = ctx;
if (reader == NULL) { return NULL; }
// T_REF_INC(reader);
reader->ctx = ctx;
return reader;
}
void TFileReaderDestroy(TFileReader *reader) {
if (reader == NULL) {
return;
}
void tfileReaderDestroy(TFileReader* reader) {
if (reader == NULL) { return; }
// T_REF_INC(reader);
fstDestroy(reader->fst);
writerCtxDestroy(reader->ctx);
free(reader);
}
TFileWriter *tfileWriterCreate(const char *suid, const char *colName);
void tfileWriterDestroy(TFileWriter *tw);
int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* result) {
SIndexTerm* term = query->term;
// refactor to callback later
if (query->qType == QUERY_TERM) {
uint64_t offset;
FstSlice key = fstSliceCreate(term->colVal, term->nColVal);
if (fstGet(reader->fst, &key, &offset)) {
return tfileReadLoadTableIds(reader, offset, result);
} else {
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, not found in tindex", term->suid, term->colName, term->colVal);
}
return 0;
} else if (query->qType == QUERY_PREFIX) {
// handle later
//
} else {
// handle later
}
return 0;
}
TFileWriter* tfileWriterCreate(WriterCtx* ctx, TFileHeader* header) {
// char pathBuf[128] = {0};
// sprintf(pathBuf, "%s/% " PRIu64 "-%d-%d.tindex", path, suid, colId, version);
// TFileHeader header = {.suid = suid, .version = version, .colName = {0}, colType = colType};
// memcpy(header.colName, );
// char buf[TFILE_HADER_PRE_SIZE];
// int len = TFILE_HADER_PRE_SIZE;
// if (len != ctx->write(ctx, buf, len)) {
// indexError("index: %" PRIu64 " failed to write header info", header->suid);
// return NULL;
//}
TFileWriter* tw = calloc(1, sizeof(TFileWriter));
if (tw == NULL) {
indexError("index: %" PRIu64 " failed to alloc TFilerWriter", header->suid);
return NULL;
}
tw->ctx = ctx;
tw->header = *header;
tfileWriteHeader(tw);
return tw;
}
int tfileWriterPut(TFileWriter* tw, void* data) {
// sort by coltype and write to tindex
__compar_fn_t fn = getComparFunc(tw->header.colType, 0);
taosArraySortPWithExt((SArray*)(data), tfileValueCompare, &fn);
int32_t bufLimit = 4096, offset = 0;
char* buf = calloc(1, sizeof(bufLimit));
char* p = buf;
int32_t sz = taosArrayGetSize((SArray*)data);
int32_t fstOffset = tw->offset;
// ugly code, refactor later
for (size_t i = 0; i < sz; i++) {
TFileValue* v = taosArrayGetP((SArray*)data, i);
int32_t tbsz = taosArrayGetSize(v->tableId);
fstOffset += TF_TABLE_TATOAL_SIZE(tbsz);
}
// check result or not
tfileWriteFstOffset(tw, fstOffset);
for (size_t i = 0; i < sz; i++) {
TFileValue* v = taosArrayGetP((SArray*)data, i);
int32_t tbsz = taosArrayGetSize(v->tableId);
// check buf has enough space or not
int32_t ttsz = TF_TABLE_TATOAL_SIZE(tbsz);
if (offset + ttsz > bufLimit) {
// batch write
tw->ctx->write(tw->ctx, buf, offset);
offset = 0;
memset(buf, 0, bufLimit);
p = buf;
}
tfileSerialTableIdsToBuf(p, v->tableId);
offset += ttsz;
p = buf + offset;
// set up value offset
v->offset = tw->offset;
tw->offset += ttsz;
}
if (offset != 0) {
// write reversed data in buf to tindex
tw->ctx->write(tw->ctx, buf, offset);
}
// write fst
for (size_t i = 0; i < sz; i++) {
// TODO, fst batch write later
TFileValue* v = taosArrayGetP((SArray*)data, i);
if (tfileWriteData(tw, v) == 0) {
//
//
}
}
tfree(buf);
return 0;
}
void tfileWriterDestroy(TFileWriter* tw) {
if (tw == NULL) { return; }
writerCtxDestroy(tw->ctx);
free(tw);
}
IndexTFile* indexTFileCreate(const char* path) {
IndexTFile* tfile = calloc(1, sizeof(IndexTFile));
tfile->cache = tfileCacheCreate(path);
if (tfile == NULL) { return NULL; }
tfile->cache = tfileCacheCreate(path);
return tfile;
}
void IndexTFileDestroy(IndexTFile *tfile) { free(tfile); }
void IndexTFileDestroy(IndexTFile* tfile) {
free(tfile);
}
int indexTFileSearch(void* tfile, SIndexTermQuery* query, SArray* result) {
if (tfile == NULL) { return -1; }
IndexTFile* pTfile = (IndexTFile*)tfile;
SIndexTerm* term = query->term;
TFileCacheKey key = {
.suid = term->suid, .colType = term->colType, .version = 0, .colName = term->colName, .nColName = term->nColName};
TFileCacheKey key = {.suid = term->suid, .colType = term->colType, .version = 0, .colName = term->colName, .nColName = term->nColName};
TFileReader* reader = tfileCacheGet(pTfile->cache, &key);
return 0;
return tfileReaderSearch(reader, query, result);
}
int indexTFilePut(void* tfile, SIndexTerm* term, uint64_t uid) {
TFileWriterOpt wOpt = {
.suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName = term->nColName, .version = 1};
TFileWriterOpt wOpt = {.suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName = term->nColName, .version = 1};
return 0;
}
static int tfileValueCompare(const void* a, const void* b, const void* param) {
__compar_fn_t fn = *(__compar_fn_t*)param;
TFileValue* av = (TFileValue*)a;
TFileValue* bv = (TFileValue*)b;
return fn(av->colVal, bv->colVal);
}
static void tfileSerialTableIdsToBuf(char* buf, SArray* ids) {
int sz = taosArrayGetSize(ids);
SERIALIZE_VAR_TO_BUF(buf, sz, int32_t);
for (size_t i = 0; i < sz; i++) {
uint64_t* v = taosArrayGet(ids, i);
SERIALIZE_VAR_TO_BUF(buf, *v, uint64_t);
}
}
static int tfileWriteFstOffset(TFileWriter* tw, int32_t offset) {
int32_t fstOffset = offset + sizeof(tw->header.fstOffset);
tw->header.fstOffset = fstOffset;
if (sizeof(fstOffset) != tw->ctx->write(tw->ctx, (char*)&fstOffset, sizeof(fstOffset))) { return -1; }
return 0;
}
static int tfileWriteHeader(TFileWriter* writer) {
char buf[TFILE_HEADER_NO_FST] = {0};
char* p = buf;
TFileHeader* header = &writer->header;
memcpy(buf, (char*)header, sizeof(buf));
int nwrite = writer->ctx->write(writer->ctx, buf, sizeof(buf));
if (sizeof(buf) != nwrite) { return -1; }
writer->offset = nwrite;
return 0;
}
static int tfileWriteData(TFileWriter* write, TFileValue* tval) {
TFileHeader* header = &write->header;
uint8_t colType = header->colType;
if (colType == TSDB_DATA_TYPE_BINARY || colType == TSDB_DATA_TYPE_NCHAR) {
FstSlice key = fstSliceCreate((uint8_t*)(tval->colVal), (size_t)strlen(tval->colVal));
if (fstBuilderInsert(write->fb, key, tval->offset)) {
fstSliceDestroy(&key);
return 0;
}
fstSliceDestroy(&key);
return -1;
} else {
// handle other type later
}
return 0;
}
static int tfileReadLoadHeader(TFileReader* reader) {
// TODO simple tfile header later
char buf[TFILE_HEADER_SIZE] = {0};
char* p = buf;
int64_t nread = reader->ctx->read(reader->ctx, buf, sizeof(buf));
assert(nread == sizeof(buf));
memcpy(&reader->header, buf, sizeof(buf));
return 0;
}
static int tfileReadLoadFst(TFileReader* reader) {
// current load fst into memory, refactor it later
static int FST_MAX_SIZE = 16 * 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);
// we assuse fst size less than FST_MAX_SIZE
assert(nread > 0 && nread < FST_MAX_SIZE);
FstSlice st = fstSliceCreate((uint8_t*)buf, nread);
reader->fst = fstCreate(&st);
free(buf);
fstSliceDestroy(&st);
return reader->fst == NULL ? 0 : -1;
}
static int tfileReadLoadTableIds(TFileReader* reader, int32_t offset, SArray* result) {
int32_t nid;
WriterCtx* ctx = reader->ctx;
int32_t nread = ctx->readFrom(ctx, (char*)&nid, sizeof(nid), offset);
assert(sizeof(nid) == nread);
char* buf = calloc(1, sizeof(uint64_t) * nid);
if (buf == NULL) { return -1; }
nread = ctx->read(ctx, buf, sizeof(uint64_t) * nid);
uint64_t* ids = (uint64_t*)buf;
for (int32_t i = 0; i < nid; i++) {
taosArrayPush(result, ids + i);
}
free(buf);
return 0;
}
static int tfileGetFileList(const char* path, SArray* result) {
DIR* dir = opendir(path);
if (NULL == dir) { return -1; }
struct dirent* entry;
while ((entry = readdir(dir)) != NULL) {
size_t len = strlen(entry->d_name);
char* buf = calloc(1, len + 1);
memcpy(buf, entry->d_name, len);
taosArrayPush(result, &buf);
}
closedir(dir);
return 0;
}
static void tfileDestroyFileName(void* elem) {
char* p = *(char**)elem;
free(p);
}
static int tfileCompare(const void* a, const void* b) {
const char* aName = *(char**)a;
const char* bName = *(char**)b;
size_t aLen = strlen(aName);
size_t bLen = strlen(bName);
return strncmp(aName, bName, aLen > bLen ? aLen : bLen);
}
// tfile name suid-colId-version.tindex
static int tfileParseFileName(const char* filename, uint64_t* suid, int* colId, int* version) {
if (3 == sscanf(filename, "%" PRIu64 "-%d-%d.tindex", suid, colId, version)) {
// read suid & colid & version success
return 0;
}
return -1;
}
static void tfileSerialCacheKey(TFileCacheKey* key, char* buf) {
SERIALIZE_MEM_TO_BUF(buf, key, suid);
SERIALIZE_VAR_TO_BUF(buf, '_', char);
SERIALIZE_MEM_TO_BUF(buf, key, colType);
SERIALIZE_VAR_TO_BUF(buf, '_', char);
SERIALIZE_MEM_TO_BUF(buf, key, version);
SERIALIZE_VAR_TO_BUF(buf, '_', char);
SERIALIZE_STR_MEM_TO_BUF(buf, key, colName, key->nColName);
}

3
source/libs/parser/inc/dataBlockMgt.h
View File

@ -170,10 +170,11 @@ int32_t schemaIdxCompar(const void *lhs, const void *rhs);
int32_t boundIdxCompar(const void *lhs, const void *rhs);
void setBoundColumnInfo(SParsedDataColInfo* pColList, SSchema* pSchema, int32_t numOfCols);
void destroyBoundColumnInfo(SParsedDataColInfo* pColList);
void destroyBlockArrayList(SArray* pDataBlockList);
int32_t initMemRowBuilder(SMemRowBuilder *pBuilder, uint32_t nRows, uint32_t nCols, uint32_t nBoundCols, int32_t allNullLen);
int32_t allocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int32_t * numOfRows);
int32_t getDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, int32_t startOffset, int32_t rowSize,
const STableMeta* pTableMeta, STableDataBlocks** dataBlocks, SArray* pBlockList);
int32_t mergeTableDataBlocks(SHashObj* pHashObj, int8_t schemaAttached, uint8_t payloadType, bool freeBlockMap);
int32_t mergeTableDataBlocks(SHashObj* pHashObj, int8_t schemaAttached, uint8_t payloadType, SArray** pVgDataBlocks);
#endif // TDENGINE_DATABLOCKMGT_H

38
source/libs/parser/src/dataBlockMgt.c
View File

@ -269,19 +269,17 @@ void destroyDataBlock(STableDataBlocks* pDataBlock) {
tfree(pDataBlock);
}
void* destroyBlockArrayList(SArray* pDataBlockList) {
void destroyBlockArrayList(SArray* pDataBlockList) {
if (pDataBlockList == NULL) {
return NULL;
return;
}
size_t size = taosArrayGetSize(pDataBlockList);
for (int32_t i = 0; i < size; i++) {
void* d = taosArrayGetP(pDataBlockList, i);
destroyDataBlock(d);
destroyDataBlock(taosArrayGetP(pDataBlockList, i));
}
taosArrayDestroy(pDataBlockList);
return NULL;
}
// data block is disordered, sort it in ascending order
@ -298,6 +296,7 @@ void sortRemoveDataBlockDupRowsRaw(STableDataBlocks *dataBuf) {
int32_t i = 0;
int32_t j = 1;
// delete rows with timestamp conflicts
while (j < pBlocks->numOfRows) {
TSKEY ti = *(TSKEY *)(pBlockData + dataBuf->rowSize * i);
TSKEY tj = *(TSKEY *)(pBlockData + dataBuf->rowSize * j);
@ -430,7 +429,7 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, SB
char* p = pTableDataBlock->pData + sizeof(SSubmitBlk);
pBlock->dataLen = 0;
int32_t numOfRows = htons(pBlock->numOfRows);
int32_t numOfRows = pBlock->numOfRows;
if (isRawPayload) {
for (int32_t i = 0; i < numOfRows; ++i) {
@ -467,18 +466,10 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, SB
}
}
int32_t len = pBlock->dataLen + pBlock->schemaLen;
pBlock->dataLen = htonl(pBlock->dataLen);
pBlock->schemaLen = htonl(pBlock->schemaLen);
return len;
return pBlock->dataLen + pBlock->schemaLen;
}
static void extractTableNameList(SHashObj* pHashObj, bool freeBlockMap) {
// todo
}
int32_t mergeTableDataBlocks(SHashObj* pHashObj, int8_t schemaAttached, uint8_t payloadType, bool freeBlockMap) {
int32_t mergeTableDataBlocks(SHashObj* pHashObj, int8_t schemaAttached, uint8_t payloadType, SArray** pVgDataBlocks) {
const int INSERT_HEAD_SIZE = sizeof(SMsgDesc) + sizeof(SSubmitMsg);
int code = 0;
bool isRawPayload = IS_RAW_PAYLOAD(payloadType);
@ -537,24 +528,13 @@ int32_t mergeTableDataBlocks(SHashObj* pHashObj, int8_t schemaAttached, uint8_t
(isRawPayload ? (pOneTableBlock->rowSize + expandSize) : getExtendedRowSize(pOneTableBlock)) +
sizeof(STColumn) * getNumOfColumns(pOneTableBlock->pTableMeta);
pBlocks->tid = htonl(pBlocks->tid);
pBlocks->uid = htobe64(pBlocks->uid);
pBlocks->sversion = htonl(pBlocks->sversion);
pBlocks->numOfRows = htons(pBlocks->numOfRows);
pBlocks->schemaLen = 0;
// erase the empty space reserved for binary data
int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock, blkKeyInfo.pKeyTuple, schemaAttached, isRawPayload);
assert(finalLen <= len);
dataBuf->size += (finalLen + sizeof(SSubmitBlk));
assert(dataBuf->size <= dataBuf->nAllocSize);
// the length does not include the SSubmitBlk structure
pBlocks->dataLen = htonl(finalLen);
dataBuf->numOfTables += 1;
pBlocks->numOfRows = 0;
}
p = taosHashIterate(pHashObj, p);
@ -565,12 +545,10 @@ int32_t mergeTableDataBlocks(SHashObj* pHashObj, int8_t schemaAttached, uint8_t
pOneTableBlock = *p;
}
extractTableNameList(pHashObj, freeBlockMap);
// free the table data blocks;
taosHashCleanup(pVnodeDataBlockHashList);
tfree(blkKeyInfo.pKeyTuple);
*pVgDataBlocks = pVnodeDataBlockList;
return TSDB_CODE_SUCCESS;
}

171
source/libs/parser/src/insertParser.c
View File

@ -31,33 +31,16 @@
pSql += index; \
} while (0)
#define NEXT_TOKEN_KEEP_SQL(pSql, sToken, index) \
do { \
sToken = tStrGetToken(pSql, &index, false); \
} while (0)
#define CHECK_CODE(expr) \
do { \
int32_t code = expr; \
if (TSDB_CODE_SUCCESS != code) { \
terrno = code; \
return terrno; \
} \
} while (0)
#define CHECK_CODE_1(expr, destroy) \
do { \
int32_t code = expr; \
if (TSDB_CODE_SUCCESS != code) { \
(void)destroy; \
terrno = code; \
return terrno; \
} \
} while (0)
#define CHECK_CODE_2(expr, destroy1, destroy2) \
do { \
int32_t code = expr; \
if (TSDB_CODE_SUCCESS != code) { \
(void)destroy1; \
(void)destroy2; \
terrno = code; \
return terrno; \
return code; \
} \
} while (0)
@ -70,9 +53,12 @@ typedef struct SInsertParseContext {
SParseContext* pComCxt;
const char* pSql;
SMsgBuf msg;
struct SCatalog* pCatalog;
STableMeta* pTableMeta;
SHashObj* pTableBlockHashObj; // data block for each table. need release
SParsedDataColInfo tags;
SKVRowBuilder tagsBuilder;
SHashObj* pTableBlockHashObj;
SArray* pTableDataBlocks;
SArray* pVgDataBlocks;
int32_t totalNum;
SInsertStmtInfo* pOutput;
} SInsertParseContext;
@ -151,14 +137,6 @@ static int32_t toInt64(const char* z, int16_t type, int32_t n, int64_t* value, b
return ret;
}
static int32_t createInsertStmtInfo(SInsertStmtInfo **pInsertInfo) {
*pInsertInfo = calloc(1, sizeof(SQueryStmtInfo));
if (NULL == *pInsertInfo) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static int32_t skipInsertInto(SInsertParseContext* pCxt) {
SToken sToken;
NEXT_TOKEN(pCxt->pSql, sToken);
@ -194,7 +172,7 @@ static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) {
char fullDbName[TSDB_FULL_DB_NAME_LEN] = {0};
char tableName[TSDB_TABLE_NAME_LEN] = {0};
CHECK_CODE(buildName(pCxt, pTname, fullDbName, tableName));
CHECK_CODE(catalogGetTableMeta(pCxt->pCatalog, pCxt->pComCxt->pRpc, pCxt->pComCxt->pEpSet, fullDbName, tableName, &pCxt->pTableMeta));
CHECK_CODE(catalogGetTableMeta(pCxt->pComCxt->pCatalog, pCxt->pComCxt->pRpc, pCxt->pComCxt->pEpSet, fullDbName, tableName, &pCxt->pTableMeta));
return TSDB_CODE_SUCCESS;
}
@ -209,6 +187,51 @@ static int32_t findCol(SToken* pColname, int32_t start, int32_t end, SSchema* pS
return -1;
}
static void fillMsgHeader(SVgDataBlocks* dst) {
SMsgDesc* desc = (SMsgDesc*)dst->pData;
desc->numOfVnodes = htonl(1);
SSubmitMsg* submit = (SSubmitMsg*)(desc + 1);
submit->header.vgId = htonl(dst->vgId);
submit->header.contLen = htonl(dst->size - sizeof(SMsgDesc));
submit->length = submit->header.contLen;
submit->numOfBlocks = htonl(dst->numOfTables);
SSubmitBlk* blk = (SSubmitBlk*)(submit + 1);
int32_t numOfBlocks = dst->numOfTables;
while (numOfBlocks--) {
int32_t dataLen = blk->dataLen;
blk->uid = htobe64(blk->uid);
blk->tid = htonl(blk->tid);
blk->padding = htonl(blk->padding);
blk->sversion = htonl(blk->sversion);
blk->dataLen = htonl(blk->dataLen);
blk->schemaLen = htonl(blk->schemaLen);
blk->numOfRows = htons(blk->numOfRows);
blk = (SSubmitBlk*)(blk->data + dataLen);
}
}
static int32_t buildOutput(SInsertParseContext* pCxt) {
size_t numOfVg = taosArrayGetSize(pCxt->pVgDataBlocks);
pCxt->pOutput->pDataBlocks = taosArrayInit(numOfVg, POINTER_BYTES);
if (NULL == pCxt->pOutput->pDataBlocks) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
for (size_t i = 0; i < numOfVg; ++i) {
STableDataBlocks* src = taosArrayGetP(pCxt->pVgDataBlocks, i);
SVgDataBlocks* dst = calloc(1, sizeof(SVgDataBlocks));
if (NULL == dst) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
dst->vgId = src->vgId;
dst->numOfTables = src->numOfTables;
dst->size = src->size;
SWAP(dst->pData, src->pData, char*);
fillMsgHeader(dst);
taosArrayPush(pCxt->pOutput->pDataBlocks, &dst);
}
return TSDB_CODE_SUCCESS;
}
static int32_t checkTimestamp(STableDataBlocks *pDataBlocks, const char *start) {
// once the data block is disordered, we do NOT keep previous timestamp any more
if (!pDataBlocks->ordered) {
@ -219,16 +242,14 @@ static int32_t checkTimestamp(STableDataBlocks *pDataBlocks, const char *start)
if (k == INT64_MIN) {
if (pDataBlocks->tsSource == TSDB_USE_CLI_TS) {
return -1;
} else if (pDataBlocks->tsSource == -1) {
pDataBlocks->tsSource = TSDB_USE_SERVER_TS;
return TSDB_CODE_FAILED; // client time/server time can not be mixed
}
pDataBlocks->tsSource = TSDB_USE_SERVER_TS;
} else {
if (pDataBlocks->tsSource == TSDB_USE_SERVER_TS) {
return -1; // client time/server time can not be mixed
} else if (pDataBlocks->tsSource == -1) {
pDataBlocks->tsSource = TSDB_USE_CLI_TS;
return TSDB_CODE_FAILED; // client time/server time can not be mixed
}
pDataBlocks->tsSource = TSDB_USE_CLI_TS;
}
if (k <= pDataBlocks->prevTS && (pDataBlocks->tsSource == TSDB_USE_CLI_TS)) {
@ -608,26 +629,24 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, SParsedDataColInfo*
}
// pSql -> tag1_value, ...)
static int32_t parseTagsClause(SInsertParseContext* pCxt, SParsedDataColInfo* pSpd, SSchema* pTagsSchema, uint8_t precision) {
SKVRowBuilder kvRowBuilder = {0};
if (tdInitKVRowBuilder(&kvRowBuilder) < 0) {
destroyBoundColumnInfo(pSpd);
static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pTagsSchema, uint8_t precision) {
if (tdInitKVRowBuilder(&pCxt->tagsBuilder) < 0) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
SKvParam param = {.builder = &kvRowBuilder};
SKvParam param = {.builder = &pCxt->tagsBuilder};
SToken sToken;
char tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0}; // used for deleting Escape character: \\, \', \"
for (int i = 0; i < pSpd->numOfBound; ++i) {
for (int i = 0; i < pCxt->tags.numOfBound; ++i) {
NEXT_TOKEN(pCxt->pSql, sToken);
SSchema* pSchema = &pTagsSchema[pSpd->boundedColumns[i]];
SSchema* pSchema = &pTagsSchema[pCxt->tags.boundedColumns[i]];
param.schema = pSchema;
CHECK_CODE_2(parseOneValue(pCxt, &sToken, pSchema, precision, tmpTokenBuf, KvRowAppend, &param), tdDestroyKVRowBuilder(&kvRowBuilder), destroyBoundColumnInfo(pSpd));
CHECK_CODE(parseOneValue(pCxt, &sToken, pSchema, precision, tmpTokenBuf, KvRowAppend, &param));
}
destroyBoundColumnInfo(pSpd);
SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
tdDestroyKVRowBuilder(&kvRowBuilder);
destroyBoundColumnInfo(&pCxt->tags);
SKVRow row = tdGetKVRowFromBuilder(&pCxt->tagsBuilder);
tdDestroyKVRowBuilder(&pCxt->tagsBuilder);
if (NULL == row) {
return buildInvalidOperationMsg(&pCxt->msg, "tag value expected");
}
@ -650,13 +669,12 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SToken* pTbnameToken)
}
SSchema* pTagsSchema = getTableTagSchema(pCxt->pTableMeta);
SParsedDataColInfo spd = {0};
setBoundColumnInfo(&spd, pTagsSchema, getNumOfTags(pCxt->pTableMeta));
setBoundColumnInfo(&pCxt->tags, pTagsSchema, getNumOfTags(pCxt->pTableMeta));
// pSql -> [(tag1_name, ...)] TAGS (tag1_value, ...)
NEXT_TOKEN(pCxt->pSql, sToken);
if (TK_LP == sToken.type) {
CHECK_CODE_1(parseBoundColumns(pCxt, &spd, pTagsSchema), destroyBoundColumnInfo(&spd));
CHECK_CODE(parseBoundColumns(pCxt, &pCxt->tags, pTagsSchema));
NEXT_TOKEN(pCxt->pSql, sToken);
}
@ -668,7 +686,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SToken* pTbnameToken)
if (TK_LP != sToken.type) {
return buildSyntaxErrMsg(&pCxt->msg, "( is expected", sToken.z);
}
CHECK_CODE(parseTagsClause(pCxt, &spd, pTagsSchema, getTableInfo(pCxt->pTableMeta).precision));
CHECK_CODE(parseTagsClause(pCxt, pTagsSchema, getTableInfo(pCxt->pTableMeta).precision));
return TSDB_CODE_SUCCESS;
}
@ -732,10 +750,12 @@ static int32_t parseValues(SInsertParseContext* pCxt, STableDataBlocks* pDataBlo
char tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0}; // used for deleting Escape character: \\, \', \"
SToken sToken;
while (1) {
NEXT_TOKEN(pCxt->pSql, sToken);
int32_t index = 0;
NEXT_TOKEN_KEEP_SQL(pCxt->pSql, sToken, index);
if (TK_LP != sToken.type) {
break;
}
pCxt->pSql += index;
if ((*numOfRows) >= maxRows || pDataBlock->size + extendedRowSize >= pDataBlock->nAllocSize) {
int32_t tSize;
@ -815,7 +835,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
if (TK_LP == sToken.type) {
// pSql -> field1_name, ...)
CHECK_CODE_1(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pCxt->pTableMeta)), destroyBoundColumnInfo(&dataBuf->boundColumnInfo));
CHECK_CODE(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pCxt->pTableMeta)));
NEXT_TOKEN(pCxt->pSql, sToken);
}
@ -842,9 +862,18 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
}
// merge according to vgId
if (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT) && taosHashGetSize(pCxt->pTableBlockHashObj) > 0) {
CHECK_CODE(mergeTableDataBlocks(pCxt->pTableBlockHashObj, pCxt->pOutput->schemaAttache, pCxt->pOutput->payloadType, true));
CHECK_CODE(mergeTableDataBlocks(pCxt->pTableBlockHashObj, pCxt->pOutput->schemaAttache, pCxt->pOutput->payloadType, &pCxt->pVgDataBlocks));
}
return TSDB_CODE_SUCCESS;
return buildOutput(pCxt);
}
static void destroyInsertParseContext(SInsertParseContext* pCxt) {
tfree(pCxt->pTableMeta);
destroyBoundColumnInfo(&pCxt->tags);
tdDestroyKVRowBuilder(&pCxt->tagsBuilder);
taosHashCleanup(pCxt->pTableBlockHashObj);
destroyBlockArrayList(pCxt->pTableDataBlocks);
destroyBlockArrayList(pCxt->pVgDataBlocks);
}
// INSERT INTO
@ -854,26 +883,30 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
// VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
// [...];
int32_t parseInsertSql(SParseContext* pContext, SInsertStmtInfo** pInfo) {
CHECK_CODE(createInsertStmtInfo(pInfo));
SInsertParseContext context = {
.pComCxt = pContext,
.pSql = pContext->pSql,
.msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
.pCatalog = NULL,
.pTableMeta = NULL,
.pTableBlockHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false),
.totalNum = 0,
.pOutput = *pInfo
.pOutput = calloc(1, sizeof(SInsertStmtInfo))
};
if (NULL == context.pTableBlockHashObj) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
if (NULL == context.pTableBlockHashObj || NULL == context.pOutput) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
return TSDB_CODE_FAILED;
}
CHECK_CODE(catalogGetHandle(pContext->pClusterId, &context.pCatalog));
CHECK_CODE(skipInsertInto(&context));
CHECK_CODE(parseInsertBody(&context));
*pInfo = context.pOutput;
context.pOutput->schemaAttache = pContext->schemaAttached;
context.pOutput->payloadType = PAYLOAD_TYPE_KV;
return TSDB_CODE_SUCCESS;
int32_t code = skipInsertInto(&context);
if (TSDB_CODE_SUCCESS == code) {
code = parseInsertBody(&context);
}
destroyInsertParseContext(&context);
terrno = code;
return (TSDB_CODE_SUCCESS == code ? TSDB_CODE_SUCCESS : TSDB_CODE_FAILED);
}

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

@ -17,5 +17,3 @@ TARGET_LINK_LIBRARIES(
parserTest
PUBLIC os util common parser catalog transport gtest function planner qcom
)
TARGET_LINK_OPTIONS(parserTest PRIVATE -Wl,-wrap,malloc)

182
source/libs/parser/test/insertParserTest.cpp Normal file
View File

@ -0,0 +1,182 @@
/*
* 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 <gtest/gtest.h>
#include "insertParser.h"
// #include "mockCatalog.h"
using namespace std;
using namespace testing;
namespace {
string toString(int32_t code) {
return tstrerror(code);
}
}
// syntax:
// INSERT INTO
// tb_name
// [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
// [(field1_name, ...)]
// VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
// [...];
class InsertTest : public Test {
protected:
void setDatabase(const string& acctId, const string& db) {
acctId_ = acctId;
db_ = db;
}
void bind(const char* sql) {
reset();
cxt_.pAcctId = acctId_.c_str();
cxt_.pDbname = db_.c_str();
strcpy(sqlBuf_, sql);
cxt_.sqlLen = strlen(sql);
sqlBuf_[cxt_.sqlLen] = '\0';
cxt_.pSql = sqlBuf_;
}
int32_t run() {
code_ = parseInsertSql(&cxt_, &res_);
if (code_ != TSDB_CODE_SUCCESS) {
cout << "code:" << toString(code_) << ", msg:" << errMagBuf_ << endl;
}
return code_;
}
SInsertStmtInfo* reslut() {
return res_;
}
void dumpReslut() {
size_t num = taosArrayGetSize(res_->pDataBlocks);
cout << "schemaAttache:" << (int32_t)res_->schemaAttache << ", payloadType:" << (int32_t)res_->payloadType << ", insertType:" << res_->insertType << ", numOfVgs:" << num << endl;
for (size_t i = 0; i < num; ++i) {
SVgDataBlocks* vg = (SVgDataBlocks*)taosArrayGetP(res_->pDataBlocks, i);
cout << "vgId:" << vg->vgId << ", numOfTables:" << vg->numOfTables << ", dataSize:" << vg->size << endl;
SMsgDesc* desc = (SMsgDesc*)(vg->pData);
cout << "numOfVnodes:" << ntohl(desc->numOfVnodes) << endl;
SSubmitMsg* submit = (SSubmitMsg*)(desc + 1);
cout << "length:" << ntohl(submit->length) << ", numOfBlocks:" << ntohl(submit->numOfBlocks) << endl;
int32_t numOfBlocks = ntohl(submit->numOfBlocks);
SSubmitBlk* blk = (SSubmitBlk*)(submit + 1);
for (int32_t i = 0; i < numOfBlocks; ++i) {
cout << "Block:" << i << endl;
cout << "\tuid:" << be64toh(blk->uid) << ", tid:" << ntohl(blk->tid) << ", padding:" << ntohl(blk->padding) << ", sversion:" << ntohl(blk->sversion)
<< ", dataLen:" << ntohl(blk->dataLen) << ", schemaLen:" << ntohl(blk->schemaLen) << ", numOfRows:" << ntohs(blk->numOfRows) << endl;
blk = (SSubmitBlk*)(blk->data + ntohl(blk->dataLen));
}
}
}
void checkReslut(int32_t numOfTables, int16_t numOfRows1, int16_t numOfRows2 = -1) {
ASSERT_EQ(res_->schemaAttache, 0);
ASSERT_EQ(res_->payloadType, PAYLOAD_TYPE_KV);
ASSERT_EQ(res_->insertType, TSDB_QUERY_TYPE_INSERT);
size_t num = taosArrayGetSize(res_->pDataBlocks);
ASSERT_GE(num, 0);
for (size_t i = 0; i < num; ++i) {
SVgDataBlocks* vg = (SVgDataBlocks*)taosArrayGetP(res_->pDataBlocks, i);
ASSERT_EQ(vg->numOfTables, numOfTables);
ASSERT_GE(vg->size, 0);
SMsgDesc* desc = (SMsgDesc*)(vg->pData);
ASSERT_EQ(ntohl(desc->numOfVnodes), 1);
SSubmitMsg* submit = (SSubmitMsg*)(desc + 1);
ASSERT_GE(ntohl(submit->length), 0);
ASSERT_GE(ntohl(submit->numOfBlocks), 0);
int32_t numOfBlocks = ntohl(submit->numOfBlocks);
SSubmitBlk* blk = (SSubmitBlk*)(submit + 1);
for (int32_t i = 0; i < numOfBlocks; ++i) {
ASSERT_EQ(ntohs(blk->numOfRows), (0 == i ? numOfRows1 : (numOfRows2 > 0 ? numOfRows2 : numOfRows1)));
blk = (SSubmitBlk*)(blk->data + ntohl(blk->dataLen));
}
}
}
private:
static const int max_err_len = 1024;
static const int max_sql_len = 1024 * 1024;
void reset() {
memset(&cxt_, 0, sizeof(cxt_));
memset(errMagBuf_, 0, max_err_len);
cxt_.pMsg = errMagBuf_;
cxt_.msgLen = max_err_len;
code_ = TSDB_CODE_SUCCESS;
res_ = nullptr;
}
string acctId_;
string db_;
char errMagBuf_[max_err_len];
char sqlBuf_[max_sql_len];
SParseContext cxt_;
int32_t code_;
SInsertStmtInfo* res_;
};
// INSERT INTO tb_name VALUES (field1_value, ...)
TEST_F(InsertTest, singleTableSingleRowTest) {
setDatabase("root", "test");
bind("insert into t1 values (now, 1, \"beijing\")");
ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
dumpReslut();
checkReslut(1, 1);
}
// INSERT INTO tb_name VALUES (field1_value, ...)(field1_value, ...)
TEST_F(InsertTest, singleTableMultiRowTest) {
setDatabase("root", "test");
bind("insert into t1 values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")");
ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
dumpReslut();
checkReslut(1, 3);
}
// INSERT INTO tb1_name VALUES (field1_value, ...) tb2_name VALUES (field1_value, ...)
TEST_F(InsertTest, multiTableSingleRowTest) {
setDatabase("root", "test");
bind("insert into st1s1 values (now, 1, \"beijing\") st1s2 values (now, 10, \"131028\")");
ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
dumpReslut();
checkReslut(2, 1);
}
// INSERT INTO tb1_name VALUES (field1_value, ...) tb2_name VALUES (field1_value, ...)
TEST_F(InsertTest, multiTableMultiRowTest) {
setDatabase("root", "test");
bind("insert into st1s1 values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")"
" st1s2 values (now, 10, \"131028\")(now+1s, 20, \"132028\")");
ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
dumpReslut();
checkReslut(2, 3, 2);
}
TEST_F(InsertTest, toleranceTest) {
setDatabase("root", "test");
bind("insert into");
ASSERT_NE(run(), TSDB_CODE_SUCCESS);
bind("insert into t");
ASSERT_NE(run(), TSDB_CODE_SUCCESS);
}

129
source/libs/parser/test/insertTest.cpp
View File

@ -1,129 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
#include "insertParser.h"
#include "mockCatalog.h"
using namespace std;
using namespace testing;
namespace {
string toString(int32_t code) {
return tstrerror(code);
}
}
extern "C" {
#include <execinfo.h>
void *__real_malloc(size_t);
void *__wrap_malloc(size_t c) {
// printf("My MALLOC called: %d\n", c);
// void *array[32];
// int size = backtrace(array, 32);
// char **symbols = backtrace_symbols(array, size);
// for (int i = 0; i < size; ++i) {
// cout << symbols[i] << endl;
// }
// free(symbols);
return __real_malloc(c);
}
}
// syntax:
// INSERT INTO
// tb_name
// [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
// [(field1_name, ...)]
// VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
// [...];
class InsertTest : public Test {
protected:
void setDatabase(const string& acctId, const string& db) {
acctId_ = acctId;
db_ = db;
}
void bind(const char* sql) {
reset();
cxt_.pAcctId = acctId_.c_str();
cxt_.pDbname = db_.c_str();
strcpy(sqlBuf_, sql);
cxt_.sqlLen = strlen(sql);
sqlBuf_[cxt_.sqlLen] = '\0';
cxt_.pSql = sqlBuf_;
}
int32_t run() {
code_ = parseInsertSql(&cxt_, &res_);
if (code_ != TSDB_CODE_SUCCESS) {
cout << "code:" << toString(code_) << ", msg:" << errMagBuf_ << endl;
}
return code_;
}
SInsertStmtInfo* reslut() {
return res_;
}
private:
static const int max_err_len = 1024;
static const int max_sql_len = 1024 * 1024;
void reset() {
memset(&cxt_, 0, sizeof(cxt_));
memset(errMagBuf_, 0, max_err_len);
cxt_.pMsg = errMagBuf_;
cxt_.msgLen = max_err_len;
code_ = TSDB_CODE_SUCCESS;
res_ = nullptr;
}
string acctId_;
string db_;
char errMagBuf_[max_err_len];
char sqlBuf_[max_sql_len];
SParseContext cxt_;
int32_t code_;
SInsertStmtInfo* res_;
};
// INSERT INTO tb_name VALUES (field1_value, ...)
TEST_F(InsertTest, simpleTest) {
setDatabase("root", "test");
bind("insert into t1 values (now, 1, \"wxy\")");
ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
SInsertStmtInfo* res = reslut();
// todo check
ASSERT_EQ(res->insertType, TSDB_QUERY_TYPE_INSERT);
// ASSERT_EQ(taosArrayGetSize(res->pDataBlocks), 1);
}
TEST_F(InsertTest, toleranceTest) {
setDatabase("root", "test");
bind("insert into");
ASSERT_NE(run(), TSDB_CODE_SUCCESS);
bind("insert into t");
ASSERT_NE(run(), TSDB_CODE_SUCCESS);
}

6
source/libs/parser/test/mockCatalog.cpp
View File

@ -25,15 +25,15 @@ namespace {
void generateTestT1(MockCatalogService* mcs) {
ITableBuilder& builder = mcs->createTableBuilder("root.test", "t1", TSDB_NORMAL_TABLE, 3)
.setPrecision(TSDB_TIME_PRECISION_MILLI).setVgid(1).addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP)
.addColumn("c1", TSDB_DATA_TYPE_INT).addColumn("c2", TSDB_DATA_TYPE_BINARY, 10);
.addColumn("c1", TSDB_DATA_TYPE_INT).addColumn("c2", TSDB_DATA_TYPE_BINARY, 20);
builder.done();
}
void generateTestST1(MockCatalogService* mcs) {
ITableBuilder& builder = mcs->createTableBuilder("root.test", "st1", TSDB_SUPER_TABLE, 3, 2)
.setPrecision(TSDB_TIME_PRECISION_MILLI).addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP)
.addTag("tag1", TSDB_DATA_TYPE_INT).addTag("tag2", TSDB_DATA_TYPE_BINARY, 10)
.addColumn("c1", TSDB_DATA_TYPE_INT).addColumn("c2", TSDB_DATA_TYPE_BINARY, 10);
.addTag("tag1", TSDB_DATA_TYPE_INT).addTag("tag2", TSDB_DATA_TYPE_BINARY, 20)
.addColumn("c1", TSDB_DATA_TYPE_INT).addColumn("c2", TSDB_DATA_TYPE_BINARY, 20);
builder.done();
mcs->createSubTable("root.test", "st1", "st1s1", 1);
mcs->createSubTable("root.test", "st1", "st1s2", 2);

4
source/libs/parser/test/mockCatalog.h
View File

@ -22,8 +22,4 @@ void initMetaDataEnv();
void generateMetaData();
void destroyMetaDataEnv();
// mock
// int32_t catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle);
// int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, STableMeta** pTableMeta);
#endif // MOCK_CATALOG_H

5
source/libs/parser/test/mockCatalogService.cpp
View File

@ -33,6 +33,7 @@ public:
col->colId = colId_++;
col->bytes = bytes;
strcpy(col->name, name.c_str());
rowsize_ += bytes;
return *this;
}
@ -147,11 +148,11 @@ public:
for (const auto& db : meta_) {
std::cout << "Databse:" << db.first << std::endl;
std::cout << SH("Table") << SH("Type") << SH("Precision") << IH("Vgid") << std::endl;
std::cout << SH("Table") << SH("Type") << SH("Precision") << IH("Vgid") << IH("RowSize") << std::endl;
std::cout << SL(3, 1) << std::endl;
for (const auto& table : db.second) {
const auto& schema = table.second->schema;
std::cout << SF(table.first) << SF(ttToString(schema->tableType)) << SF(pToString(schema->tableInfo.precision)) << IF(schema->vgId) << std::endl;
std::cout << SF(table.first) << SF(ttToString(schema->tableType)) << SF(pToString(schema->tableInfo.precision)) << IF(schema->vgId) << IF(schema->tableInfo.rowSize) << std::endl;
}
std::cout << std::endl;
}

2
source/libs/planner/src/physicalPlan.c
View File

@ -221,7 +221,7 @@ static SPhyNode* createPhyNode(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
node->pChildren = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
size_t size = taosArrayGetSize(pPlanNode->pChildren);
for(int32_t i = 0; i < size; ++i) {
SPhyNode* child = createPhyNode(pCxt, taosArrayGet(pPlanNode->pChildren, i));
SPhyNode* child = createPhyNode(pCxt, taosArrayGetP(pPlanNode->pChildren, i));
child->pParent = node;
taosArrayPush(node->pChildren, &child);
}

12
source/libs/qworker/CMakeLists.txt Normal file
View File

@ -0,0 +1,12 @@
aux_source_directory(src QWORKER_SRC)
add_library(qworker ${QWORKER_SRC})
target_include_directories(
qworker
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/qworker"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
qworker
PRIVATE os util transport planner qcom
)

68
source/libs/qworker/inc/qworkerInt.h Normal file
View File

@ -0,0 +1,68 @@
/*
* 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_QWORKER_INT_H_
#define _TD_QWORKER_INT_H_
#ifdef __cplusplus
extern "C" {
#endif
#define QWORKER_DEFAULT_SCHEDULER_NUMBER 10000
#define QWORKER_DEFAULT_RES_CACHE_NUMBER 10000
#define QWORKER_DEFAULT_SCH_TASK_NUMBER 10000
enum {
QW_READY_NOT_RECEIVED = 0,
QW_READY_RECEIVED,
QW_READY_RESPONSED,
};
typedef struct SQWorkerTaskStatus {
int8_t status;
int8_t ready;
} SQWorkerTaskStatus;
typedef struct SQWorkerResCache {
void *data;
} SQWorkerResCache;
typedef struct SQWorkerSchTaskStatus {
int32_t lastAccessTs; // timestamp in second
SHashObj *taskStatus; // key:queryId+taskId, value: SQWorkerTaskStatus
} SQWorkerSchTaskStatus;
// Qnode/Vnode level task management
typedef struct SQWorkerMgmt {
SQWorkerCfg cfg;
SHashObj *scheduleHash; //key: schedulerId, value: SQWorkerSchTaskStatus
SHashObj *resHash; //key: queryId+taskId, value: SQWorkerResCache
} SQWorkerMgmt;
#define QW_TASK_DONE(status) (status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == status == JOB_TASK_STATUS_CANCELLED)
#define QW_SET_QTID(id, qid, tid) do { *(uint64_t *)(id) = (qid); *(uint64_t *)((char *)(id) + sizeof(qid)) = (tid); } while (0)
#define QW_GET_QTID(id, qid, tid) do { (qid) = *(uint64_t *)(id); (tid) = *(uint64_t *)((char *)(id) + sizeof(qid)); } while (0)
#define QW_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define QW_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define QW_ERR_LRET(c,...) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { qError(__VA_ARGS__); terrno = _code; return _code; } } while (0)
#define QW_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
#ifdef __cplusplus
}
#endif
#endif /*_TD_QWORKER_INT_H_*/

298
source/libs/qworker/src/qworker.c Normal file
View File

@ -0,0 +1,298 @@
#include "taosmsg.h"
#include "query.h"
#include "qworker.h"
#include "qworkerInt.h"
#include "planner.h"
int32_t qwAddTaskStatus(SQWorkerMgmt *mgmt, uint64_t schedulerId, uint64_t queryId, uint64_t taskId, int8_t taskStatus) {
SQWorkerTaskStatus tStatus = {0};
tStatus.status = taskStatus;
char id[sizeof(queryId) + sizeof(taskId)] = {0};
QW_SET_QTID(id, queryId, taskId);
SQWorkerSchTaskStatus *schStatus = taosHashGet(mgmt->scheduleHash, &schedulerId, sizeof(schedulerId));
if (NULL == schStatus) {
SQWorkerSchTaskStatus newSchStatus = {0};
newSchStatus.taskStatus = taosHashInit(mgmt->cfg.maxSchTaskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
if (NULL == newSchStatus.taskStatus) {
qError("taosHashInit %d failed", mgmt->cfg.maxSchTaskNum);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
if (0 != taosHashPut(newSchStatus.taskStatus, id, sizeof(id), &tStatus, sizeof(tStatus))) {
qError("taosHashPut schedulerId[%"PRIx64"]queryId[%"PRIx64"] taskId[%"PRIx64"] to scheduleHash failed", schedulerId, queryId, taskId);
taosHashCleanup(newSchStatus.taskStatus);
return TSDB_CODE_QRY_APP_ERROR;
}
newSchStatus.lastAccessTs = taosGetTimestampSec();
if (0 != taosHashPut(mgmt->scheduleHash, &schedulerId, sizeof(schedulerId), &newSchStatus, sizeof(newSchStatus))) {
qError("taosHashPut schedulerId[%"PRIx64"] to scheduleHash failed", schedulerId);
taosHashCleanup(newSchStatus.taskStatus);
return TSDB_CODE_QRY_APP_ERROR;
}
return TSDB_CODE_SUCCESS;
}
schStatus->lastAccessTs = taosGetTimestampSec();
if (0 != taosHashPut(schStatus->taskStatus, id, sizeof(id), &tStatus, sizeof(tStatus))) {
qError("taosHashPut schedulerId[%"PRIx64"]queryId[%"PRIx64"] taskId[%"PRIx64"] to scheduleHash failed", schedulerId, queryId, taskId);
return TSDB_CODE_QRY_APP_ERROR;
}
return TSDB_CODE_SUCCESS;
}
int32_t qwGetTaskStatus(SQWorkerMgmt *mgmt, uint64_t schedulerId, uint64_t queryId, uint64_t taskId, SQWorkerTaskStatus **taskStatus) {
SQWorkerSchTaskStatus *schStatus = taosHashGet(mgmt->scheduleHash, &schedulerId, sizeof(schedulerId));
if (NULL == schStatus) {
qError("no scheduler for schedulerId[%"PRIx64"]", schedulerId);
return TSDB_CODE_QRY_APP_ERROR;
}
schStatus->lastAccessTs = taosGetTimestampSec();
char id[sizeof(queryId) + sizeof(taskId)] = {0};
QW_SET_QTID(id, queryId, taskId);
SQWorkerTaskStatus *tStatus = taosHashGet(schStatus->taskStatus, id, sizeof(id));
if (NULL == tStatus) {
qError("no task status for schedulerId[%"PRIx64"] queryId[%"PRIx64"] taskId[%"PRIx64"]", schedulerId, queryId, taskId);
return TSDB_CODE_QRY_APP_ERROR;
}
*taskStatus = tStatus;
return TSDB_CODE_SUCCESS;
}
int32_t qwAddTaskResult(SQWorkerMgmt *mgmt, uint64_t queryId, uint64_t taskId, void *data) {
char id[sizeof(queryId) + sizeof(taskId)] = {0};
QW_SET_QTID(id, queryId, taskId);
SQWorkerResCache resCache = {0};
resCache.data = data;
if (0 != taosHashPut(mgmt->resHash, id, sizeof(id), &resCache, sizeof(SQWorkerResCache))) {
qError("taosHashPut queryId[%"PRIx64"] taskId[%"PRIx64"] to resHash failed", queryId, taskId);
return TSDB_CODE_QRY_APP_ERROR;
}
return TSDB_CODE_SUCCESS;
}
int32_t qwGetTaskResult(SQWorkerMgmt *mgmt, uint64_t queryId, uint64_t taskId, void **data) {
char id[sizeof(queryId) + sizeof(taskId)] = {0};
QW_SET_QTID(id, queryId, taskId);
SQWorkerResCache *resCache = taosHashGet(mgmt->resHash, id, sizeof(id));
if (NULL == resCache) {
qError("no task res for queryId[%"PRIx64"] taskId[%"PRIx64"]", queryId, taskId);
return TSDB_CODE_QRY_APP_ERROR;
}
*data = resCache->data;
return TSDB_CODE_SUCCESS;
}
int32_t qwUpdateSchLastAccess(SQWorkerMgmt *mgmt, uint64_t schedulerId) {
SQWorkerSchTaskStatus *schStatus = taosHashGet(mgmt->scheduleHash, &schedulerId, sizeof(schedulerId));
if (NULL == schStatus) {
qError("no scheduler for schedulerId[%"PRIx64"]", schedulerId);
return TSDB_CODE_QRY_APP_ERROR;
}
schStatus->lastAccessTs = taosGetTimestampSec();
return TSDB_CODE_SUCCESS;
}
int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t schedulerId, SSchedulerStatusRsp **rsp) {
SQWorkerSchTaskStatus *schStatus = taosHashGet(mgmt->scheduleHash, &schedulerId, sizeof(schedulerId));
if (NULL == schStatus) {
qError("no scheduler for schedulerId[%"PRIx64"]", schedulerId);
return TSDB_CODE_QRY_APP_ERROR;
}
schStatus->lastAccessTs = taosGetTimestampSec();
int32_t i = 0;
int32_t taskNum = taosHashGetSize(schStatus->taskStatus);
int32_t size = sizeof(SSchedulerStatusRsp) + sizeof((*rsp)->status[0]) * taskNum;
*rsp = calloc(1, size);
if (NULL == *rsp) {
qError("calloc %d failed", size);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
void *key = NULL;
size_t keyLen = 0;
void *pIter = taosHashIterate(schStatus->taskStatus, NULL);
while (pIter) {
SQWorkerTaskStatus *taskStatus = (SQWorkerTaskStatus *)pIter;
taosHashGetKey(pIter, &key, &keyLen);
QW_GET_QTID(key, (*rsp)->status[i].queryId, (*rsp)->status[i].taskId);
(*rsp)->status[i].status = taskStatus->status;
pIter = taosHashIterate(schStatus->taskStatus, pIter);
}
(*rsp)->num = taskNum;
return TSDB_CODE_SUCCESS;
}
int32_t qwBuildRspMsg(void *data, int32_t msgType);
int32_t qWorkerInit(SQWorkerCfg *cfg, void **qWorkerMgmt) {
SQWorkerMgmt *mgmt = calloc(1, sizeof(SQWorkerMgmt));
if (NULL == mgmt) {
qError("calloc %d failed", (int32_t)sizeof(SQWorkerMgmt));
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
if (cfg) {
mgmt->cfg = *cfg;
} else {
mgmt->cfg.maxSchedulerNum = QWORKER_DEFAULT_SCHEDULER_NUMBER;
mgmt->cfg.maxResCacheNum = QWORKER_DEFAULT_RES_CACHE_NUMBER;
mgmt->cfg.maxSchTaskNum = QWORKER_DEFAULT_SCH_TASK_NUMBER;
}
mgmt->scheduleHash = taosHashInit(mgmt->cfg.maxSchedulerNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == mgmt->scheduleHash) {
tfree(mgmt);
QW_ERR_LRET(TSDB_CODE_QRY_OUT_OF_MEMORY, "init %d schduler hash failed", mgmt->cfg.maxSchedulerNum);
}
mgmt->resHash = taosHashInit(mgmt->cfg.maxResCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
if (NULL == mgmt->resHash) {
taosHashCleanup(mgmt->scheduleHash);
mgmt->scheduleHash = NULL;
tfree(mgmt);
QW_ERR_LRET(TSDB_CODE_QRY_OUT_OF_MEMORY, "init %d res cache hash failed", mgmt->cfg.maxResCacheNum);
}
*qWorkerMgmt = mgmt;
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerProcessQueryMsg(void *qWorkerMgmt, SSchedulerQueryMsg *msg, SRpcMsg *rsp) {
if (NULL == qWorkerMgmt || NULL == msg || NULL == rsp) {
return TSDB_CODE_QRY_INVALID_INPUT;
}
SSubplan *plan = NULL;
SQWorkerTaskStatus *tStatus = NULL;
int32_t code = qStringToSubplan(msg->msg, &plan);
if (TSDB_CODE_SUCCESS != code) {
qError("schId:%"PRIx64",qId:%"PRIx64",taskId:%"PRIx64" string to subplan failed, code:%d", msg->schedulerId, msg->queryId, msg->taskId, code);
return code;
}
//TODO call executer to init subquery
QW_ERR_JRET(qwAddTaskStatus(qWorkerMgmt, msg->schedulerId, msg->queryId, msg->taskId, JOB_TASK_STATUS_EXECUTING));
QW_ERR_JRET(qwBuildRspMsg(NULL, TSDB_MSG_TYPE_QUERY_RSP));
//TODO call executer to execute subquery
code = 0;
void *data = NULL;
//TODO call executer to execute subquery
QW_ERR_JRET(qwGetTaskStatus(qWorkerMgmt, msg->schedulerId, msg->queryId, msg->taskId, &tStatus));
tStatus->status = (code) ? JOB_TASK_STATUS_FAILED : JOB_TASK_STATUS_SUCCEED;
QW_ERR_JRET(qwAddTaskResult(qWorkerMgmt, msg->queryId, msg->taskId, data));
_return:
if (tStatus && QW_TASK_DONE(tStatus->status) && QW_READY_RECEIVED == tStatus->ready) {
QW_ERR_RET(qwBuildRspMsg(NULL, TSDB_MSG_TYPE_RES_READY_RSP));
}
qDestroySubplan(plan);
return code;
}
int32_t qWorkerProcessReadyMsg(void *qWorkerMgmt, SSchedulerReadyMsg *msg, SRpcMsg *rsp){
if (NULL == qWorkerMgmt || NULL == msg || NULL == rsp) {
return TSDB_CODE_QRY_INVALID_INPUT;
}
SQWorkerTaskStatus *tStatus = NULL;
QW_ERR_RET(qwGetTaskStatus(qWorkerMgmt, msg->schedulerId, msg->queryId, msg->taskId, &tStatus));
if (QW_TASK_DONE(tStatus->status)) {
QW_ERR_RET(qwBuildRspMsg(tStatus, TSDB_MSG_TYPE_RES_READY_RSP));
} else {
tStatus->ready = QW_READY_RECEIVED;
return TSDB_CODE_SUCCESS;
}
tStatus->ready = QW_READY_RESPONSED;
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerProcessStatusMsg(void *qWorkerMgmt, SSchedulerStatusMsg *msg, SRpcMsg *rsp) {
if (NULL == qWorkerMgmt || NULL == msg || NULL == rsp) {
return TSDB_CODE_QRY_INVALID_INPUT;
}
SSchedulerStatusRsp *sStatus = NULL;
QW_ERR_RET(qwGetSchTasksStatus(qWorkerMgmt, msg->schedulerId, &sStatus));
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerProcessFetchMsg(void *qWorkerMgmt, SSchedulerFetchMsg *msg, SRpcMsg *rsp) {
if (NULL == qWorkerMgmt || NULL == msg || NULL == rsp) {
return TSDB_CODE_QRY_INVALID_INPUT;
}
QW_ERR_RET(qwUpdateSchLastAccess(qWorkerMgmt, msg->schedulerId));
void *data = NULL;
QW_ERR_RET(qwGetTaskResult(qWorkerMgmt, msg->queryId, msg->taskId, &data));
QW_ERR_RET(qwBuildRspMsg(data, TSDB_MSG_TYPE_FETCH_RSP));
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerProcessCancelMsg(void *qWorkerMgmt, SSchedulerCancelMsg *msg, SRpcMsg *rsp);
void qWorkerDestroy(void **qWorkerMgmt) {
if (NULL == qWorkerMgmt || NULL == *qWorkerMgmt) {
return;
}
SQWorkerMgmt *mgmt = *qWorkerMgmt;
//TODO STOP ALL QUERY
//TODO FREE ALL
tfree(*qWorkerMgmt);
}

10
source/libs/scheduler/inc/schedulerInt.h
View File

@ -31,17 +31,9 @@ extern "C" {
#define SCH_MAX_CONDIDATE_EP_NUM TSDB_MAX_REPLICA
enum {
SCH_STATUS_NOT_START = 1,
SCH_STATUS_EXECUTING,
SCH_STATUS_SUCCEED,
SCH_STATUS_FAILED,
SCH_STATUS_CANCELLING,
SCH_STATUS_CANCELLED
};
typedef struct SSchedulerMgmt {
uint64_t taskId;
uint64_t schedulerId;
SSchedulerCfg cfg;
SHashObj *Jobs; // key: queryId, value: SQueryJob*
} SSchedulerMgmt;

58
source/libs/scheduler/src/scheduler.c
View File

@ -161,7 +161,7 @@ int32_t schValidateAndBuildJob(SQueryDag *dag, SQueryJob *job) {
SArray *levelPlans = NULL;
int32_t levelPlanNum = 0;
level.status = SCH_STATUS_NOT_START;
level.status = JOB_TASK_STATUS_NOT_START;
for (int32_t i = 0; i < levelNum; ++i) {
level.level = i;
@ -191,7 +191,7 @@ int32_t schValidateAndBuildJob(SQueryDag *dag, SQueryJob *job) {
task.taskId = atomic_add_fetch_64(&schMgmt.taskId, 1);
task.plan = plan;
task.status = SCH_STATUS_NOT_START;
task.status = JOB_TASK_STATUS_NOT_START;
void *p = taosArrayPush(level.subTasks, &task);
if (NULL == p) {
@ -304,13 +304,15 @@ int32_t schAsyncSendMsg(SQueryJob *job, SQueryTask *task, int32_t msgType) {
}
SSchedulerQueryMsg *pMsg = msg;
pMsg->queryId = job->queryId;
pMsg->taskId = task->taskId;
pMsg->contentLen = len;
pMsg->schedulerId = htobe64(schMgmt.schedulerId);
pMsg->queryId = htobe64(job->queryId);
pMsg->taskId = htobe64(task->taskId);
pMsg->contentLen = htonl(len);
memcpy(pMsg->msg, task->msg, len);
break;
}
case TSDB_MSG_TYPE_RSP_READY: {
case TSDB_MSG_TYPE_RES_READY: {
msgSize = sizeof(SSchedulerReadyMsg);
msg = calloc(1, msgSize);
if (NULL == msg) {
@ -319,8 +321,8 @@ int32_t schAsyncSendMsg(SQueryJob *job, SQueryTask *task, int32_t msgType) {
}
SSchedulerReadyMsg *pMsg = msg;
pMsg->queryId = job->queryId;
pMsg->taskId = task->taskId;
pMsg->queryId = htobe64(job->queryId);
pMsg->taskId = htobe64(task->taskId);
break;
}
case TSDB_MSG_TYPE_FETCH: {
@ -332,8 +334,8 @@ int32_t schAsyncSendMsg(SQueryJob *job, SQueryTask *task, int32_t msgType) {
}
SSchedulerFetchMsg *pMsg = msg;
pMsg->queryId = job->queryId;
pMsg->taskId = task->taskId;
pMsg->queryId = htobe64(job->queryId);
pMsg->taskId = htobe64(task->taskId);
break;
}
default:
@ -376,7 +378,7 @@ _return:
int32_t schProcessOnJobSuccess(SQueryJob *job) {
job->status = SCH_STATUS_SUCCEED;
job->status = JOB_TASK_STATUS_SUCCEED;
if (job->userFetch) {
SCH_ERR_RET(schFetchFromRemote(job));
@ -386,7 +388,7 @@ int32_t schProcessOnJobSuccess(SQueryJob *job) {
}
int32_t schProcessOnJobFailure(SQueryJob *job) {
job->status = SCH_STATUS_FAILED;
job->status = JOB_TASK_STATUS_FAILED;
atomic_val_compare_exchange_32(&job->remoteFetch, 1, 0);
@ -413,7 +415,7 @@ int32_t schProcessOnTaskSuccess(SQueryJob *job, SQueryTask *task) {
return TSDB_CODE_SUCCESS;
}
task->status = SCH_STATUS_SUCCEED;
task->status = JOB_TASK_STATUS_SUCCEED;
int32_t parentNum = (int32_t)taosArrayGetSize(task->parents);
if (parentNum == 0) {
@ -459,7 +461,7 @@ int32_t schProcessOnTaskFailure(SQueryJob *job, SQueryTask *task, int32_t errCod
if (!needRetry) {
SCH_TASK_ERR_LOG("task failed[%x], no more retry", errCode);
job->status = SCH_STATUS_FAILED;
job->status = JOB_TASK_STATUS_FAILED;
SCH_ERR_RET(schProcessOnJobFailure(job));
return TSDB_CODE_SUCCESS;
@ -478,13 +480,13 @@ int32_t schHandleRspMsg(SQueryJob *job, SQueryTask *task, int32_t msgType, int32
if (rspCode != TSDB_CODE_SUCCESS) {
SCH_ERR_JRET(schProcessOnTaskFailure(job, task, rspCode));
} else {
code = schAsyncSendMsg(job, task, TSDB_MSG_TYPE_RSP_READY);
code = schAsyncSendMsg(job, task, TSDB_MSG_TYPE_RES_READY);
if (code) {
goto _task_error;
}
}
break;
case TSDB_MSG_TYPE_RSP_READY:
case TSDB_MSG_TYPE_RES_READY:
if (rspCode != TSDB_CODE_SUCCESS) {
SCH_ERR_JRET(schProcessOnTaskFailure(job, task, rspCode));
} else {
@ -534,7 +536,7 @@ int32_t schLaunchTask(SQueryJob *job, SQueryTask *task) {
SCH_ERR_RET(schPushTaskToExecList(job, task));
task->status = SCH_STATUS_EXECUTING;
task->status = JOB_TASK_STATUS_EXECUTING;
return TSDB_CODE_SUCCESS;
}
@ -546,22 +548,26 @@ int32_t schLaunchJob(SQueryJob *job) {
SCH_ERR_RET(schLaunchTask(job, task));
}
job->status = SCH_STATUS_EXECUTING;
job->status = JOB_TASK_STATUS_EXECUTING;
return TSDB_CODE_SUCCESS;
}
int32_t schedulerInit(SSchedulerCfg *cfg) {
schMgmt.Jobs = taosHashInit(SCHEDULE_DEFAULT_JOB_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == schMgmt.Jobs) {
SCH_ERR_LRET(TSDB_CODE_QRY_OUT_OF_MEMORY, "init %d schduler jobs failed", SCHEDULE_DEFAULT_JOB_NUMBER);
}
if (cfg) {
schMgmt.cfg = *cfg;
} else {
schMgmt.cfg.maxJobNum = SCHEDULE_DEFAULT_JOB_NUMBER;
}
schMgmt.Jobs = taosHashInit(schMgmt.cfg.maxJobNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
if (NULL == schMgmt.Jobs) {
SCH_ERR_LRET(TSDB_CODE_QRY_OUT_OF_MEMORY, "init %d schduler jobs failed", schMgmt.cfg.maxJobNum);
}
schMgmt.schedulerId = 1; //TODO GENERATE A UUID
return TSDB_CODE_SUCCESS;
}
@ -605,7 +611,7 @@ int32_t scheduleExecJob(void *transport, SArray *qnodeList, SQueryDag* pDag, voi
SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR);
}
job->status = SCH_STATUS_NOT_START;
job->status = JOB_TASK_STATUS_NOT_START;
SCH_ERR_JRET(schLaunchJob(job));
@ -634,7 +640,7 @@ int32_t scheduleFetchRows(void *pJob, void **data) {
return TSDB_CODE_QRY_APP_ERROR;
}
if (job->status == SCH_STATUS_SUCCEED) {
if (job->status == JOB_TASK_STATUS_SUCCEED) {
SCH_ERR_JRET(schFetchFromRemote(job));
}
@ -668,7 +674,7 @@ void scheduleFreeJob(void *pJob) {
return;
}
if (job->status == SCH_STATUS_EXECUTING) {
if (job->status == JOB_TASK_STATUS_EXECUTING) {
scheduleCancelJob(pJob);
}
}

11
source/util/src/thash.c
View File

@ -776,9 +776,16 @@ size_t taosHashGetMemSize(const SHashObj *pHashObj) {
return (pHashObj->capacity * (sizeof(SHashEntry) + POINTER_BYTES)) + sizeof(SHashNode) * taosHashGetSize(pHashObj) + sizeof(SHashObj);
}
FORCE_INLINE void *taosHashGetDataKey(SHashObj *pHashObj, void *data) {
FORCE_INLINE int32_t taosHashGetKey(void *data, void** key, size_t* keyLen) {
if (NULL == data || NULL == key) {
return -1;
}
SHashNode * node = GET_HASH_PNODE(data);
return GET_HASH_NODE_KEY(node);
*key = GET_HASH_NODE_KEY(node);
*keyLen = node->keyLen;
return 0;
}
FORCE_INLINE uint32_t taosHashGetDataKeyLen(SHashObj *pHashObj, void *data) {