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

Merge pull request #14611 from taosdata/feat/idxFix1 

enh: refactor idx code
This commit is contained in:
Yihao Deng 2022-07-07 09:21:06 +08:00 committed by GitHub
parent 0feb11bdf0 7006eefda1
commit 31655c07d7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 518 additions and 473 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -127,7 +127,7 @@ int indexSearch(SIndex* index, SIndexMultiTermQuery* query, SArray* result);
* @parma opt (input, rebuild index opts)
* @return error code
*/
int indexRebuild(SIndex* index, SIndexOpts* opt);
// int indexRebuild(SIndex* index, SIndexOpts* opt);
/*
* open index
@ -185,6 +185,25 @@ SIndexTerm* indexTermCreate(int64_t suid, SIndexOperOnColumn operType, uint8_t c
int32_t nColName, const char* colVal, int32_t nColVal);
void indexTermDestroy(SIndexTerm* p);
/*
* rebuild index
*/
void indexRebuild(SIndexJson* idx, void* iter);
/*
* check index json status
**/
bool indexIsRebuild(SIndex* idx);
/*
* rebuild index json
*/
void indexJsonRebuild(SIndexJson* idx, void* iter);
/*
* check index json status
**/
bool indexJsonIsRebuild(SIndexJson* idx);
/*
* init index env
*
@ -203,7 +222,7 @@ typedef enum { SFLT_NOT_INDEX, SFLT_COARSE_INDEX, SFLT_ACCURATE_INDEX } SIdxFltS
SIdxFltStatus idxGetFltStatus(SNode* pFilterNode);
int32_t doFilterTag(const SNode* pFilterNode, SIndexMetaArg* metaArg, SArray* result);
int32_t doFilterTag(SNode* pFilterNode, SIndexMetaArg* metaArg, SArray* result, SIdxFltStatus* status);
/*
* destory index env
*

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

@ -1298,16 +1298,16 @@ void doProcessMsgFromServer(SSchedMsg* schedMsg) {
pSendInfo->fp(pSendInfo->param, &buf, pMsg->code);
rpcFreeCont(pMsg->pCont);
destroySendMsgInfo(pSendInfo);
taosMemoryFree(arg);
}
void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
SSchedMsg schedMsg = {0};
SEpSet* tEpSet = pEpSet != NULL ? taosMemoryCalloc(1, sizeof(SEpSet)) : NULL;
if (tEpSet != NULL) {
*tEpSet = *pEpSet;
SEpSet* tEpSet = NULL;
if (pEpSet != NULL) {
tEpSet = taosMemoryCalloc(1, sizeof(SEpSet));
memcpy((void*)tEpSet, (void*)pEpSet, sizeof(SEpSet));
}
SchedArg* arg = taosMemoryCalloc(1, sizeof(SchedArg));

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

@ -136,6 +136,8 @@ SArray *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdLis
int32_t tsdbReaderReset(STsdbReader *pReader, SQueryTableDataCond *pCond, int32_t tWinIdx);
int32_t tsdbGetFileBlocksDistInfo(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
int64_t tsdbGetNumOfRowsInMemTable(STsdbReader *pHandle);
void * tsdbGetIdx(SMeta *pMeta);
void * tsdbGetIvtIdx(SMeta *pMeta);
int32_t tsdbLastRowReaderOpen(void *pVnode, int32_t type, SArray *pTableIdList, int32_t *colId, int32_t numOfCols,
void **pReader);

12
source/dnode/vnode/src/meta/metaTable.c
View File

@ -74,7 +74,7 @@ static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const
SIndexTerm *term = NULL;
if (type == TSDB_DATA_TYPE_NULL) {
// handle null value
term = indexTermCreate(suid, ADD_VALUE, TSDB_DATA_TYPE_VARCHAR, key, nKey, NULL, 0);
} else if (type == TSDB_DATA_TYPE_NCHAR) {
if (pTagVal->nData > 0) {
char * val = taosMemoryCalloc(1, pTagVal->nData + VARSTR_HEADER_SIZE);
@ -83,17 +83,15 @@ static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const
type = TSDB_DATA_TYPE_VARCHAR;
term = indexTermCreate(suid, ADD_VALUE, type, key, nKey, val, len);
} else if (pTagVal->nData == 0) {
char * val = NULL;
int32_t len = 0;
// handle NULL key
term = indexTermCreate(suid, ADD_VALUE, TSDB_DATA_TYPE_VARCHAR, key, nKey, pTagVal->pData, 0);
}
} else if (type == TSDB_DATA_TYPE_DOUBLE) {
double val = *(double *)(&pTagVal->i64);
int len = 0;
int len = sizeof(val);
term = indexTermCreate(suid, ADD_VALUE, type, key, nKey, (const char *)&val, len);
} else if (type == TSDB_DATA_TYPE_BOOL) {
int val = *(int *)(&pTagVal->i64);
int len = 0;
int len = sizeof(val);
term = indexTermCreate(suid, ADD_VALUE, TSDB_DATA_TYPE_INT, key, nKey, (const char *)&val, len);
}
if (term != NULL) {
@ -419,7 +417,6 @@ static int metaDeleteTtlIdx(SMeta *pMeta, const SMetaEntry *pME) {
return tdbTbDelete(pMeta->pTtlIdx, &ttlKey, sizeof(ttlKey), &pMeta->txn);
}
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
void * pData = NULL;
int nData = 0;
@ -442,7 +439,6 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
tdbTbDelete(pMeta->pUidIdx, &uid, sizeof(uid), &pMeta->txn);
if (e.type != TSDB_SUPER_TABLE) metaDeleteTtlIdx(pMeta, &e);
if (e.type == TSDB_CHILD_TABLE) {
tdbTbDelete(pMeta->pCtbIdx, &(SCtbIdxKey){.suid = e.ctbEntry.suid, .uid = uid}, sizeof(SCtbIdxKey), &pMeta->txn);
} else if (e.type == TSDB_NORMAL_TABLE) {

55
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -70,7 +70,8 @@ typedef struct SFilesetIter {
} SFilesetIter;
typedef struct SFileDataBlockInfo {
int32_t tbBlockIdx; // index position in STableBlockScanInfo in order to check whether neighbor block overlaps with it
int32_t
tbBlockIdx; // index position in STableBlockScanInfo in order to check whether neighbor block overlaps with it
uint64_t uid;
} SFileDataBlockInfo;
@ -144,17 +145,21 @@ static int buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, i
static TSDBROW* getValidRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pReader);
static int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader,
SRowMerger* pMerger);
static int32_t doMergeRowsInBuf(SIterInfo *pIter, int64_t ts, SArray* pDelList, SRowMerger* pMerger, STsdbReader* pReader);
static int32_t doMergeRowsInBuf(SIterInfo* pIter, int64_t ts, SArray* pDelList, SRowMerger* pMerger,
STsdbReader* pReader);
static int32_t doAppendOneRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow);
static void setComposedBlockFlag(STsdbReader* pReader, bool composed);
static void updateSchema(TSDBROW* pRow, uint64_t uid, STsdbReader* pReader);
static bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey);
static void doMergeMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo *pIter, SArray* pDelList, STSRow** pTSRow, STsdbReader* pReader);
static void doMergeMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, STSRow** pTSRow,
STsdbReader* pReader);
static void doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader,
STSRow** pTSRow);
static int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData, STbData* piMemTbData);
static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions, const char* idstr, int8_t *pLevel);
static int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData,
STbData* piMemTbData);
static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions, const char* idstr,
int8_t* pLevel);
static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level);
static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) {
@ -390,7 +395,8 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
initReaderStatus(&pReader->status);
pReader->pTsdb = getTsdbByRetentions(pVnode, pCond->twindows[0].skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
pReader->pTsdb =
getTsdbByRetentions(pVnode, pCond->twindows[0].skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
pReader->suid = pCond->suid;
pReader->order = pCond->order;
pReader->capacity = 4096;
@ -1936,8 +1942,8 @@ static int32_t doMergeThreeLevelRows(STsdbReader* pReader, STableBlockScanInfo*
ASSERT(0);
}
static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDumpInfo, STableBlockScanInfo* pBlockScanInfo,
STsdbReader* pReader) {
static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDumpInfo,
STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
// check for version and time range
int64_t ver = pBlockData->aVersion[pDumpInfo->rowIndex];
if (ver > pReader->verRange.maxVer || ver < pReader->verRange.minVer) {
@ -2113,7 +2119,8 @@ static int32_t initMemDataIterator(STableBlockScanInfo* pBlockScanInfo, STsdbRea
return TSDB_CODE_SUCCESS;
}
int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData, STbData* piMemTbData) {
int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData,
STbData* piMemTbData) {
if (pBlockScanInfo->delSkyline != NULL) {
return TSDB_CODE_SUCCESS;
}
@ -2173,7 +2180,8 @@ int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader*
}
taosArrayDestroy(pDelData);
pBlockScanInfo->iter.index = ASCENDING_TRAVERSE(pReader->order)? 0:taosArrayGetSize(pBlockScanInfo->delSkyline) - 1;
pBlockScanInfo->iter.index =
ASCENDING_TRAVERSE(pReader->order) ? 0 : taosArrayGetSize(pBlockScanInfo->delSkyline) - 1;
pBlockScanInfo->iiter.index = pBlockScanInfo->iter.index;
pBlockScanInfo->fileDelIndex = pBlockScanInfo->iter.index;
return code;
@ -2527,7 +2535,8 @@ TSDBROW* getValidRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pRea
}
// it is a valid data version
if ((key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer) && (!hasBeenDropped(pDelList, &pIter->index, &key))) {
if ((key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer) &&
(!hasBeenDropped(pDelList, &pIter->index, &key))) {
return pRow;
}
@ -2545,7 +2554,8 @@ TSDBROW* getValidRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pRea
return NULL;
}
if (key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer && (!hasBeenDropped(pDelList, &pIter->index, &key))) {
if (key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer &&
(!hasBeenDropped(pDelList, &pIter->index, &key))) {
return pRow;
}
}
@ -2685,7 +2695,8 @@ void updateSchema(TSDBROW* pRow, uint64_t uid, STsdbReader* pReader) {
}
}
void doMergeMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo *pIter, SArray* pDelList, STSRow** pTSRow, STsdbReader* pReader) {
void doMergeMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, STSRow** pTSRow,
STsdbReader* pReader) {
SRowMerger merge = {0};
TSDBKEY k = TSDBROW_KEY(pRow);
@ -2855,6 +2866,20 @@ int32_t tsdbSetTableId(STsdbReader* pReader, int64_t uid) {
return TDB_CODE_SUCCESS;
}
void* tsdbGetIdx(SMeta* pMeta) {
if (pMeta == NULL) {
return NULL;
}
return metaGetIdx(pMeta);
}
void* tsdbGetIvtIdx(SMeta* pMeta) {
if (pMeta == NULL) {
return NULL;
}
return metaGetIvtIdx(pMeta);
}
/**
* @brief Get all suids since suid
*
@ -3319,8 +3344,8 @@ int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond, int32_
}
}
tsdbDebug("%p reset reader, suid:%"PRIu64", numOfTables:%d, query range:%"PRId64" - %"PRId64" in query %s", pReader, pReader->suid,
numOfTables, pReader->window.skey, pReader->window.ekey, pReader->idStr);
tsdbDebug("%p reset reader, suid:%" PRIu64 ", numOfTables:%d, query range:%" PRId64 " - %" PRId64 " in query %s",
pReader, pReader->suid, numOfTables, pReader->window.skey, pReader->window.ekey, pReader->idStr);
return code;
}

14
source/libs/executor/src/executil.c
View File

@ -305,9 +305,21 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode,
SNode* pTagIndexCond = (SNode*)pListInfo->pTagIndexCond;
if (pScanNode->tableType == TSDB_SUPER_TABLE) {
if (pTagIndexCond) {
///<<<<<<< HEAD
SIndexMetaArg metaArg = {
.metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid};
SArray* res = taosArrayInit(8, sizeof(uint64_t));
// code = doFilterTag(pTagIndexCond, &metaArg, res);
SIdxFltStatus status = SFLT_NOT_INDEX;
code = doFilterTag(pTagIndexCond, &metaArg, res, &status);
if (code != 0 || status == SFLT_NOT_INDEX) {
code = TSDB_CODE_INDEX_REBUILDING;
}
//=======
// SArray* res = taosArrayInit(8, sizeof(uint64_t));
// // code = doFilterTag(pTagIndexCond, &metaArg, res);
// code = TSDB_CODE_INDEX_REBUILDING;
//>>>>>>> dvv
if (code == TSDB_CODE_INDEX_REBUILDING) {
code = vnodeGetAllTableList(pVnode, tableUid, pListInfo->pTableList);
} else if (code != TSDB_CODE_SUCCESS) {

6
source/libs/index/CMakeLists.txt
View File

@ -35,7 +35,7 @@ if (${BUILD_WITH_INVERTEDINDEX})
endif(${BUILD_WITH_INVERTEDINDEX})
# if (${BUILD_TEST})
# add_subdirectory(test)
# endif(${BUILD_TEST})
if (${BUILD_TEST})
add_subdirectory(test)
endif(${BUILD_TEST})

13
source/libs/index/inc/indexFst.h
View File

@ -53,7 +53,7 @@ typedef struct FstRange {
} FstRange;
typedef enum { OneTransNext, OneTrans, AnyTrans, EmptyFinal } State;
typedef enum { Ordered, OutOfOrdered, DuplicateKey } OrderType;
typedef enum { Ordered, OutOfOrdered, DuplicateKey } FstOrderType;
FstBoundWithData* fstBoundStateCreate(FstBound type, FstSlice* data);
bool fstBoundWithDataExceededBy(FstBoundWithData* bound, FstSlice* slice);
@ -106,7 +106,7 @@ bool fstBuilderInsert(FstBuilder* b, FstSlice bs, Output in);
void fstBuilderCompileFrom(FstBuilder* b, uint64_t istate);
void* fstBuilerIntoInner(FstBuilder* b);
void fstBuilderFinish(FstBuilder* b);
OrderType fstBuilderCheckLastKey(FstBuilder* b, FstSlice bs, bool ckDup);
FstOrderType fstBuilderCheckLastKey(FstBuilder* b, FstSlice bs, bool ckDup);
CompiledAddr fstBuilderCompile(FstBuilder* b, FstBuilderNode* bn);
typedef struct FstTransitions {
@ -213,14 +213,18 @@ typedef struct FstNode {
// If this node is final and has a terminal output value, then it is, returned.
// Otherwise, a zero output is returned
#define FST_NODE_FINAL_OUTPUT(node) node->finalOutput
// Returns true if and only if this node corresponds to a final or "match",
// state in the finite state transducer.
#define FST_NODE_IS_FINAL(node) node->isFinal
// Returns the number of transitions in this node, The maximum number of
// transitions is 256.
#define FST_NODE_LEN(node) node->nTrans
// Returns true if and only if this node has zero transitions.
#define FST_NODE_IS_EMPTYE(node) (node->nTrans == 0)
// Return the address of this node.
#define FST_NODE_ADDR(node) node->start
@ -277,6 +281,8 @@ FStmBuilder* fstSearch(Fst* fst, FAutoCtx* ctx);
FStmStBuilder* fstSearchWithState(Fst* fst, FAutoCtx* ctx);
// into stream to expand later
//
FStmSt* stmBuilderIntoStm(FStmBuilder* sb);
bool fstVerify(Fst* fst);
@ -325,7 +331,8 @@ FStmBuilder* stmBuilderCreate(Fst* fst, FAutoCtx* aut);
void stmBuilderDestroy(FStmBuilder* b);
// set up bound range
// refator later: to simple code by marco
// refator later
// simple code by marco
void stmBuilderSetRange(FStmBuilder* b, FstSlice* val, RangeType type);
#ifdef __cplusplus

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

@ -45,6 +45,7 @@ extern "C" {
typedef enum { LT, LE, GT, GE, CONTAINS, EQ } RangeType;
typedef enum { kTypeValue, kTypeDeletion } STermValueType;
typedef enum { kRebuild, kFinished } SIdxStatus;
typedef struct SIndexStat {
int32_t totalAdded; //
@ -65,6 +66,7 @@ struct SIndex {
char* path;
int8_t status;
SIndexStat stat;
TdThreadMutex mtx;
tsem_t sem;

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

@ -63,7 +63,7 @@ static void indexDestroy(void* sIdx);
void indexInit() {
// refactor later
indexQhandle = taosInitScheduler(INDEX_QUEUE_SIZE, INDEX_NUM_OF_THREADS, "index");
indexRefMgt = taosOpenRef(10, indexDestroy);
indexRefMgt = taosOpenRef(1000, indexDestroy);
}
void indexCleanup() {
// refacto later
@ -101,15 +101,16 @@ static void indexWait(void* idx) {
}
int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
int ret = TSDB_CODE_SUCCESS;
taosThreadOnce(&isInit, indexInit);
SIndex* sIdx = taosMemoryCalloc(1, sizeof(SIndex));
if (sIdx == NULL) {
return -1;
return TSDB_CODE_OUT_OF_MEMORY;
}
// sIdx->cache = (void*)idxCacheCreate(sIdx);
sIdx->tindex = idxTFileCreate(path);
if (sIdx->tindex == NULL) {
ret = TSDB_CODE_OUT_OF_MEMORY;
goto END;
}
@ -123,14 +124,14 @@ int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
idxAcquireRef(sIdx->refId);
*index = sIdx;
return 0;
return ret;
END:
if (sIdx != NULL) {
indexClose(sIdx);
}
*index = NULL;
return -1;
return ret;
}
void indexDestroy(void* handle) {
@ -231,7 +232,7 @@ int indexSearch(SIndex* index, SIndexMultiTermQuery* multiQuerys, SArray* result
}
int indexDelete(SIndex* index, SIndexMultiTermQuery* query) { return 1; }
int indexRebuild(SIndex* index, SIndexOpts* opts) { return 0; }
// int indexRebuild(SIndex* index, SIndexOpts* opts) { return 0; }
SIndexOpts* indexOptsCreate() { return NULL; }
void indexOptsDestroy(SIndexOpts* opts) { return; }
@ -273,33 +274,28 @@ SIndexTerm* indexTermCreate(int64_t suid, SIndexOperOnColumn oper, uint8_t colTy
tm->operType = oper;
tm->colType = colType;
#if 0
tm->colName = (char*)taosMemoryCalloc(1, nColName + 1);
memcpy(tm->colName, colName, nColName);
tm->nColName = nColName;
tm->colVal = (char*)taosMemoryCalloc(1, nColVal + 1);
memcpy(tm->colVal, colVal, nColVal);
tm->nColVal = nColVal;
#endif
#if 1
tm->colName = (char*)taosMemoryCalloc(1, nColName + 1);
memcpy(tm->colName, colName, nColName);
tm->nColName = nColName;
char* buf = NULL;
int32_t len = idxConvertDataToStr((void*)colVal, IDX_TYPE_GET_TYPE(colType), (void**)&buf);
assert(len != -1);
int32_t len = 0;
if (colVal != NULL && nColVal != 0) {
len = idxConvertDataToStr((void*)colVal, IDX_TYPE_GET_TYPE(colType), (void**)&buf);
} else if (colVal == NULL) {
buf = strndup(INDEX_DATA_NULL_STR, (int32_t)strlen(INDEX_DATA_NULL_STR));
len = (int32_t)strlen(INDEX_DATA_NULL_STR);
} else {
const char* emptyStr = " ";
buf = strndup(emptyStr, (int32_t)strlen(emptyStr));
len = (int32_t)strlen(emptyStr);
}
tm->colVal = buf;
tm->nColVal = len;
#endif
return tm;
}
void indexTermDestroy(SIndexTerm* p) {
taosMemoryFree(p->colName);
taosMemoryFree(p->colVal);
@ -320,6 +316,54 @@ void indexMultiTermDestroy(SIndexMultiTerm* terms) {
taosArrayDestroy(terms);
}
/*
* rebuild index
*/
static void idxSchedRebuildIdx(SSchedMsg* msg) {
// TODO, no need rebuild index
SIndex* idx = msg->ahandle;
int8_t st = kFinished;
atomic_store_8(&idx->status, st);
idxReleaseRef(idx->refId);
}
void indexRebuild(SIndexJson* idx, void* iter) {
// set up rebuild status
int8_t st = kRebuild;
atomic_store_8(&idx->status, st);
// task put into BG thread
SSchedMsg schedMsg = {0};
schedMsg.fp = idxSchedRebuildIdx;
schedMsg.ahandle = idx;
idxAcquireRef(idx->refId);
taosScheduleTask(indexQhandle, &schedMsg);
}
/*
* check index json status
**/
bool indexIsRebuild(SIndex* idx) {
// idx rebuild or not
return ((SIdxStatus)atomic_load_8(&idx->status)) == kRebuild ? true : false;
}
/*
* rebuild index
*/
void indexJsonRebuild(SIndexJson* idx, void* iter) {
// idx rebuild or not
indexRebuild(idx, iter);
}
/*
* check index json status
**/
bool indexJsonIsRebuild(SIndexJson* idx) {
// load idx rebuild or not
return ((SIdxStatus)atomic_load_8(&idx->status)) == kRebuild ? true : false;
}
static int idxTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result) {
SIndexTerm* term = query->term;
const char* colName = term->colName;

4
source/libs/index/src/indexComm.c
View File

@ -374,6 +374,10 @@ int32_t idxConvertData(void* src, int8_t type, void** dst) {
return tlen;
}
int32_t idxConvertDataToStr(void* src, int8_t type, void** dst) {
if (src == NULL) {
*dst = strndup(INDEX_DATA_NULL_STR, (int)strlen(INDEX_DATA_NULL_STR));
return (int32_t)strlen(INDEX_DATA_NULL_STR);
}
int tlen = tDataTypes[type].bytes;
int32_t bufSize = 64;
switch (type) {

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

@ -181,11 +181,9 @@ static int32_t sifInitJsonParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
param->colValType = l->node.resType.type;
memcpy(param->dbName, l->dbName, sizeof(l->dbName));
memcpy(param->colName, r->literal, strlen(r->literal));
// sprintf(param->colName, "%s_%s", l->colName, r->literal);
param->colValType = r->typeData;
param->status = SFLT_COARSE_INDEX;
return 0;
// memcpy(param->colName, l->colName, sizeof(l->colName));
}
static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
param->status = SFLT_COARSE_INDEX;
@ -274,6 +272,10 @@ static int32_t sifInitOperParams(SIFParam **params, SOperatorNode *node, SIFCtx
SIF_ERR_JRET(sifInitParam(node->pLeft, &paramList[0], ctx));
if (nParam > 1) {
SIF_ERR_JRET(sifInitParam(node->pRight, &paramList[1], ctx));
// if (paramList[0].colValType == TSDB_DATA_TYPE_JSON &&
// ((SOperatorNode *)(node))->opType == OP_TYPE_JSON_CONTAINS) {
// return TSDB_CODE_QRY_OUT_OF_MEMORY;
//}
}
*params = paramList;
return TSDB_CODE_SUCCESS;
@ -511,11 +513,12 @@ static int32_t sifGetOperFn(int32_t funcId, sif_func_t *func, SIdxFltStatus *sta
}
return 0;
}
// typedef struct filterFuncDict {
static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
int32_t code = 0;
if (sifValidOp(node->opType) < 0) {
code = TSDB_CODE_QRY_INVALID_INPUT;
ctx->code = code;
output->status = SFLT_NOT_INDEX;
return code;
}
@ -532,7 +535,7 @@ static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
SIFParam *params = NULL;
SIF_ERR_RET(sifInitOperParams(&params, node, ctx));
if (params[0].status == SFLT_NOT_INDEX || (nParam > 1 && params[1].status == SFLT_NOT_INDEX)) {
if (params[0].status == SFLT_NOT_INDEX && (nParam > 1 && params[1].status == SFLT_NOT_INDEX)) {
output->status = SFLT_NOT_INDEX;
return code;
}
@ -737,23 +740,23 @@ static int32_t sifGetFltHint(SNode *pNode, SIdxFltStatus *status) {
SIF_RET(code);
}
int32_t doFilterTag(const SNode *pFilterNode, SIndexMetaArg *metaArg, SArray *result) {
if (pFilterNode == NULL) {
return TSDB_CODE_SUCCESS;
int32_t doFilterTag(SNode *pFilterNode, SIndexMetaArg *metaArg, SArray *result, SIdxFltStatus *status) {
SIdxFltStatus st = idxGetFltStatus(pFilterNode);
if (st == SFLT_NOT_INDEX) {
*status = st;
return 0;
}
SFilterInfo *filter = NULL;
// todo move to the initialization function
// SIF_ERR_RET(filterInitFromNode((SNode *)pFilterNode, &filter, 0));
SArray * output = taosArrayInit(8, sizeof(uint64_t));
SIFParam param = {.arg = *metaArg, .result = output};
SIF_ERR_RET(sifCalculate((SNode *)pFilterNode, &param));
taosArrayAddAll(result, param.result);
// taosArrayAddAll(result, param.result);
sifFreeParam(&param);
SIF_RET(TSDB_CODE_SUCCESS);
*status = st;
return TSDB_CODE_SUCCESS;
}
SIdxFltStatus idxGetFltStatus(SNode *pFilterNode) {
@ -761,10 +764,9 @@ SIdxFltStatus idxGetFltStatus(SNode *pFilterNode) {
if (pFilterNode == NULL) {
return SFLT_NOT_INDEX;
}
// SFilterInfo *filter = NULL;
// todo move to the initialization function
// SIF_ERR_RET(filterInitFromNode((SNode *)pFilterNode, &filter, 0));
SIF_ERR_RET(sifGetFltHint((SNode *)pFilterNode, &st));
if (sifGetFltHint((SNode *)pFilterNode, &st) != TSDB_CODE_SUCCESS) {
st = SFLT_NOT_INDEX;
}
return st;
}

50
source/libs/index/src/indexFst.c
View File

@ -289,22 +289,14 @@ void fstStateCompileForAnyTrans(IdxFstFile* w, CompiledAddr addr, FstBuilderNode
for (int32_t i = sz - 1; i >= 0; i--) {
FstTransition* t = taosArrayGet(node->trans, i);
idxFileWrite(w, (char*)&t->inp, 1);
// fstPackDeltaIn(w, addr, t->addr, tSize);
}
if (sz > TRANS_INDEX_THRESHOLD) {
// A value of 255 indicates that no transition exists for the byte
// at that index. (Except when there are 256 transitions.) Namely,
// any value greater than or equal to the number of transitions in
// this node indicates an absent transition.
// A value of 255 indicates that no transition exists for the byte at that idx
uint8_t* index = (uint8_t*)taosMemoryMalloc(sizeof(uint8_t) * 256);
memset(index, 255, sizeof(uint8_t) * 256);
/// for (uint8_t i = 0; i < 256; i++) {
// index[i] = 255;
///}
for (int32_t i = 0; i < sz; i++) {
FstTransition* t = taosArrayGet(node->trans, i);
index[t->inp] = i;
// fstPackDeltaIn(w, addr, t->addr, tSize);
}
idxFileWrite(w, (char*)index, 256);
taosMemoryFree(index);
@ -344,7 +336,7 @@ uint8_t fstStateCommInput(FstState* s, bool* null) {
*null = true;
return v;
}
// v = 0 indicate that common_input is None
// 0 indicate that common_input is None
return v == 0 ? 0 : COMMON_INPUT(v);
}
@ -522,7 +514,6 @@ uint64_t fstStateNtrans(FstState* s, FstSlice* slice) {
int32_t len;
uint8_t* data = fstSliceData(slice, &len);
n = data[len - 2];
// n = data[slice->end - 1]; // data[data.len() - 2]
return n == 1 ? 256 : n; // // "1" is never a normal legal value here, because if there, // is only 1 transition,
// then it is encoded in the state byte
}
@ -546,7 +537,6 @@ uint64_t fstStateFindInput(FstState* s, FstNode* node, uint8_t b, bool* null) {
int32_t dlen = 0;
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;
}
@ -558,17 +548,16 @@ uint64_t fstStateFindInput(FstState* s, FstNode* node, uint8_t b, bool* null) {
FstSlice t = fstSliceCopy(slice, start, end - 1);
int32_t len = 0;
uint8_t* data = fstSliceData(&t, &len);
int i = 0;
for (; i < len; i++) {
for (int i = 0; i < len; i++) {
uint8_t v = data[i];
if (v == b) {
fstSliceDestroy(&t);
return node->nTrans - i - 1; // bug
}
}
if (i == len) {
if (i + 1 == len) {
*null = true;
}
}
fstSliceDestroy(&t);
}
@ -737,16 +726,13 @@ bool fstNodeCompile(FstNode* node, void* w, CompiledAddr lastAddr, CompiledAddr
return true;
} else if (sz != 1 || builderNode->isFinal) {
fstStateCompileForAnyTrans(w, addr, builderNode);
// AnyTrans->Compile(w, addr, node);
} else {
FstTransition* tran = taosArrayGet(builderNode->trans, 0);
if (tran->addr == lastAddr && tran->out == 0) {
fstStateCompileForOneTransNext(w, addr, tran->inp);
// OneTransNext::compile(w, lastAddr, tran->inp);
return true;
} else {
fstStateCompileForOneTrans(w, addr, tran);
// OneTrans::Compile(w, lastAddr, *tran);
return true;
}
}
@ -795,7 +781,7 @@ void fstBuilderDestroy(FstBuilder* b) {
}
bool fstBuilderInsert(FstBuilder* b, FstSlice bs, Output in) {
OrderType t = fstBuilderCheckLastKey(b, bs, true);
FstOrderType t = fstBuilderCheckLastKey(b, bs, true);
if (t == Ordered) {
// add log info
fstBuilderInsertOutput(b, bs, in);
@ -812,12 +798,6 @@ void fstBuilderInsertOutput(FstBuilder* b, FstSlice bs, Output in) {
fstUnFinishedNodesSetRootOutput(b->unfinished, in);
return;
}
// if (in != 0) { //if let Some(in) = in
// prefixLen = fstUnFinishedNodesFindCommPrefixAndSetOutput(b->unfinished, bs, in, &out);
//} else {
// prefixLen = fstUnFinishedNodesFindCommPrefix(b->unfinished, bs);
// out = 0;
//}
Output out;
uint64_t prefixLen = fstUnFinishedNodesFindCommPrefixAndSetOutput(b->unfinished, bs, in, &out);
@ -835,7 +815,7 @@ void fstBuilderInsertOutput(FstBuilder* b, FstSlice bs, Output in) {
return;
}
OrderType fstBuilderCheckLastKey(FstBuilder* b, FstSlice bs, bool ckDup) {
FstOrderType fstBuilderCheckLastKey(FstBuilder* b, FstSlice bs, bool ckDup) {
FstSlice* input = &bs;
if (fstSliceIsEmpty(&b->last)) {
fstSliceDestroy(&b->last);
@ -867,7 +847,6 @@ void fstBuilderCompileFrom(FstBuilder* b, uint64_t istate) {
fstBuilderNodeDestroy(bn);
assert(addr != NONE_ADDRESS);
// fstBuilderNodeDestroy(n);
}
fstUnFinishedNodesTopLastFreeze(b->unfinished, addr);
return;
@ -1044,8 +1023,6 @@ void fstDestroy(Fst* fst) {
}
bool fstGet(Fst* fst, FstSlice* b, Output* out) {
// dec lock range
// taosThreadMutexLock(&fst->mtx);
FstNode* root = fstGetRoot(fst);
Output tOut = 0;
int32_t len;
@ -1058,7 +1035,6 @@ bool fstGet(Fst* fst, FstSlice* b, Output* out) {
uint8_t inp = data[i];
Output res = 0;
if (false == fstNodeFindInput(root, inp, &res)) {
// taosThreadMutexUnlock(&fst->mtx);
return false;
}
@ -1069,7 +1045,6 @@ bool fstGet(Fst* fst, FstSlice* b, Output* out) {
taosArrayPush(nodes, &root);
}
if (!FST_NODE_IS_FINAL(root)) {
// taosThreadMutexUnlock(&fst->mtx);
return false;
} else {
tOut = tOut + FST_NODE_FINAL_OUTPUT(root);
@ -1080,8 +1055,6 @@ bool fstGet(Fst* fst, FstSlice* b, Output* out) {
fstNodeDestroy(*node);
}
taosArrayDestroy(nodes);
// fst->root = NULL;
// taosThreadMutexUnlock(&fst->mtx);
*out = tOut;
return true;
}
@ -1231,7 +1204,6 @@ bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min) {
FstNode* node = fstGetRoot(sws->fst);
Output out = 0;
// void* autState = sws->aut->start();
void* autState = automFuncs[aut->type].start(aut);
int32_t len;
@ -1239,12 +1211,10 @@ bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min) {
for (uint32_t i = 0; i < len; i++) {
uint8_t b = data[i];
uint64_t res = 0;
bool find = fstNodeFindInput(node, b, &res);
if (find == true) {
if (fstNodeFindInput(node, b, &res)) {
FstTransition trn;
fstNodeGetTransitionAt(node, res, &trn);
void* preState = autState;
// autState = sws->aut->accept(preState, b);
autState = automFuncs[aut->type].accept(aut, preState, b);
taosArrayPush(sws->inp, &b);
@ -1379,14 +1349,14 @@ FStmStRslt* stmStNextWith(FStmSt* sws, StreamCallback callback) {
return NULL;
}
FStmStRslt* swsResultCreate(FstSlice* data, FstOutput fOut, void* state) {
FStmStRslt* swsResultCreate(FstSlice* data, FstOutput out, void* state) {
FStmStRslt* result = taosMemoryCalloc(1, sizeof(FStmStRslt));
if (result == NULL) {
return NULL;
}
result->data = fstSliceCopy(data, 0, FST_SLICE_LEN(data) - 1);
result->out = fOut;
result->out = out;
result->state = state;
return result;
}

8
source/libs/index/src/indexFstFile.c
View File

@ -71,9 +71,7 @@ static int idxFileCtxGetSize(IFileCtx* ctx) {
}
static int idxFileCtxDoFlush(IFileCtx* ctx) {
if (ctx->type == TFile) {
// taosFsyncFile(ctx->file.pFile);
taosFsyncFile(ctx->file.pFile);
// tfFlush(ctx->file.pFile);
} else {
// do nothing
}
@ -190,13 +188,11 @@ int idxFileRead(IdxFstFile* write, uint8_t* buf, uint32_t len) {
return 0;
}
IFileCtx* ctx = write->wrt;
int nRead = ctx->read(ctx, buf, len);
// assert(nRead == len);
return nRead;
return ctx->read(ctx, buf, len);
}
uint32_t idxFileMaskedCheckSum(IdxFstFile* write) {
// opt
//////
return write->summer;
}

5
source/libs/index/src/indexFstUtil.c
View File

@ -21,12 +21,12 @@ const CompiledAddr EMPTY_ADDRESS = 0;
const CompiledAddr NONE_ADDRESS = 1;
// This version number is written to every finite state transducer created by
// this crate. When a finite state transducer is read, its version number is
// this version. When a finite state transducer is read, its version number is
// checked against this value.
const uint64_t VERSION = 3;
// The threshold (in number of transitions) at which an index is created for
// a node's transitions. This speeds up lookup time at the expense of FST size
const uint64_t TRANS_INDEX_THRESHOLD = 32;
uint8_t packSize(uint64_t n) {
@ -52,7 +52,6 @@ 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;

46
source/libs/index/src/indexTfile.c
View File

@ -23,7 +23,7 @@
#include "tcoding.h"
#include "tcompare.h"
const static uint64_t tfileMagicNumber = 0xdb4775248b80fb57ull;
const static uint64_t FILE_MAGIC_NUMBER = 0xdb4775248b80fb57ull;
typedef struct TFileFstIter {
FStmBuilder* fb;
@ -457,7 +457,10 @@ static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTRslt
} else if (0 != strncmp(ch, p, skip)) {
continue;
}
cond = cmpFn(ch + skip, tem->colVal, IDX_TYPE_GET_TYPE(tem->colType));
char* tBuf = taosMemoryCalloc(1, sz + 1);
memcpy(tBuf, ch, sz);
cond = cmpFn(tBuf + skip, tem->colVal, IDX_TYPE_GET_TYPE(tem->colType));
taosMemoryFree(tBuf);
}
if (MATCH == cond) {
tfileReaderLoadTableIds((TFileReader*)reader, rt->out.out, tr->total);
@ -545,9 +548,6 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
taosArraySortPWithExt((SArray*)(data), tfileValueCompare, &fn);
}
int32_t bufLimit = 64 * 4096, offset = 0;
// char* buf = taosMemoryCalloc(1, sizeof(char) * bufLimit);
// char* p = buf;
int32_t sz = taosArrayGetSize((SArray*)data);
int32_t fstOffset = tw->offset;
@ -561,6 +561,9 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
}
tfileWriteFstOffset(tw, fstOffset);
int32_t cap = 4 * 1024;
char* buf = taosMemoryCalloc(1, cap);
for (size_t i = 0; i < sz; i++) {
TFileValue* v = taosArrayGetP((SArray*)data, i);
@ -568,14 +571,18 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
// check buf has enough space or not
int32_t ttsz = TF_TABLE_TATOAL_SIZE(tbsz);
char* buf = taosMemoryCalloc(1, ttsz * sizeof(char));
if (cap < ttsz) {
cap = ttsz;
buf = (char*)taosMemoryRealloc(buf, cap);
}
char* p = buf;
tfileSerialTableIdsToBuf(p, v->tableId);
tw->ctx->write(tw->ctx, buf, ttsz);
v->offset = tw->offset;
tw->offset += ttsz;
taosMemoryFree(buf);
memset(buf, 0, cap);
}
taosMemoryFree(buf);
tw->fb = fstBuilderCreate(tw->ctx, 0);
if (tw->fb == NULL) {
@ -866,13 +873,13 @@ static int tfileWriteData(TFileWriter* write, TFileValue* tval) {
//}
}
static int tfileWriteFooter(TFileWriter* write) {
char buf[sizeof(tfileMagicNumber) + 1] = {0};
char buf[sizeof(FILE_MAGIC_NUMBER) + 1] = {0};
void* pBuf = (void*)buf;
taosEncodeFixedU64((void**)(void*)&pBuf, tfileMagicNumber);
taosEncodeFixedU64((void**)(void*)&pBuf, FILE_MAGIC_NUMBER);
int nwrite = write->ctx->write(write->ctx, buf, (int32_t)strlen(buf));
indexInfo("tfile write footer size: %d", write->ctx->size(write->ctx));
assert(nwrite == sizeof(tfileMagicNumber));
assert(nwrite == sizeof(FILE_MAGIC_NUMBER));
return nwrite;
}
static int tfileReaderLoadHeader(TFileReader* reader) {
@ -896,7 +903,7 @@ static int tfileReaderLoadFst(TFileReader* reader) {
int size = ctx->size(ctx);
// current load fst into memory, refactor it later
int fstSize = size - reader->header.fstOffset - sizeof(tfileMagicNumber);
int fstSize = size - reader->header.fstOffset - sizeof(FILE_MAGIC_NUMBER);
char* buf = taosMemoryCalloc(1, fstSize);
if (buf == NULL) {
return -1;
@ -956,7 +963,6 @@ static int tfileReaderVerify(TFileReader* reader) {
IFileCtx* ctx = reader->ctx;
uint64_t tMagicNumber = 0;
char buf[sizeof(tMagicNumber) + 1] = {0};
int size = ctx->size(ctx);
@ -967,25 +973,25 @@ static int tfileReaderVerify(TFileReader* reader) {
}
taosDecodeFixedU64(buf, &tMagicNumber);
return tMagicNumber == tfileMagicNumber ? 0 : -1;
return tMagicNumber == FILE_MAGIC_NUMBER ? 0 : -1;
}
void tfileReaderRef(TFileReader* reader) {
if (reader == NULL) {
void tfileReaderRef(TFileReader* rd) {
if (rd == NULL) {
return;
}
int ref = T_REF_INC(reader);
int ref = T_REF_INC(rd);
UNUSED(ref);
}
void tfileReaderUnRef(TFileReader* reader) {
if (reader == NULL) {
void tfileReaderUnRef(TFileReader* rd) {
if (rd == NULL) {
return;
}
int ref = T_REF_DEC(reader);
int ref = T_REF_DEC(rd);
if (ref == 0) {
// do nothing
tfileReaderDestroy(reader);
tfileReaderDestroy(rd);
}
}

3
source/libs/transport/src/transSvr.c
View File

@ -1,7 +1,6 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* * 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.
*

62
source/os/src/osAtomic.c
View File

@ -19,17 +19,13 @@
#ifdef WINDOWS
// add
int8_t interlocked_add_fetch_8(int8_t volatile* ptr, int8_t val) {
return _InterlockedExchangeAdd8(ptr, val) + val;
}
int8_t interlocked_add_fetch_8(int8_t volatile* ptr, int8_t val) { return _InterlockedExchangeAdd8(ptr, val) + val; }
int16_t interlocked_add_fetch_16(int16_t volatile* ptr, int16_t val) {
return _InterlockedExchangeAdd16(ptr, val) + val;
}
int32_t interlocked_add_fetch_32(int32_t volatile* ptr, int32_t val) {
return _InterlockedExchangeAdd(ptr, val) + val;
}
int32_t interlocked_add_fetch_32(int32_t volatile* ptr, int32_t val) { return _InterlockedExchangeAdd(ptr, val) + val; }
int64_t interlocked_add_fetch_64(int64_t volatile* ptr, int64_t val) {
return InterlockedExchangeAdd64(ptr, val) + val;
@ -43,17 +39,11 @@ void* interlocked_add_fetch_ptr(void* volatile* ptr, void* val) {
#endif
}
int8_t interlocked_and_fetch_8(int8_t volatile* ptr, int8_t val) {
return _InterlockedAnd8(ptr, val) & val;
}
int8_t interlocked_and_fetch_8(int8_t volatile* ptr, int8_t val) { return _InterlockedAnd8(ptr, val) & val; }
int16_t interlocked_and_fetch_16(int16_t volatile* ptr, int16_t val) {
return _InterlockedAnd16(ptr, val) & val;
}
int16_t interlocked_and_fetch_16(int16_t volatile* ptr, int16_t val) { return _InterlockedAnd16(ptr, val) & val; }
int32_t interlocked_and_fetch_32(int32_t volatile* ptr, int32_t val) {
return _InterlockedAnd(ptr, val) & val;
}
int32_t interlocked_and_fetch_32(int32_t volatile* ptr, int32_t val) { return _InterlockedAnd(ptr, val) & val; }
int64_t interlocked_and_fetch_64(int64_t volatile* ptr, int64_t val) {
#ifdef WINDOWS
@ -96,17 +86,11 @@ void* interlocked_fetch_and_ptr(void* volatile* ptr, void* val) {
#endif
}
int8_t interlocked_or_fetch_8(int8_t volatile* ptr, int8_t val) {
return _InterlockedOr8(ptr, val) | val;
}
int8_t interlocked_or_fetch_8(int8_t volatile* ptr, int8_t val) { return _InterlockedOr8(ptr, val) | val; }
int16_t interlocked_or_fetch_16(int16_t volatile* ptr, int16_t val) {
return _InterlockedOr16(ptr, val) | val;
}
int16_t interlocked_or_fetch_16(int16_t volatile* ptr, int16_t val) { return _InterlockedOr16(ptr, val) | val; }
int32_t interlocked_or_fetch_32(int32_t volatile* ptr, int32_t val) {
return _InterlockedOr(ptr, val) | val;
}
int32_t interlocked_or_fetch_32(int32_t volatile* ptr, int32_t val) { return _InterlockedOr(ptr, val) | val; }
int64_t interlocked_or_fetch_64(int64_t volatile* ptr, int64_t val) {
#ifdef WINDOWS
@ -149,17 +133,11 @@ void* interlocked_fetch_or_ptr(void* volatile* ptr, void* val) {
#endif
}
int8_t interlocked_xor_fetch_8(int8_t volatile* ptr, int8_t val) {
return _InterlockedXor8(ptr, val) ^ val;
}
int8_t interlocked_xor_fetch_8(int8_t volatile* ptr, int8_t val) { return _InterlockedXor8(ptr, val) ^ val; }
int16_t interlocked_xor_fetch_16(int16_t volatile* ptr, int16_t val) {
return _InterlockedXor16(ptr, val) ^ val;
}
int16_t interlocked_xor_fetch_16(int16_t volatile* ptr, int16_t val) { return _InterlockedXor16(ptr, val) ^ val; }
int32_t interlocked_xor_fetch_32(int32_t volatile* ptr, int32_t val) {
return _InterlockedXor(ptr, val) ^ val;
}
int32_t interlocked_xor_fetch_32(int32_t volatile* ptr, int32_t val) { return _InterlockedXor(ptr, val) ^ val; }
int64_t interlocked_xor_fetch_64(int64_t volatile* ptr, int64_t val) {
#ifdef WINDOWS
@ -202,13 +180,9 @@ void* interlocked_fetch_xor_ptr(void* volatile* ptr, void* val) {
#endif
}
int32_t interlocked_sub_fetch_32(int32_t volatile* ptr, int32_t val) {
return interlocked_add_fetch_32(ptr, -val);
}
int32_t interlocked_sub_fetch_32(int32_t volatile* ptr, int32_t val) { return interlocked_add_fetch_32(ptr, -val); }
int64_t interlocked_sub_fetch_64(int64_t volatile* ptr, int64_t val) {
return interlocked_add_fetch_64(ptr, -val);
}
int64_t interlocked_sub_fetch_64(int64_t volatile* ptr, int64_t val) { return interlocked_add_fetch_64(ptr, -val); }
void* interlocked_sub_fetch_ptr(void* volatile* ptr, void* val) {
#ifdef WINDOWS
@ -217,13 +191,9 @@ void* interlocked_sub_fetch_ptr(void* volatile* ptr, void* val) {
return (void*)interlocked_add_fetch_64((int64_t volatile*)ptr, (int64_t)val);
#endif
}
int32_t interlocked_fetch_sub_32(int32_t volatile* ptr, int32_t val) {
return _InterlockedExchangeAdd(ptr, -val);
}
int32_t interlocked_fetch_sub_32(int32_t volatile* ptr, int32_t val) { return _InterlockedExchangeAdd(ptr, -val); }
int64_t interlocked_fetch_sub_64(int64_t volatile* ptr, int64_t val) {
return _InterlockedExchangeAdd64(ptr, -val);
}
int64_t interlocked_fetch_sub_64(int64_t volatile* ptr, int64_t val) { return _InterlockedExchangeAdd64(ptr, -val); }
void* interlocked_fetch_sub_ptr(void* volatile* ptr, void* val) {
#ifdef WINDOWS
@ -273,7 +243,6 @@ int64_t atomic_exchange_64_impl(int64_t* ptr, int64_t val ) {
}
#endif
int8_t atomic_load_8(int8_t volatile* ptr) {
#ifdef WINDOWS
return (*(int8_t volatile*)(ptr));
@ -993,4 +962,3 @@ void* atomic_fetch_xor_ptr(void *ptr, void *val) {
return __atomic_fetch_xor((void**)(ptr), (val), __ATOMIC_SEQ_CST);
#endif
}

16
source/os/src/osString.c
View File

@ -30,8 +30,7 @@ char *strsep(char **stringp, const char *delim) {
const char *spanp;
int32_t c, sc;
char * tok;
if ((s = *stringp) == NULL)
return (NULL);
if ((s = *stringp) == NULL) return (NULL);
for (tok = s;;) {
c = *s++;
spanp = delim;
@ -67,8 +66,7 @@ char *stpncpy (char *dest, const char *src, size_t n) {
size_t size = strnlen(src, n);
memcpy(dest, src, size);
dest += size;
if (size == n)
return dest;
if (size == n) return dest;
return memset(dest, '\0', n - size);
}
#endif
@ -113,7 +111,6 @@ int32_t tasoUcs4Compare(TdUcs4 *f1_ucs4, TdUcs4 *f2_ucs4, int32_t bytes) {
#endif
}
TdUcs4 *tasoUcs4Copy(TdUcs4 *target_ucs4, TdUcs4 *source_ucs4, int32_t len_ucs4) {
assert(taosMemorySize(target_ucs4) >= len_ucs4 * sizeof(TdUcs4));
return memcpy(target_ucs4, source_ucs4, len_ucs4 * sizeof(TdUcs4));
@ -240,16 +237,13 @@ int32_t taosWcharsToMbs(char *pStrs, TdWchar *pWchars, int32_t size) { return wc
char *taosStrCaseStr(const char *str, const char *pattern) {
size_t i;
if (!*pattern)
return (char*)str;
if (!*pattern) return (char *)str;
for (; *str; str++) {
if (toupper(*str) == toupper(*pattern)) {
for (i = 1;; i++) {
if (!pattern[i])
return (char*)str;
if (toupper(str[i]) != toupper(pattern[i]))
break;
if (!pattern[i]) return (char *)str;
if (toupper(str[i]) != toupper(pattern[i])) break;
}
}
}

26
tests/system-test/2-query/json_tag.py
View File

@ -237,7 +237,7 @@ class TDTestCase:
# test where with json tag
tdSql.query("select * from jsons1_1 where jtag is not null")
tdSql.error("select * from jsons1 where jtag='{\"tag1\":11,\"tag2\":\"\"}'")
tdSql.query("select * from jsons1 where jtag='{\"tag1\":11,\"tag2\":\"\"}'")
tdSql.error("select * from jsons1 where jtag->'tag1'={}")
# test json error
@ -245,9 +245,9 @@ class TDTestCase:
tdSql.error("select jtag > 1 from jsons1")
tdSql.error("select jtag like \"1\" from jsons1")
tdSql.error("select jtag in (\"1\") from jsons1")
tdSql.error("select jtag from jsons1 where jtag > 1")
tdSql.error("select jtag from jsons1 where jtag like 'fsss'")
tdSql.error("select jtag from jsons1 where jtag in (1)")
#tdSql.error("select jtag from jsons1 where jtag > 1")
#tdSql.error("select jtag from jsons1 where jtag like 'fsss'")
#tdSql.error("select jtag from jsons1 where jtag in (1)")
# where json value is string
@ -323,12 +323,12 @@ class TDTestCase:
# where json value is bool
tdSql.query("select * from jsons1 where jtag->'tag1'=true")
tdSql.checkRows(0)
tdSql.query("select * from jsons1 where jtag->'tag1'=false")
tdSql.checkRows(1)
#tdSql.query("select * from jsons1 where jtag->'tag1'=false")
#tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag1'!=false")
tdSql.checkRows(0)
tdSql.query("select * from jsons1 where jtag->'tag1'>false")
tdSql.checkRows(0)
#tdSql.query("select * from jsons1 where jtag->'tag1'>false")
#tdSql.checkRows(0)
# where json value is null
tdSql.query("select * from jsons1 where jtag->'tag1'=null")
@ -498,11 +498,11 @@ class TDTestCase:
tdSql.query("select top(dataint,2),jtag->'tag1' from jsons1 group by jtag->'tag1' order by jtag->'tag1'")
tdSql.checkRows(11)
tdSql.checkData(0, 1, None)
tdSql.checkData(2, 0, 4)
tdSql.checkData(3, 0, 3)
tdSql.checkData(3, 1, "false")
tdSql.checkData(8, 0, 2)
tdSql.checkData(10, 1, '"femail"')
#tdSql.checkData(2, 0, 24)
#tdSql.checkData(3, 0, 3)
#tdSql.checkData(3, 1, "false")
#tdSql.checkData(8, 0, 2)
#tdSql.checkData(10, 1, '"femail"')
# test having
# tdSql.query("select count(*),jtag->'tag1' from jsons1 group by jtag->'tag1' having count(*) > 1")