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

Merge branch '3.0' into fix/TD-31163-3.0

This commit is contained in:
kailixu 2024-08-20 13:55:35 +08:00
parent 5e77f6f6ca 24747af339
commit fe3415c602
35 changed files with 675 additions and 354 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/util/taoserror.h
View File

@ -604,6 +604,9 @@ int32_t taosGetErrSize();
#define TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x0732)
#define TSDB_CODE_QRY_INVALID_JOIN_CONDITION TAOS_DEF_ERROR_CODE(0, 0x0733)
#define TSDB_CODE_QRY_FILTER_NOT_SUPPORT_TYPE TAOS_DEF_ERROR_CODE(0, 0x0734)
#define TSDB_CODE_QRY_FILTER_WRONG_OPTR_TYPE TAOS_DEF_ERROR_CODE(0, 0x0735)
#define TSDB_CODE_QRY_FILTER_RANGE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0736)
#define TSDB_CODE_QRY_FILTER_INVALID_TYPE TAOS_DEF_ERROR_CODE(0, 0x0737)
// grant
#define TSDB_CODE_GRANT_EXPIRED TAOS_DEF_ERROR_CODE(0, 0x0800)
@ -878,6 +881,8 @@ int32_t taosGetErrSize();
#define TSDB_CODE_FUNC_INVALID_VALUE_RANGE TAOS_DEF_ERROR_CODE(0, 0x280C)
#define TSDB_CODE_FUNC_SETUP_ERROR TAOS_DEF_ERROR_CODE(0, 0x280D)
#define TSDB_CODE_FUNC_INVALID_RES_LENGTH TAOS_DEF_ERROR_CODE(0, 0x280E)
#define TSDB_CODE_FUNC_HISTOGRAM_ERROR TAOS_DEF_ERROR_CODE(0, 0x280F)
#define TSDB_CODE_FUNC_PERCENTILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x2810)
//udf

3
include/util/tcoding.h
View File

@ -213,7 +213,6 @@ static FORCE_INLINE int32_t taosEncodeVariantU16(void **buf, uint16_t value) {
if (buf != NULL) ((uint8_t *)(*buf))[i] = (uint8_t)(value | ENCODE_LIMIT);
value >>= 7;
i++;
ASSERT(i < 3);
}
if (buf != NULL) {
@ -261,7 +260,6 @@ static FORCE_INLINE int32_t taosEncodeVariantU32(void **buf, uint32_t value) {
if (buf != NULL) ((uint8_t *)(*buf))[i] = (value | ENCODE_LIMIT);
value >>= 7;
i++;
ASSERT(i < 5);
}
if (buf != NULL) {
@ -309,7 +307,6 @@ static FORCE_INLINE int32_t taosEncodeVariantU64(void **buf, uint64_t value) {
if (buf != NULL) ((uint8_t *)(*buf))[i] = (uint8_t)(value | ENCODE_LIMIT);
value >>= 7;
i++;
ASSERT(i < 10);
}
if (buf != NULL) {

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

@ -12,7 +12,7 @@ extern char tsS3AccessKeySecret[][TSDB_FQDN_LEN];
extern char tsS3BucketName[TSDB_FQDN_LEN];
extern char tsS3AppId[][TSDB_FQDN_LEN];
extern char tsS3Hostname[][TSDB_FQDN_LEN];
extern int8_t tsS3Https;
extern int8_t tsS3Https[];
static int32_t s3ListBucketByEp(char const *bucketname, int8_t epIndex);
static int32_t s3PutObjectFromFileOffsetByEp(const char *file, const char *object_name, int64_t offset, int64_t size,
@ -33,13 +33,13 @@ static int verifyPeerG = 0;
static const char *awsRegionG = NULL;
static int forceG = 0;
static int showResponsePropertiesG = 0;
static S3Protocol protocolG = S3ProtocolHTTPS;
static S3Protocol protocolG[TSDB_MAX_EP_NUM] = {S3ProtocolHTTPS};
// static S3Protocol protocolG = S3ProtocolHTTP;
static S3UriStyle uriStyleG = S3UriStylePath;
static S3UriStyle uriStyleG[TSDB_MAX_EP_NUM] = {S3UriStylePath};
static int retriesG = 5;
static int timeoutMsG = 0;
extern int8_t tsS3Oss;
extern int8_t tsS3Oss[];
int32_t s3Begin() {
S3Status status;
@ -55,9 +55,9 @@ int32_t s3Begin() {
TAOS_RETURN(TSDB_CODE_FAILED);
}
protocolG = !tsS3Https;
if (tsS3Oss) {
uriStyleG = S3UriStyleVirtualHost;
for (int i = 0; i < tsS3EpNum; i++) {
protocolG[i] = tsS3Https[i] ? S3ProtocolHTTPS : S3ProtocolHTTP;
uriStyleG[i] = tsS3Oss[i] ? S3UriStyleVirtualHost : S3UriStylePath;
}
TAOS_RETURN(TSDB_CODE_SUCCESS);
@ -976,8 +976,8 @@ int32_t s3PutObjectFromFile2ByEp(const char *file, const char *object_name, int8
S3BucketContext bucketContext = {tsS3Hostname[epIndex],
tsS3BucketName,
protocolG,
uriStyleG,
protocolG[epIndex],
uriStyleG[epIndex],
tsS3AccessKeyId[epIndex],
tsS3AccessKeySecret[epIndex],
0,
@ -1059,8 +1059,8 @@ static int32_t s3PutObjectFromFileOffsetByEp(const char *file, const char *objec
S3BucketContext bucketContext = {tsS3Hostname[epIndex],
tsS3BucketName,
protocolG,
uriStyleG,
protocolG[epIndex],
uriStyleG[epIndex],
tsS3AccessKeyId[epIndex],
tsS3AccessKeySecret[epIndex],
0,
@ -1155,8 +1155,8 @@ static void s3FreeObjectKey(void *pItem) {
static SArray *getListByPrefixByEp(const char *prefix, int8_t epIndex) {
S3BucketContext bucketContext = {tsS3Hostname[epIndex],
tsS3BucketName,
protocolG,
uriStyleG,
protocolG[epIndex],
uriStyleG[epIndex],
tsS3AccessKeyId[epIndex],
tsS3AccessKeySecret[epIndex],
0,
@ -1223,8 +1223,8 @@ static int32_t s3DeleteObjectsByEp(const char *object_name[], int nobject, int8_
S3BucketContext bucketContext = {tsS3Hostname[epIndex],
tsS3BucketName,
protocolG,
uriStyleG,
protocolG[epIndex],
uriStyleG[epIndex],
tsS3AccessKeyId[epIndex],
tsS3AccessKeySecret[epIndex],
0,
@ -1299,8 +1299,8 @@ static int32_t s3GetObjectBlockByEp(const char *object_name, int64_t offset, int
S3BucketContext bucketContext = {tsS3Hostname[epIndex],
tsS3BucketName,
protocolG,
uriStyleG,
protocolG[epIndex],
uriStyleG[epIndex],
tsS3AccessKeyId[epIndex],
tsS3AccessKeySecret[epIndex],
0,
@ -1372,8 +1372,8 @@ static int32_t s3GetObjectToFileByEp(const char *object_name, const char *fileNa
S3BucketContext bucketContext = {tsS3Hostname[epIndex],
tsS3BucketName,
protocolG,
uriStyleG,
protocolG[epIndex],
uriStyleG[epIndex],
tsS3AccessKeyId[epIndex],
tsS3AccessKeySecret[epIndex],
0,
@ -1449,8 +1449,8 @@ static long s3SizeByEp(const char *object_name, int8_t epIndex) {
S3BucketContext bucketContext = {tsS3Hostname[epIndex],
tsS3BucketName,
protocolG,
uriStyleG,
protocolG[epIndex],
uriStyleG[epIndex],
tsS3AccessKeyId[epIndex],
tsS3AccessKeySecret[epIndex],
0,

9
source/common/src/tglobal.c
View File

@ -303,10 +303,10 @@ char tsS3BucketName[TSDB_FQDN_LEN] = "<bucketname>";
char tsS3AppId[TSDB_MAX_EP_NUM][TSDB_FQDN_LEN] = {"<appid>"};
int8_t tsS3Enabled = false;
int8_t tsS3EnabledCfg = false;
int8_t tsS3Oss = false;
int8_t tsS3Oss[TSDB_MAX_EP_NUM] = {false};
int8_t tsS3StreamEnabled = false;
int8_t tsS3Https = true;
int8_t tsS3Https[TSDB_MAX_EP_NUM] = {true};
char tsS3Hostname[TSDB_MAX_EP_NUM][TSDB_FQDN_LEN] = {"<hostname>"};
int32_t tsS3BlockSize = -1; // number of tsdb pages (4096)
@ -431,11 +431,10 @@ int32_t taosSetS3Cfg(SConfig *pCfg) {
tstrncpy(tsS3AppId[i], appid + 1, TSDB_FQDN_LEN);
}
}
tsS3Https[i] = (strstr(tsS3Endpoint[i], "https://") != NULL);
tsS3Oss[i] = (strstr(tsS3Endpoint[i], "aliyuncs.") != NULL);
}
tsS3Https = (strstr(tsS3Endpoint[0], "https://") != NULL);
tsS3Oss = (strstr(tsS3Endpoint[0], "aliyuncs.") != NULL);
if (tsS3BucketName[0] != '<') {
#if defined(USE_COS) || defined(USE_S3)
#ifdef TD_ENTERPRISE

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

@ -898,7 +898,7 @@ typedef struct SSttDataInfoForTable {
int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf, SSttDataInfoForTable *pTableInfo);
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree);
int32_t tMergeTreeNext(SMergeTree *pMTree, bool* pHasNext);
void tMergeTreePinSttBlock(SMergeTree *pMTree);
void tMergeTreeUnpinSttBlock(SMergeTree *pMTree);
bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree);

4
source/dnode/vnode/src/tq/tq.c
View File

@ -1204,11 +1204,11 @@ int32_t tqProcessTaskCheckPointSourceReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg* pRsp)
}
if (req.mndTrigger) {
qInfo("s-task:%s (vgId:%d) level:%d receive checkpoint-source msg chkpt:%" PRId64 ", transId:%d, ", pTask->id.idStr,
tqInfo("s-task:%s (vgId:%d) level:%d receive checkpoint-source msg chkpt:%" PRId64 ", transId:%d, ", pTask->id.idStr,
vgId, pTask->info.taskLevel, req.checkpointId, req.transId);
} else {
const char* pPrevStatus = streamTaskGetStatusStr(streamTaskGetPrevStatus(pTask));
qInfo("s-task:%s (vgId:%d) level:%d receive checkpoint-source msg chkpt:%" PRId64
tqInfo("s-task:%s (vgId:%d) level:%d receive checkpoint-source msg chkpt:%" PRId64
", transId:%d after transfer-state, prev status:%s",
pTask->id.idStr, vgId, pTask->info.taskLevel, req.checkpointId, req.transId, pPrevStatus);
}

11
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -2245,17 +2245,20 @@ static int32_t lastIterClose(SFSLastIter **iter) {
}
static int32_t lastIterNext(SFSLastIter *iter, TSDBROW **ppRow) {
int32_t code = 0;
bool hasVal = false;
*ppRow = NULL;
int32_t code = tMergeTreeNext(iter->pMergeTree, &hasVal);
if (code != 0) {
return code;
}
bool hasVal = tMergeTreeNext(iter->pMergeTree);
if (!hasVal) {
*ppRow = NULL;
TAOS_RETURN(code);
}
*ppRow = tMergeTreeGetRow(iter->pMergeTree);
TAOS_RETURN(code);
}

41
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
View File

@ -115,16 +115,14 @@ void destroySttBlockReader(SArray *pLDataIterArray, SSttBlockLoadCostInfo *pLoad
SArray *pList = taosArrayGetP(pLDataIterArray, i);
for (int32_t j = 0; j < taosArrayGetSize(pList); ++j) {
SLDataIter *pIter = taosArrayGetP(pList, j);
if (pIter->pBlockLoadInfo == NULL) {
continue;
}
SSttBlockLoadCostInfo *pCost = &pIter->pBlockLoadInfo->cost;
if (pLoadCost != NULL) {
pLoadCost->loadBlocks += pCost->loadBlocks;
pLoadCost->loadStatisBlocks += pCost->loadStatisBlocks;
pLoadCost->blockElapsedTime += pCost->blockElapsedTime;
pLoadCost->statisElapsedTime += pCost->statisElapsedTime;
if (pIter->pBlockLoadInfo != NULL) {
SSttBlockLoadCostInfo *pCost = &pIter->pBlockLoadInfo->cost;
if (pLoadCost != NULL) {
pLoadCost->loadBlocks += pCost->loadBlocks;
pLoadCost->loadStatisBlocks += pCost->loadStatisBlocks;
pLoadCost->blockElapsedTime += pCost->blockElapsedTime;
pLoadCost->statisElapsedTime += pCost->statisElapsedTime;
}
}
destroyLDataIter(pIter);
@ -903,6 +901,7 @@ int32_t tLDataIterNextRow(SLDataIter *pIter, const char *idStr, bool* hasNext) {
pIter->rInfo.row = tsdbRowFromBlockData(pBlockData, pIter->iRow);
_exit:
tsdbError("failed to exec stt-file nextIter, lino:%d, code:%s, %s", lino, tstrerror(code), idStr);
*hasNext = (code == TSDB_CODE_SUCCESS) && (pIter->pSttBlk != NULL) && (pBlockData != NULL);
return code;
}
@ -1102,13 +1101,22 @@ void tMergeTreeUnpinSttBlock(SMergeTree *pMTree) {
tLDataIterUnpinSttBlock(pIter, pMTree->idStr);
}
bool tMergeTreeNext(SMergeTree *pMTree) {
int32_t tMergeTreeNext(SMergeTree *pMTree, bool *pHasNext) {
int32_t code = 0;
if (pHasNext == NULL) {
return TSDB_CODE_INVALID_PARA;
}
*pHasNext = false;
if (pMTree->pIter) {
SLDataIter *pIter = pMTree->pIter;
bool hasVal = false;
int32_t code = tLDataIterNextRow(pIter, pMTree->idStr, &hasVal);
bool hasVal = false;
code = tLDataIterNextRow(pIter, pMTree->idStr, &hasVal);
if (!hasVal || (code != 0)) {
if (code == TSDB_CODE_FILE_CORRUPTED) {
code = 0; // suppress the file corrupt error to enable all queries within this cluster can run without failed.
}
pMTree->pIter = NULL;
}
@ -1117,7 +1125,7 @@ bool tMergeTreeNext(SMergeTree *pMTree) {
if (pMTree->pIter && pIter) {
int32_t c = pMTree->rbt.cmprFn(&pMTree->pIter->node, &pIter->node);
if (c > 0) {
(void) tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);
(void)tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);
pMTree->pIter = NULL;
} else {
ASSERT(c);
@ -1132,7 +1140,8 @@ bool tMergeTreeNext(SMergeTree *pMTree) {
}
}
return pMTree->pIter != NULL;
*pHasNext = (pMTree->pIter != NULL);
return code;
}
void tMergeTreeClose(SMergeTree *pMTree) {

92
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

@ -1383,7 +1383,6 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, SRowKey* pLastPro
static FORCE_INLINE STSchema* getTableSchemaImpl(STsdbReader* pReader, uint64_t uid) {
ASSERT(pReader->info.pSchema == NULL);
int32_t code = metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->info.suid, uid, -1, &pReader->info.pSchema);
if (code != TSDB_CODE_SUCCESS || pReader->info.pSchema == NULL) {
terrno = code;
@ -1414,7 +1413,9 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI
if (pReader->info.pSchema == NULL) {
pSchema = getTableSchemaImpl(pReader, uid);
if (pSchema == NULL) {
tsdbDebug("%p table uid:%" PRIu64 " has been dropped, no data existed, %s", pReader, uid, pReader->idStr);
code = terrno;
tsdbError("%p table uid:%" PRIu64 " failed to get tableschema, code:%s, %s", pReader, uid, tstrerror(code),
pReader->idStr);
return code;
}
}
@ -1449,7 +1450,7 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI
pReader->cost.blockLoadTime += elapsedTime;
pDumpInfo->allDumped = false;
return TSDB_CODE_SUCCESS;
return code;
}
/**
@ -1759,14 +1760,22 @@ static bool tryCopyDistinctRowFromFileBlock(STsdbReader* pReader, SBlockData* pB
return code;
}
static bool nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, int32_t pkSrcSlot,
SVersionRange* pVerRange) {
static int32_t nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, int32_t pkSrcSlot,
SVersionRange* pVerRange) {
int32_t code = 0;
int32_t order = pSttBlockReader->order;
int32_t step = ASCENDING_TRAVERSE(order) ? 1 : -1;
SRowKey* pNextProc = &pScanInfo->sttKeyInfo.nextProcKey;
while (1) {
bool hasVal = tMergeTreeNext(&pSttBlockReader->mergeTree);
bool hasVal = false;
code = tMergeTreeNext(&pSttBlockReader->mergeTree, &hasVal);
if (code) {
tsdbError("failed to iter the next row in stt-file merge tree, code:%s, %s", tstrerror(code),
pSttBlockReader->mergeTree.idStr);
return code;
}
if (!hasVal) { // the next value will be the accessed key in stt
pScanInfo->sttKeyInfo.status = STT_FILE_NO_DATA;
@ -1779,7 +1788,7 @@ static bool nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockSc
memset(pNextProc->pks[0].pData, 0, pNextProc->pks[0].nData);
}
}
return false;
return code;
}
TSDBROW* pRow = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
@ -1798,13 +1807,15 @@ static bool nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockSc
if (!hasBeenDropped(pScanInfo->delSkyline, &pScanInfo->sttBlockDelIndex, key, ver, order, pVerRange,
pSttBlockReader->numOfPks > 0)) {
pScanInfo->sttKeyInfo.status = STT_FILE_HAS_DATA;
return true;
return code;
}
} else {
pScanInfo->sttKeyInfo.status = STT_FILE_HAS_DATA;
return true;
return code;
}
}
return code;
}
static void doPinSttBlock(SSttBlockReader* pSttBlockReader) { tMergeTreePinSttBlock(&pSttBlockReader->mergeTree); }
@ -1819,9 +1830,14 @@ static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SSttBlockReader* pSttB
// avoid the fetch next row replace the referenced stt block in buffer
doPinSttBlock(pSttBlockReader);
bool hasVal = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
code = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
doUnpinSttBlock(pSttBlockReader);
if (hasVal) {
if (code) {
return code;
}
if (hasDataInSttBlock(pScanInfo)) {
SRowKey* pNext = getCurrentKeyInSttBlock(pSttBlockReader);
if (pkCompEx(pSttKey, pNext) != 0) {
code = doAppendRowFromFileBlock(pReader->resBlockInfo.pResBlock, pReader, fRow->pBlockData, fRow->iRow);
@ -2125,7 +2141,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
if (piRow->type == TSDBROW_ROW_FMT) {
piSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(piRow), pReader, pBlockScanInfo->uid);
if (piSchema == NULL) {
return code;
return terrno;
}
}
@ -2380,14 +2396,13 @@ static bool isValidFileBlockRow(SBlockData* pBlockData, int32_t rowIndex, STable
return true;
}
static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
bool hasData = true;
static void initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
int32_t order = pReader->info.order;
bool asc = ASCENDING_TRAVERSE(order);
// the stt block reader has been initialized for this table.
if (pSttBlockReader->uid == pScanInfo->uid) {
return hasDataInSttBlock(pScanInfo);
return;
}
if (pSttBlockReader->uid != 0) {
@ -2396,9 +2411,14 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
pSttBlockReader->uid = pScanInfo->uid;
// second time init stt block reader
// second or third time init stt block reader
if (pScanInfo->cleanSttBlocks && (pReader->info.execMode == READER_EXEC_ROWS)) {
return !pScanInfo->sttBlockReturned;
// only allowed to retrieve clean stt blocks for count once
if (pScanInfo->sttBlockReturned) {
pScanInfo->sttKeyInfo.status = STT_FILE_NO_DATA;
tsdbDebug("uid:%" PRIu64 " set no stt-file data after stt-block retrieved, %s", pScanInfo->uid, pReader->idStr);
}
return;
}
STimeWindow w = pSttBlockReader->window;
@ -2435,28 +2455,28 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
SSttDataInfoForTable info = {.pKeyRangeList = taosArrayInit(4, sizeof(SSttKeyRange))};
if (info.pKeyRangeList == NULL) {
pReader->code = terrno;
return false;
return;
}
int32_t code = tMergeTreeOpen2(&pSttBlockReader->mergeTree, &conf, &info);
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(info.pKeyRangeList);
pReader->code = code;
return false;
return;
}
code = initMemDataIterator(pScanInfo, pReader);
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(info.pKeyRangeList);
pReader->code = code;
return false;
return;
}
code = initDelSkylineIterator(pScanInfo, pReader->info.order, &pReader->cost);
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(info.pKeyRangeList);
pReader->code = code;
return false;
return;
}
if (conf.rspRows) {
@ -2484,27 +2504,26 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
SRowKey* p = asc ? &pScanInfo->sttRange.skey : &pScanInfo->sttRange.ekey;
tRowKeyAssign(&pScanInfo->sttKeyInfo.nextProcKey, p);
hasData = (pScanInfo->sttKeyInfo.status == STT_FILE_HAS_DATA);
} else { // not clean stt blocks
INIT_KEYRANGE(&pScanInfo->sttRange); // reset the time window
pScanInfo->sttBlockReturned = false;
hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
code = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
}
} else {
pScanInfo->cleanSttBlocks = false;
INIT_KEYRANGE(&pScanInfo->sttRange); // reset the time window
pScanInfo->sttBlockReturned = false;
hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
code = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
}
pScanInfo->sttBlockReturned = false;
taosArrayDestroy(info.pKeyRangeList);
int64_t el = taosGetTimestampUs() - st;
pReader->cost.initSttBlockReader += (el / 1000.0);
tsdbDebug("init stt block reader completed, elapsed time:%" PRId64 "us %s", el, pReader->idStr);
return hasData;
if (code != 0) {
pReader->code = code;
}
}
static bool hasDataInSttBlock(STableBlockScanInfo* pInfo) { return pInfo->sttKeyInfo.status == STT_FILE_HAS_DATA; }
@ -2772,7 +2791,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
}
SBlockData* pBlockData = &pReader->status.fileBlockData;
(void) initSttBlockReader(pSttBlockReader, pBlockScanInfo, pReader);
initSttBlockReader(pSttBlockReader, pBlockScanInfo, pReader);
if (pReader->code != 0) {
code = pReader->code;
goto _end;
@ -3190,12 +3209,12 @@ static int32_t doLoadSttBlockSequentially(STsdbReader* pReader) {
continue;
}
bool hasDataInSttFile = initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
if (pReader->code != TSDB_CODE_SUCCESS) {
return pReader->code;
}
if (!hasDataInSttFile) {
if (!hasDataInSttBlock(pScanInfo)) {
bool hasNexTable = moveToNextTable(pUidList, pStatus);
if (!hasNexTable) {
return TSDB_CODE_SUCCESS;
@ -3287,7 +3306,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
}
if (pScanInfo->sttKeyInfo.status == STT_FILE_READER_UNINIT) {
(void) initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
if (pReader->code != 0) {
return pReader->code;
}
@ -3331,7 +3350,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
int64_t st = taosGetTimestampUs();
// let's load data from stt files, make sure clear the cleanStt block flag before load the data from stt files
(void) initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
if (pReader->code != 0) {
return pReader->code;
}
@ -4087,7 +4106,12 @@ int32_t doMergeRowsInSttBlock(SSttBlockReader* pSttBlockReader, STableBlockScanI
SRowKey* pRowKey = &pScanInfo->lastProcKey;
int32_t code = TSDB_CODE_SUCCESS;
while (nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pkSrcSlot, pVerRange)) {
while (1) {
code = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pkSrcSlot, pVerRange);
if (code || (!hasDataInSttBlock(pScanInfo))) {
return code;
}
SRowKey* pNextKey = getCurrentKeyInSttBlock(pSttBlockReader);
int32_t ret = pkCompEx(pRowKey, pNextKey);

36
source/dnode/vnode/src/tsdb/tsdbSttFileRW.c
View File

@ -61,7 +61,9 @@ int32_t tsdbSttFileReaderOpen(const char *fname, const SSttFileReaderConfig *con
// // open each segment reader
int64_t offset = config->file->size - sizeof(SSttFooter);
ASSERT(offset >= TSDB_FHDR_SIZE);
if (offset < TSDB_FHDR_SIZE) {
TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
}
int32_t encryptAlgoirthm = config->tsdb->pVnode->config.tsdbCfg.encryptAlgorithm;
char *encryptKey = config->tsdb->pVnode->config.tsdbCfg.encryptKey;
@ -115,7 +117,9 @@ int32_t tsdbSttFileReaderClose(SSttFileReader **reader) {
int32_t tsdbSttFileReadStatisBlk(SSttFileReader *reader, const TStatisBlkArray **statisBlkArray) {
if (!reader->ctx->statisBlkLoaded) {
if (reader->footer->statisBlkPtr->size > 0) {
ASSERT(reader->footer->statisBlkPtr->size % sizeof(SStatisBlk) == 0);
if (reader->footer->statisBlkPtr->size % sizeof(SStatisBlk) != 0) {
return TSDB_CODE_FILE_CORRUPTED;
}
int32_t size = reader->footer->statisBlkPtr->size / sizeof(SStatisBlk);
void *data = taosMemoryMalloc(reader->footer->statisBlkPtr->size);
@ -147,7 +151,9 @@ int32_t tsdbSttFileReadStatisBlk(SSttFileReader *reader, const TStatisBlkArray *
int32_t tsdbSttFileReadTombBlk(SSttFileReader *reader, const TTombBlkArray **tombBlkArray) {
if (!reader->ctx->tombBlkLoaded) {
if (reader->footer->tombBlkPtr->size > 0) {
ASSERT(reader->footer->tombBlkPtr->size % sizeof(STombBlk) == 0);
if (reader->footer->tombBlkPtr->size % sizeof(STombBlk) != 0) {
return TSDB_CODE_FILE_CORRUPTED;
}
int32_t size = reader->footer->tombBlkPtr->size / sizeof(STombBlk);
void *data = taosMemoryMalloc(reader->footer->tombBlkPtr->size);
@ -179,7 +185,9 @@ int32_t tsdbSttFileReadTombBlk(SSttFileReader *reader, const TTombBlkArray **tom
int32_t tsdbSttFileReadSttBlk(SSttFileReader *reader, const TSttBlkArray **sttBlkArray) {
if (!reader->ctx->sttBlkLoaded) {
if (reader->footer->sttBlkPtr->size > 0) {
ASSERT(reader->footer->sttBlkPtr->size % sizeof(SSttBlk) == 0);
if (reader->footer->sttBlkPtr->size % sizeof(SSttBlk) != 0) {
return TSDB_CODE_FILE_CORRUPTED;
}
int32_t size = reader->footer->sttBlkPtr->size / sizeof(SSttBlk);
void *data = taosMemoryMalloc(reader->footer->sttBlkPtr->size);
@ -256,7 +264,9 @@ int32_t tsdbSttFileReadBlockDataByColumn(SSttFileReader *reader, const SSttBlk *
SBufferReader br = BUFFER_READER_INITIALIZER(0, buffer0);
TAOS_CHECK_GOTO(tGetDiskDataHdr(&br, &hdr), &lino, _exit);
ASSERT(hdr.delimiter == TSDB_FILE_DLMT);
if (hdr.delimiter != TSDB_FILE_DLMT) {
TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
}
// set data container
tBlockDataReset(bData);
@ -266,7 +276,9 @@ int32_t tsdbSttFileReadBlockDataByColumn(SSttFileReader *reader, const SSttBlk *
// key part
TAOS_CHECK_GOTO(tBlockDataDecompressKeyPart(&hdr, &br, bData, assist), &lino, _exit);
ASSERT(br.offset == buffer0->size);
if (br.offset != buffer0->size) {
TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
}
bool loadExtra = false;
for (int i = 0; i < ncid; i++) {
@ -376,7 +388,10 @@ int32_t tsdbSttFileReadTombBlock(SSttFileReader *reader, const STombBlk *tombBlk
br.offset += tombBlk->size[i];
}
ASSERT(br.offset == tombBlk->dp->size);
if (br.offset != tombBlk->dp->size) {
TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
}
_exit:
if (code) {
tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(reader->config->tsdb->pVnode), __func__, __FILE__, lino,
@ -444,7 +459,9 @@ int32_t tsdbSttFileReadStatisBlock(SSttFileReader *reader, const SStatisBlk *sta
}
}
ASSERT(br.offset == buffer0->size);
if (br.offset != buffer0->size) {
TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
}
_exit:
if (code) {
@ -814,8 +831,6 @@ _exit:
}
static void tsdbSttFWriterDoClose(SSttFileWriter *writer) {
ASSERT(writer->fd == NULL);
for (int32_t i = 0; i < ARRAY_SIZE(writer->local); ++i) {
tBufferDestroy(writer->local + i);
}
@ -854,7 +869,6 @@ static int32_t tsdbSttFWriterCloseCommit(SSttFileWriter *writer, TFileOpArray *o
tsdbCloseFile(&writer->fd);
ASSERT(writer->file->size > 0);
STFileOp op = (STFileOp){
.optype = TSDB_FOP_CREATE,
.fid = writer->config->fid,

2
source/dnode/vnode/src/tsdb/tsdbUtil2.c
View File

@ -171,7 +171,7 @@ static int32_t tStatisBlockUpdate(STbStatisBlock *block, SRowInfo *row) {
TAOS_CHECK_RETURN(tBufferPutAt(&block->counts, (block->numOfRecords - 1) * sizeof(record.count), &record.count,
sizeof(record.count)));
} else {
ASSERT(0);
return TSDB_CODE_INVALID_PARA;
}
return 0;

15
source/dnode/vnode/src/vnd/vnodeAsync.c
View File

@ -165,9 +165,7 @@ static int32_t vnodeAsyncTaskDone(SVAsync *async, SVATask *task) {
}
ret = vHashDrop(async->taskTable, task);
if (ret != 0) {
ASSERT(0);
}
TAOS_UNUSED(ret);
async->numTasks--;
if (task->numWait == 0) {
@ -403,7 +401,6 @@ static int32_t vnodeAsyncDestroy(SVAsync **async) {
}
(void)taosThreadJoin((*async)->workers[i].thread, NULL);
ASSERT((*async)->workers[i].state == EVA_WORKER_STATE_STOP);
(*async)->workers[i].state = EVA_WORKER_STATE_UINIT;
}
@ -413,18 +410,11 @@ static int32_t vnodeAsyncDestroy(SVAsync **async) {
channel->prev->next = channel->next;
int32_t ret = vHashDrop((*async)->channelTable, channel);
if (ret) {
ASSERT(0);
}
TAOS_UNUSED(ret);
(*async)->numChannels--;
taosMemoryFree(channel);
}
ASSERT((*async)->numLaunchWorkers == 0);
ASSERT((*async)->numIdleWorkers == 0);
ASSERT((*async)->numChannels == 0);
ASSERT((*async)->numTasks == 0);
(void)taosThreadMutexDestroy(&(*async)->mutex);
(void)taosThreadCondDestroy(&(*async)->hasTask);
@ -438,7 +428,6 @@ static int32_t vnodeAsyncDestroy(SVAsync **async) {
static int32_t vnodeAsyncLaunchWorker(SVAsync *async) {
for (int32_t i = 0; i < async->numWorkers; i++) {
ASSERT(async->workers[i].state != EVA_WORKER_STATE_IDLE);
if (async->workers[i].state == EVA_WORKER_STATE_ACTIVE) {
continue;
} else if (async->workers[i].state == EVA_WORKER_STATE_STOP) {

3
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -302,7 +302,6 @@ static int32_t vnodePrepareCommit(SVnode *pVnode, SCommitInfo *pInfo) {
TSDB_CHECK_CODE(code, lino, _exit);
(void)taosThreadMutexLock(&pVnode->mutex);
ASSERT(pVnode->onCommit == NULL);
pVnode->onCommit = pVnode->inUse;
pVnode->inUse = NULL;
(void)taosThreadMutexUnlock(&pVnode->mutex);
@ -339,7 +338,7 @@ static void vnodeReturnBufPool(SVnode *pVnode) {
pVnode->recycleTail = pPool;
}
} else {
ASSERT(0);
vError("vgId:%d, buffer pool %p of id %d nRef:%d", TD_VID(pVnode), pPool, pPool->id, nRef);
}
(void)taosThreadMutexUnlock(&pVnode->mutex);

1
source/dnode/vnode/src/vnd/vnodeHash.c
View File

@ -77,7 +77,6 @@ int32_t vHashDestroy(SVHashTable** ht) {
}
if (*ht) {
ASSERT((*ht)->numEntries == 0);
taosMemoryFree((*ht)->buckets);
taosMemoryFree(*ht);
(*ht) = NULL;

4
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -558,7 +558,9 @@ void vnodeClose(SVnode *pVnode) {
// start the sync timer after the queue is ready
int32_t vnodeStart(SVnode *pVnode) {
ASSERT(pVnode);
if (pVnode == NULL) {
return TSDB_CODE_INVALID_PARA;
}
return vnodeSyncStart(pVnode);
}

5
source/libs/executor/src/exchangeoperator.c
View File

@ -225,7 +225,10 @@ static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) {
} else {
concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
}
if (TSDB_CODE_SUCCESS != pOperator->pTaskInfo->code) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
T_LONG_JMP(pTaskInfo->env, pOperator->pTaskInfo->code);
}
if (taosArrayGetSize(pExchangeInfo->pResultBlockList) == 0) {
return NULL;
} else {

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

@ -663,7 +663,7 @@ int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds, bo
if (isTaskKilled(pTaskInfo)) {
atomic_store_64(&pTaskInfo->owner, 0);
qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));
return TSDB_CODE_SUCCESS;
return pTaskInfo->code;
}
// error occurs, record the error code and return to client
@ -785,7 +785,7 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) {
qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));
taosRUnLockLatch(&pTaskInfo->lock);
return TSDB_CODE_SUCCESS;
return pTaskInfo->code;
}
if (pTaskInfo->owner != 0) {

7
source/libs/executor/src/scanoperator.c
View File

@ -464,15 +464,10 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
SSDataBlock* p = NULL;
code = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pTableScanInfo->dataReader, &p, NULL);
if (p == NULL || code != TSDB_CODE_SUCCESS) {
if (p == NULL || code != TSDB_CODE_SUCCESS || p != pBlock) {
return code;
}
if(p != pBlock) {
qError("[loadDataBlock] p != pBlock");
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
code = doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
if (code) {
return code;

2
source/libs/function/inc/thistogram.h
View File

@ -59,7 +59,7 @@ int32_t tHistogramCreate(int32_t numOfEntries, SHistogramInfo** pHisto);
SHistogramInfo* tHistogramCreateFrom(void* pBuf, int32_t numOfBins);
int32_t tHistogramAdd(SHistogramInfo** pHisto, double val);
int64_t tHistogramSum(SHistogramInfo* pHisto, double v);
int32_t tHistogramSum(SHistogramInfo* pHisto, double v, int64_t *res);
int32_t tHistogramUniform(SHistogramInfo* pHisto, double* ratio, int32_t num, double** pVal);
int32_t tHistogramMerge(SHistogramInfo* pHisto1, SHistogramInfo* pHisto2, int32_t numOfEntries,

2
source/libs/function/inc/tpercentile.h
View File

@ -47,7 +47,7 @@ typedef struct tMemBucketSlot {
} tMemBucketSlot;
struct tMemBucket;
typedef int32_t (*__perc_hash_func_t)(struct tMemBucket *pBucket, const void *value);
typedef int32_t (*__perc_hash_func_t)(struct tMemBucket *pBucket, const void *value, int32_t *index);
typedef struct tMemBucket {
int16_t numOfSlots;

4
source/libs/function/src/builtinsimpl.c
View File

@ -4018,7 +4018,9 @@ int32_t saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock*
if (NULL == pColInfo) {
return TSDB_CODE_OUT_OF_RANGE;
}
ASSERT(pColInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
if (pColInfo->info.type != TSDB_DATA_TYPE_TIMESTAMP) {
return TSDB_CODE_FUNC_FUNTION_PARA_TYPE;
}
key.groupId = pSrcBlock->info.id.groupId;
key.ts = *(int64_t*)colDataGetData(pColInfo, rowIndex);
}

89
source/libs/function/src/thistogram.c
View File

@ -14,6 +14,7 @@
*/
#include "os.h"
#include "query.h"
#include "taosdef.h"
#include "thistogram.h"
#include "tlosertree.h"
@ -81,9 +82,9 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
#if defined(USE_ARRAYLIST)
int32_t idx = histoBinarySearch((*pHisto)->elems, (*pHisto)->numOfEntries, val);
if (ASSERTS(idx >= 0 && idx <= (*pHisto)->maxEntries && (*pHisto)->elems != NULL, "tHistogramAdd Error, idx:%d, maxEntries:%d, elems:%p",
idx, (*pHisto)->maxEntries, (*pHisto)->elems)) {
return TSDB_CODE_FAILED;
if (idx < 0 || idx > (*pHisto)->maxEntries || (*pHisto)->elems == NULL) {
qError("tHistogramAdd Error, idx:%d, maxEntries:%d, elems:%p", idx, (*pHisto)->maxEntries, (*pHisto)->elems);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
if ((*pHisto)->elems[idx].val == val && idx >= 0) {
@ -95,21 +96,19 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
} else { /* insert a new slot */
if ((*pHisto)->numOfElems >= 1 && idx < (*pHisto)->numOfEntries) {
if (idx > 0) {
if (ASSERTS((*pHisto)->elems[idx - 1].val <= val, "tHistogramAdd Error, elems[%d].val:%lf, val:%lf",
idx - 1, (*pHisto)->elems[idx - 1].val, val)) {
return TSDB_CODE_FAILED;
if ((*pHisto)->elems[idx - 1].val > val) {
qError("tHistogramAdd Error, elems[%d].val:%lf, val:%lf", idx - 1, (*pHisto)->elems[idx - 1].val, val);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
} else {
if (ASSERTS((*pHisto)->elems[idx].val > val, "tHistogramAdd Error, elems[%d].val:%lf, val:%lf",
idx, (*pHisto)->elems[idx].val, val)) {
return TSDB_CODE_FAILED;
if ((*pHisto)->elems[idx].val <= val) {
qError("tHistogramAdd Error, elems[%d].val:%lf, val:%lf", idx, (*pHisto)->elems[idx].val, val);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
}
} else if ((*pHisto)->numOfElems > 0) {
if (ASSERTS((*pHisto)->elems[(*pHisto)->numOfEntries].val <= val, "tHistogramAdd Error, elems[%d].val:%lf, val:%lf",
(*pHisto)->numOfEntries, (*pHisto)->elems[idx].val, val)) {
return TSDB_CODE_FAILED;
}
} else if ((*pHisto)->numOfElems > 0 && (*pHisto)->elems[(*pHisto)->numOfEntries].val > val) {
qError("tHistogramAdd Error, elems[%d].val:%lf, val:%lf", (*pHisto)->numOfEntries, (*pHisto)->elems[idx].val, val);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
code = histogramCreateBin(*pHisto, idx, val);
@ -225,9 +224,9 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
tSkipListNode* pNext = pNode->pForward[0];
SHistBin* pNextEntry = (SHistBin*)pNext->pData;
if (ASSERTS(pNextEntry->val - pEntry->val == pEntry->delta, "tHistogramAdd Error, pNextEntry->val:%lf, pEntry->val:%lf, pEntry->delta:%lf",
pNextEntry->val, pEntry->val, pEntry->delta)) {
return -1;
if (pNextEntry->val - pEntry->val != pEntry->delta) {
qError("tHistogramAdd Error, pNextEntry->val:%lf, pEntry->val:%lf, pEntry->delta:%lf", pNextEntry->val, pEntry->val, pEntry->delta);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
double newVal = (pEntry->val * pEntry->num + pNextEntry->val * pNextEntry->num) / (pEntry->num + pNextEntry->num);
@ -278,8 +277,9 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
} else {
SHistBin* pEntry = (SHistBin*)pResNode->pData;
if (ASSERTS(pEntry->val == val, "tHistogramAdd Error, pEntry->val:%lf, val:%lf")) {
return -1;
if (pEntry->val != val) {
qError("tHistogramAdd Error, pEntry->val:%lf, val:%lf", pEntry->val, val);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
pEntry->num += 1;
}
@ -356,9 +356,9 @@ int32_t histogramCreateBin(SHistogramInfo* pHisto, int32_t index, double val) {
(void)memmove(&pHisto->elems[index + 1], &pHisto->elems[index], sizeof(SHistBin) * remain);
}
if (ASSERTS(index >= 0 && index <= pHisto->maxEntries, "histogramCreateBin Error, index:%d, maxEntries:%d",
index, pHisto->maxEntries)) {
return TSDB_CODE_FAILED;
if (index < 0 || index > pHisto->maxEntries) {
qError("histogramCreateBin Error, index:%d, maxEntries:%d", index, pHisto->maxEntries);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
pHisto->elems[index].num = 1;
@ -373,9 +373,9 @@ int32_t histogramCreateBin(SHistogramInfo* pHisto, int32_t index, double val) {
pHisto->elems[pHisto->maxEntries].num = 0;
}
#endif
if (ASSERTS(pHisto->numOfEntries <= pHisto->maxEntries, "histogramCreateBin Error, numOfEntries:%d, maxEntries:%d",
pHisto->numOfEntries, pHisto->maxEntries)) {
return TSDB_CODE_FAILED;
if (pHisto->numOfEntries > pHisto->maxEntries) {
qError("histogramCreateBin Error, numOfEntries:%d, maxEntries:%d", pHisto->numOfEntries, pHisto->maxEntries);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
return TSDB_CODE_SUCCESS;
@ -411,8 +411,9 @@ void tHistogramPrint(SHistogramInfo* pHisto) {
* Estimated number of points in the interval (−inf,b].
* @param pHisto
* @param v
* @param res
*/
int64_t tHistogramSum(SHistogramInfo* pHisto, double v) {
int32_t tHistogramSum(SHistogramInfo* pHisto, double v, int64_t *res) {
#if defined(USE_ARRAYLIST)
int32_t slotIdx = histoBinarySearch(pHisto->elems, pHisto->numOfEntries, v);
if (pHisto->elems[slotIdx].val != v) {
@ -420,14 +421,18 @@ int64_t tHistogramSum(SHistogramInfo* pHisto, double v) {
if (slotIdx < 0) {
slotIdx = 0;
ASSERTS(v <= pHisto->elems[slotIdx].val, "tHistogramSum Error, elems[%d].val:%lf, v:%lf",
slotIdx, pHisto->elems[slotIdx].val, v);
if (v > pHisto->elems[slotIdx].val) {
qError("tHistogramSum Error, elems[%d].val:%lf, v:%lf", slotIdx, pHisto->elems[slotIdx].val, v);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
} else {
ASSERTS(v >= pHisto->elems[slotIdx].val, "tHistogramSum Error, elems[%d].val:%lf, v:%lf",
slotIdx, pHisto->elems[slotIdx].val, v);
if (slotIdx + 1 < pHisto->numOfEntries) {
ASSERTS(v < pHisto->elems[slotIdx + 1].val, "tHistogramSum Error, elems[%d].val:%lf, v:%lf",
slotIdx + 1, pHisto->elems[slotIdx + 1].val, v);
if (v < pHisto->elems[slotIdx].val) {
qError("tHistogramSum Error, elems[%d].val:%lf, v:%lf", slotIdx, pHisto->elems[slotIdx].val, v);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
if (slotIdx + 1 < pHisto->numOfEntries && v >= pHisto->elems[slotIdx + 1].val) {
qError("tHistogramSum Error, elems[%d].val:%lf, v:%lf", slotIdx + 1, pHisto->elems[slotIdx + 1].val, v);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
}
}
@ -447,8 +452,9 @@ int64_t tHistogramSum(SHistogramInfo* pHisto, double v) {
s1 = s1 + m1 / 2;
return (int64_t)s1;
*res = (int64_t)s1;
#endif
return TSDB_CODE_SUCCESS;
}
int32_t tHistogramUniform(SHistogramInfo* pHisto, double* ratio, int32_t num, double** pVal) {
@ -484,9 +490,11 @@ int32_t tHistogramUniform(SHistogramInfo* pHisto, double* ratio, int32_t num, do
j += 1;
}
ASSERTS(total <= numOfElem && total + pHisto->elems[j + 1].num > numOfElem,
"tHistogramUniform Error, total:%ld, numOfElem:%ld, elems[%d].num:%ld",
total, (int64_t)numOfElem, j + 1, pHisto->elems[j + 1].num);
if (total > numOfElem || total + pHisto->elems[j + 1].num <= numOfElem) {
qError("tHistogramUniform Error, total:%d, numOfElem:%d, elems[%d].num:%d",
(int32_t)total, (int32_t)numOfElem, j + 1, (int32_t)pHisto->elems[j + 1].num);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
double delta = numOfElem - total;
if (fabs(delta) < FLT_EPSILON) {
@ -545,9 +553,10 @@ int32_t tHistogramUniform(SHistogramInfo* pHisto, double* ratio, int32_t num, do
j += 1;
}
ASSERTS(total <= numOfElem && total + pEntry->num > numOfElem,
"tHistogramUniform Error, total:%d, numOfElem:%d, pEntry->num:%d",
total, numOfElem, pEntry->num);
if (total > numOfElem || total + pEntry->num <= numOfElem) {
qError("tHistogramUniform Error, total:%d, numOfElem:%d, pEntry->num:%d", (int32_t)total, (int32_t)numOfElem, (int32_t)pEntry->num);
return TSDB_CODE_FUNC_HISTOGRAM_ERROR;
}
double delta = numOfElem - total;
if (fabs(delta) < FLT_EPSILON) {

126
source/libs/function/src/tpercentile.c
View File

@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "query.h"
#include "taoserror.h"
#include "tcompare.h"
#include "tglobal.h"
@ -107,7 +108,10 @@ static void resetPosInfo(SSlotInfo *pInfo) {
}
int32_t findOnlyResult(tMemBucket *pMemBucket, double *result) {
ASSERT(pMemBucket->total == 1);
if (pMemBucket->total != 1) {
qError("MemBucket:%p, total:%d, but only one element is expected", pMemBucket, pMemBucket->total);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
terrno = 0;
for (int32_t i = 0; i < pMemBucket->numOfSlots; ++i) {
@ -120,7 +124,10 @@ int32_t findOnlyResult(tMemBucket *pMemBucket, double *result) {
SArray **pList = taosHashGet(pMemBucket->groupPagesMap, &groupId, sizeof(groupId));
if (pList != NULL) {
SArray *list = *pList;
ASSERT(list->size == 1);
if (list->size != 1) {
qError("list:%p, total list size:%zu, but only one element is expected", list, list->size);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
int32_t *pageId = taosArrayGet(list, 0);
if (NULL == pageId) {
@ -130,7 +137,11 @@ int32_t findOnlyResult(tMemBucket *pMemBucket, double *result) {
if (pPage == NULL) {
return terrno;
}
ASSERT(pPage->num == 1);
if (pPage->num != 1) {
qError("page:%p, total num:%d, but only one element is expected", pPage, pPage->num);
releaseBufPage(pMemBucket->pBuffer, pPage);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
GET_TYPED_DATA(*result, double, pMemBucket->type, pPage->data);
return TSDB_CODE_SUCCESS;
@ -141,64 +152,69 @@ int32_t findOnlyResult(tMemBucket *pMemBucket, double *result) {
return TSDB_CODE_SUCCESS;
}
int32_t tBucketIntHash(tMemBucket *pBucket, const void *value) {
int32_t tBucketIntHash(tMemBucket *pBucket, const void *value, int32_t *index) {
int64_t v = 0;
GET_TYPED_DATA(v, int64_t, pBucket->type, value);
int32_t index = -1;
*index = -1;
if (v > pBucket->range.dMaxVal || v < pBucket->range.dMinVal) {
return index;
return TSDB_CODE_SUCCESS;
}
// divide the value range into 1024 buckets
uint64_t span = pBucket->range.dMaxVal - pBucket->range.dMinVal;
if (span < pBucket->numOfSlots) {
int64_t delta = v - pBucket->range.dMinVal;
index = (delta % pBucket->numOfSlots);
*index = (delta % pBucket->numOfSlots);
} else {
double slotSpan = ((double)span) / pBucket->numOfSlots;
uint64_t delta = (uint64_t)(v - pBucket->range.dMinVal);
index = delta / slotSpan;
if (v == pBucket->range.dMaxVal || index == pBucket->numOfSlots) {
index -= 1;
*index = delta / slotSpan;
if (v == pBucket->range.dMaxVal || *index == pBucket->numOfSlots) {
*index -= 1;
}
}
ASSERTS(index >= 0 && index < pBucket->numOfSlots, "tBucketIntHash Error, index:%d, numOfSlots:%d",
index, pBucket->numOfSlots);
return index;
if (*index < 0 || *index >= pBucket->numOfSlots) {
qError("tBucketIntHash Error, index:%d, numOfSlots:%d", *index, pBucket->numOfSlots);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
return TSDB_CODE_SUCCESS;
}
int32_t tBucketUintHash(tMemBucket *pBucket, const void *value) {
int32_t tBucketUintHash(tMemBucket *pBucket, const void *value, int32_t *index) {
int64_t v = 0;
GET_TYPED_DATA(v, uint64_t, pBucket->type, value);
int32_t index = -1;
*index = -1;
if (v > pBucket->range.u64MaxVal || v < pBucket->range.u64MinVal) {
return index;
return TSDB_CODE_SUCCESS;
}
// divide the value range into 1024 buckets
uint64_t span = pBucket->range.u64MaxVal - pBucket->range.u64MinVal;
if (span < pBucket->numOfSlots) {
int64_t delta = v - pBucket->range.u64MinVal;
index = (int32_t)(delta % pBucket->numOfSlots);
*index = (int32_t)(delta % pBucket->numOfSlots);
} else {
double slotSpan = (double)span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.u64MinVal) / slotSpan);
*index = (int32_t)((v - pBucket->range.u64MinVal) / slotSpan);
if (v == pBucket->range.u64MaxVal) {
index -= 1;
*index -= 1;
}
}
ASSERT(index >= 0 && index < pBucket->numOfSlots);
return index;
if (*index < 0 || *index >= pBucket->numOfSlots) {
qError("tBucketUintHash Error, index:%d, numOfSlots:%d", *index, pBucket->numOfSlots);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
return TSDB_CODE_SUCCESS;
}
int32_t tBucketDoubleHash(tMemBucket *pBucket, const void *value) {
int32_t tBucketDoubleHash(tMemBucket *pBucket, const void *value, int32_t *index) {
double v = 0;
if (pBucket->type == TSDB_DATA_TYPE_FLOAT) {
v = GET_FLOAT_VAL(value);
@ -206,27 +222,30 @@ int32_t tBucketDoubleHash(tMemBucket *pBucket, const void *value) {
v = GET_DOUBLE_VAL(value);
}
int32_t index = -1;
*index = -1;
if (v > pBucket->range.dMaxVal || v < pBucket->range.dMinVal) {
return index;
return TSDB_CODE_SUCCESS;
}
// divide a range of [dMinVal, dMaxVal] into 1024 buckets
double span = pBucket->range.dMaxVal - pBucket->range.dMinVal;
if (span < pBucket->numOfSlots) {
int32_t delta = (int32_t)(v - pBucket->range.dMinVal);
index = (delta % pBucket->numOfSlots);
*index = (delta % pBucket->numOfSlots);
} else {
double slotSpan = span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.dMinVal) / slotSpan);
*index = (int32_t)((v - pBucket->range.dMinVal) / slotSpan);
if (v == pBucket->range.dMaxVal) {
index -= 1;
*index -= 1;
}
}
ASSERT(index >= 0 && index < pBucket->numOfSlots);
return index;
if (*index < 0 || *index >= pBucket->numOfSlots) {
qError("tBucketDoubleHash Error, index:%d, numOfSlots:%d", *index, pBucket->numOfSlots);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
return TSDB_CODE_SUCCESS;
}
static __perc_hash_func_t getHashFunc(int32_t type) {
@ -333,7 +352,7 @@ void tMemBucketDestroy(tMemBucket *pBucket) {
taosMemoryFreeClear(pBucket);
}
void tMemBucketUpdateBoundingBox(MinMaxEntry *r, const char *data, int32_t dataType) {
int32_t tMemBucketUpdateBoundingBox(MinMaxEntry *r, const char *data, int32_t dataType) {
if (IS_SIGNED_NUMERIC_TYPE(dataType)) {
int64_t v = 0;
GET_TYPED_DATA(v, int64_t, dataType, data);
@ -368,8 +387,10 @@ void tMemBucketUpdateBoundingBox(MinMaxEntry *r, const char *data, int32_t dataT
r->dMaxVal = v;
}
} else {
ASSERT(0);
qError("tMemBucketUpdateBoundingBox Error, invalid data type:%d", dataType);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
return TSDB_CODE_SUCCESS;
}
/*
@ -378,9 +399,14 @@ void tMemBucketUpdateBoundingBox(MinMaxEntry *r, const char *data, int32_t dataT
int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
int32_t count = 0;
int32_t bytes = pBucket->bytes;
int32_t code = TSDB_CODE_SUCCESS;
for (int32_t i = 0; i < size; ++i) {
char *d = (char *)data + i * bytes;
int32_t index = (pBucket->hashFunc)(pBucket, d);
int32_t index = -1;
code = (pBucket->hashFunc)(pBucket, d, &index);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
if (index < 0) {
continue;
}
@ -388,7 +414,10 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
count += 1;
tMemBucketSlot *pSlot = &pBucket->pSlots[index];
tMemBucketUpdateBoundingBox(&pSlot->range, d, pBucket->type);
code = tMemBucketUpdateBoundingBox(&pSlot->range, d, pBucket->type);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
// ensure available memory pages to allocate
int32_t groupId = getGroupId(pBucket->numOfSlots, index, pBucket->times);
@ -396,7 +425,11 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
if (pSlot->info.data == NULL || pSlot->info.data->num >= pBucket->elemPerPage) {
if (pSlot->info.data != NULL) {
ASSERT(pSlot->info.data->num >= pBucket->elemPerPage && pSlot->info.size > 0);
if (pSlot->info.data->num < pBucket->elemPerPage || pSlot->info.size <= 0) {
qError("tMemBucketPut failed since wrong pSLot info dataNum : %d, size : %d",
pSlot->info.data->num, pSlot->info.size);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
// keep the pointer in memory
setBufPageDirty(pSlot->info.data, true);
@ -411,7 +444,7 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
if (NULL == pPageIdList) {
return terrno;
}
int32_t code = taosHashPut(pBucket->groupPagesMap, &groupId, sizeof(groupId), &pPageIdList, POINTER_BYTES);
code = taosHashPut(pBucket->groupPagesMap, &groupId, sizeof(groupId), &pPageIdList, POINTER_BYTES);
if (TSDB_CODE_SUCCESS != code) {
taosArrayDestroy(pPageIdList);
return code;
@ -449,21 +482,23 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
* j is the last slot of current segment, we need to get the first
* slot of the next segment.
*/
static MinMaxEntry getMinMaxEntryOfNextSlotWithData(tMemBucket *pMemBucket, int32_t slotIdx) {
static int32_t getMinMaxEntryOfNextSlotWithData(tMemBucket *pMemBucket, int32_t slotIdx, MinMaxEntry *next) {
int32_t j = slotIdx + 1;
while (j < pMemBucket->numOfSlots && (pMemBucket->pSlots[j].info.size == 0)) {
++j;
}
ASSERT(j < pMemBucket->numOfSlots);
return pMemBucket->pSlots[j].range;
if (j >= pMemBucket->numOfSlots) {
qError("getMinMaxEntryOfNextSlotWithData can not get valid slot, start with slotIdx:%d", slotIdx);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
*next = pMemBucket->pSlots[j].range;
return TSDB_CODE_SUCCESS;
}
static bool isIdenticalData(tMemBucket *pMemBucket, int32_t index);
static double getIdenticalDataVal(tMemBucket *pMemBucket, int32_t slotIndex) {
ASSERT(isIdenticalData(pMemBucket, slotIndex));
tMemBucketSlot *pSlot = &pMemBucket->pSlots[slotIndex];
double finalResult = 0.0;
@ -494,7 +529,11 @@ int32_t getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction
* now, we need to find the minimum value of the next slot for interpolating the percentile value
* j is the last slot of current segment, we need to get the first slot of the next segment.
*/
MinMaxEntry next = getMinMaxEntryOfNextSlotWithData(pMemBucket, i);
MinMaxEntry next;
int32_t code = getMinMaxEntryOfNextSlotWithData(pMemBucket, i, &next);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
double maxOfThisSlot = 0;
double minOfNextSlot = 0;
@ -509,7 +548,10 @@ int32_t getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction
minOfNextSlot = (double)next.dMinVal;
}
ASSERT(minOfNextSlot > maxOfThisSlot);
if (minOfNextSlot <= maxOfThisSlot) {
qError("getPercentileImpl get minOfNextSlot : %f less equal than maxOfThisSlot : %f", minOfNextSlot, maxOfThisSlot);
return TSDB_CODE_FUNC_PERCENTILE_ERROR;
}
*result = (1 - fraction) * maxOfThisSlot + fraction * minOfNextSlot;
return TSDB_CODE_SUCCESS;

6
source/libs/geometry/src/geomFunc.c
View File

@ -131,12 +131,12 @@ _exit:
int32_t executeMakePointFunc(SColumnInfoData *pInputData[], int32_t iLeft, int32_t iRight,
SColumnInfoData *pOutputData) {
int32_t code = TSDB_CODE_FAILED;
unsigned char *output = NULL;
_getDoubleValue_fn_t getDoubleValueFn[2];
getDoubleValueFn[0] = getVectorDoubleValueFn(pInputData[0]->info.type);
getDoubleValueFn[1] = getVectorDoubleValueFn(pInputData[1]->info.type);
TAOS_CHECK_GOTO(getVectorDoubleValueFn(pInputData[0]->info.type, &getDoubleValueFn[0]), NULL, _exit);
TAOS_CHECK_GOTO(getVectorDoubleValueFn(pInputData[1]->info.type, &getDoubleValueFn[1]), NULL, _exit);
unsigned char *output = NULL;
double leftRes = 0;
double rightRes = 0;

58
source/libs/parser/src/parTranslater.c
View File

@ -12628,6 +12628,27 @@ static int32_t createParOperatorNode(EOperatorType opType, const char* pLeftCol,
return TSDB_CODE_SUCCESS;
}
static int32_t createIsOperatorNode(EOperatorType opType, const char* pColName, SNode** pOp) {
SOperatorNode* pOper = NULL;
int32_t code = nodesMakeNode(QUERY_NODE_OPERATOR, (SNode**)&pOper);
if (NULL == pOper) {
return code;
}
pOper->opType = opType;
code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)&pOper->pLeft);
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode((SNode*)pOper);
return code;
}
pOper->pRight = NULL;
snprintf(((SColumnNode*)pOper->pLeft)->colName, sizeof(((SColumnNode*)pOper->pLeft)->colName), "%s", pColName);
*pOp = (SNode*)pOper;
return TSDB_CODE_SUCCESS;
}
static const char* getTbNameColName(ENodeType type) {
const char* colName;
switch (type) {
@ -15035,7 +15056,7 @@ static int32_t rewriteShowAliveStmt(STranslateContext* pCxt, SQuery* pQuery) {
// pWhenThenlist and pElse need to free
// case when (v1_status = "leader" or v2_status = "lead er" or v3_status = "leader" or v4_status = "leader") then 1
// case when (v1_status = "leader" or v2_status = "leader" or v3_status = "leader" or v4_status = "leader") then 1
// else 0 end
SNode* pCaseWhen = NULL;
code = createParCaseWhenNode(NULL, pWhenThenlist, pElse, NULL, &pCaseWhen);
@ -15190,23 +15211,42 @@ static int32_t rewriteShowAliveStmt(STranslateContext* pCxt, SQuery* pQuery) {
return code;
}
// pSubSelect, pTemp1, pTempVal need to free
pThen = NULL;
code = nodesMakeValueNodeFromInt32(1, &pThen);
SNode* pCondIsNULL = NULL;
code = createIsOperatorNode(OP_TYPE_IS_NULL, pSumColAlias, &pCondIsNULL);
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode((SNode*)pSubSelect);
nodesDestroyNode(pTemp1);
nodesDestroyNode(pTempVal);
return code;
}
// pSubSelect, pTemp1, pThen, pTempVal need to free
pWhenThen = NULL;
code = createParWhenThenNode(pTemp1, pThen, &pWhenThen);
SNode* pCondFull1 = NULL;
code = createLogicCondNode(&pTemp1, &pCondIsNULL, &pCondFull1, LOGIC_COND_TYPE_OR);
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode((SNode*)pSubSelect);
nodesDestroyNode(pTemp1);
nodesDestroyNode(pTempVal);
nodesDestroyNode(pCondIsNULL);
return code;
}
// pSubSelect, pCondFull1, pTempVal need to free
pThen = NULL;
code = nodesMakeValueNodeFromInt32(1, &pThen);
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode((SNode*)pSubSelect);
nodesDestroyNode(pCondFull1);
nodesDestroyNode(pTempVal);
return code;
}
// pSubSelect, pCondFull1, pThen, pTempVal need to free
pWhenThen = NULL;
code = createParWhenThenNode(pCondFull1, pThen, &pWhenThen);
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode((SNode*)pSubSelect);
nodesDestroyNode(pCondFull1);
nodesDestroyNode(pThen);
nodesDestroyNode(pTempVal);
return code;
@ -15289,7 +15329,7 @@ static int32_t rewriteShowAliveStmt(STranslateContext* pCxt, SQuery* pQuery) {
}
// pSubSelect, pWhenThenlist need to free
// case when leader_col = count_col and count_col > 0 then 1 when leader_col < count_col and count_col > 0 then 2 else
// case when leader_col = count_col and leader_col > 0 then 1 when leader_col < count_col and leader_col > 0 then 2 else
// 0 end as status
pElse = NULL;
code = nodesMakeValueNodeFromInt32(0, &pElse);

37
source/libs/scalar/inc/sclvector.h
View File

@ -78,40 +78,41 @@ static FORCE_INLINE int32_t getVectorDoubleValue_BOOL(void *src, int32_t index,
int32_t getVectorDoubleValue_JSON(void *src, int32_t index, double *out);
static FORCE_INLINE _getDoubleValue_fn_t getVectorDoubleValueFn(int32_t srcType) {
_getDoubleValue_fn_t p = NULL;
static FORCE_INLINE int32_t getVectorDoubleValueFn(int32_t srcType, _getDoubleValue_fn_t *p) {
*p = NULL;
if (srcType == TSDB_DATA_TYPE_TINYINT) {
p = getVectorDoubleValue_TINYINT;
*p = getVectorDoubleValue_TINYINT;
} else if (srcType == TSDB_DATA_TYPE_UTINYINT) {
p = getVectorDoubleValue_UTINYINT;
*p = getVectorDoubleValue_UTINYINT;
} else if (srcType == TSDB_DATA_TYPE_SMALLINT) {
p = getVectorDoubleValue_SMALLINT;
*p = getVectorDoubleValue_SMALLINT;
} else if (srcType == TSDB_DATA_TYPE_USMALLINT) {
p = getVectorDoubleValue_USMALLINT;
*p = getVectorDoubleValue_USMALLINT;
} else if (srcType == TSDB_DATA_TYPE_INT) {
p = getVectorDoubleValue_INT;
*p = getVectorDoubleValue_INT;
} else if (srcType == TSDB_DATA_TYPE_UINT) {
p = getVectorDoubleValue_UINT;
*p = getVectorDoubleValue_UINT;
} else if (srcType == TSDB_DATA_TYPE_BIGINT) {
p = getVectorDoubleValue_BIGINT;
*p = getVectorDoubleValue_BIGINT;
} else if (srcType == TSDB_DATA_TYPE_UBIGINT) {
p = getVectorDoubleValue_UBIGINT;
*p = getVectorDoubleValue_UBIGINT;
} else if (srcType == TSDB_DATA_TYPE_FLOAT) {
p = getVectorDoubleValue_FLOAT;
*p = getVectorDoubleValue_FLOAT;
} else if (srcType == TSDB_DATA_TYPE_DOUBLE) {
p = getVectorDoubleValue_DOUBLE;
*p = getVectorDoubleValue_DOUBLE;
} else if (srcType == TSDB_DATA_TYPE_TIMESTAMP) {
p = getVectorDoubleValue_BIGINT;
*p = getVectorDoubleValue_BIGINT;
} else if (srcType == TSDB_DATA_TYPE_JSON) {
p = getVectorDoubleValue_JSON;
*p = getVectorDoubleValue_JSON;
} else if (srcType == TSDB_DATA_TYPE_BOOL) {
p = getVectorDoubleValue_BOOL;
*p = getVectorDoubleValue_BOOL;
} else if (srcType == TSDB_DATA_TYPE_NULL) {
p = NULL;
*p = NULL;
} else {
ASSERT(0);
*p = NULL;
return TSDB_CODE_SCALAR_CONVERT_ERROR;
}
return p;
return TSDB_CODE_SUCCESS;
}
typedef int32_t (*_bufConverteFunc)(char *buf, SScalarParam *pOut, int32_t outType, int32_t *overflow);

172
source/libs/scalar/src/filter.c
View File

@ -92,7 +92,9 @@ rangeCompFunc gRangeCompare[] = {filterRangeCompee, filterRangeCompei, filterRan
int8_t filterGetRangeCompFuncFromOptrs(uint8_t optr, uint8_t optr2) {
if (optr2) {
ASSERT(optr2 == OP_TYPE_LOWER_THAN || optr2 == OP_TYPE_LOWER_EQUAL);
if (optr2 != OP_TYPE_LOWER_THAN && optr2 != OP_TYPE_LOWER_EQUAL) {
return -1;
}
if (optr == OP_TYPE_GREATER_THAN) {
if (optr2 == OP_TYPE_LOWER_THAN) {
@ -763,7 +765,10 @@ int32_t filterAddRangeCtx(void *dst, void *src, int32_t optr) {
SFilterRangeCtx *dctx = (SFilterRangeCtx *)dst;
SFilterRangeCtx *sctx = (SFilterRangeCtx *)src;
ASSERT(optr == LOGIC_COND_TYPE_OR);
if (optr != LOGIC_COND_TYPE_OR) {
fltError("filterAddRangeCtx get invalid optr:%d", optr);
return TSDB_CODE_QRY_FILTER_WRONG_OPTR_TYPE;
}
if (sctx->rs == NULL) {
return TSDB_CODE_SUCCESS;
@ -1204,7 +1209,10 @@ int32_t filterAddUnitImpl(SFilterInfo *info, uint8_t optr, SFilterFieldId *left,
if (u->right.type == FLD_TYPE_VALUE) {
SFilterField *val = FILTER_UNIT_RIGHT_FIELD(info, u);
ASSERT(FILTER_GET_FLAG(val->flag, FLD_TYPE_VALUE));
if (!FILTER_GET_FLAG(val->flag, FLD_TYPE_VALUE)) {
fltError("filterAddUnitImpl get invalid flag : %d in val", val->flag);
return TSDB_CODE_APP_ERROR;
}
} else {
int32_t paramNum = scalarGetOperatorParamNum(optr);
if (1 != paramNum) {
@ -1214,7 +1222,10 @@ int32_t filterAddUnitImpl(SFilterInfo *info, uint8_t optr, SFilterFieldId *left,
}
SFilterField *col = FILTER_UNIT_LEFT_FIELD(info, u);
ASSERT(FILTER_GET_FLAG(col->flag, FLD_TYPE_COLUMN));
if (!FILTER_GET_FLAG(col->flag, FLD_TYPE_COLUMN)) {
fltError("filterAddUnitImpl get invalid flag : %d in col", col->flag);
return TSDB_CODE_APP_ERROR;
}
info->units[info->unitNum].compare.type = FILTER_GET_COL_FIELD_TYPE(col);
info->units[info->unitNum].compare.precision = FILTER_GET_COL_FIELD_PRECISION(col);
@ -1398,29 +1409,48 @@ int32_t filterAddGroupUnitFromCtx(SFilterInfo *dst, SFilterInfo *src, SFilterRan
if (optr == LOGIC_COND_TYPE_AND) {
if (ctx->isnull) {
ASSERT(ctx->notnull == false && ctx->isrange == false);
if (ctx->notnull || ctx->isrange) {
fltError("filterAddGroupUnitFromCtx get invalid ctx : isnull %d, notnull %d, isrange %d",
ctx->isnull, ctx->notnull, ctx->isrange);
FLT_ERR_RET(TSDB_CODE_QRY_FILTER_RANGE_ERROR);
}
FLT_ERR_RET(filterAddUnit(dst, OP_TYPE_IS_NULL, &left, NULL, &uidx));
FLT_ERR_RET(filterAddUnitToGroup(g, uidx));
return TSDB_CODE_SUCCESS;
}
if (ctx->notnull) {
ASSERT(ctx->isnull == false && ctx->isrange == false);
if (ctx->isnull || ctx->isrange) {
fltError("filterAddGroupUnitFromCtx get invalid ctx : isnull %d, notnull %d, isrange %d",
ctx->isnull, ctx->notnull, ctx->isrange);
FLT_ERR_RET(TSDB_CODE_QRY_FILTER_RANGE_ERROR);
}
FLT_ERR_RET(filterAddUnit(dst, OP_TYPE_IS_NOT_NULL, &left, NULL, &uidx));
FLT_ERR_RET(filterAddUnitToGroup(g, uidx));
return TSDB_CODE_SUCCESS;
}
if (!ctx->isrange) {
ASSERT(ctx->isnull || ctx->notnull);
if (!ctx->isnull && !ctx->notnull) {
fltError("filterAddGroupUnitFromCtx get invalid ctx : isnull %d, notnull %d, isrange %d",
ctx->isnull, ctx->notnull, ctx->isrange);
FLT_ERR_RET(TSDB_CODE_QRY_FILTER_RANGE_ERROR);
}
return TSDB_CODE_SUCCESS;
}
ASSERT(ctx->rs && ctx->rs->next == NULL);
if (!ctx->rs || ctx->rs->next != NULL) {
fltError("filterAddGroupUnitFromCtx get invalid range node with rs:%p", ctx->rs);
FLT_ERR_RET(TSDB_CODE_QRY_FILTER_RANGE_ERROR);
}
SFilterRange *ra = &ctx->rs->ra;
ASSERT(!((FILTER_GET_FLAG(ra->sflag, RANGE_FLG_NULL)) && (FILTER_GET_FLAG(ra->eflag, RANGE_FLG_NULL))));
if (((FILTER_GET_FLAG(ra->sflag, RANGE_FLG_NULL)) && (FILTER_GET_FLAG(ra->eflag, RANGE_FLG_NULL)))) {
fltError("filterAddGroupUnitFromCtx get invalid range with sflag:%d, eflag:%d",
FILTER_GET_FLAG(ra->sflag, RANGE_FLG_NULL), FILTER_GET_FLAG(ra->eflag, RANGE_FLG_NULL));
FLT_ERR_RET(TSDB_CODE_QRY_FILTER_RANGE_ERROR);
}
if ((!FILTER_GET_FLAG(ra->sflag, RANGE_FLG_NULL)) && (!FILTER_GET_FLAG(ra->eflag, RANGE_FLG_NULL))) {
__compar_fn_t func = getComparFunc(type, 0);
@ -1489,7 +1519,11 @@ int32_t filterAddGroupUnitFromCtx(SFilterInfo *dst, SFilterInfo *src, SFilterRan
SFilterGroup ng = {0};
g = &ng;
ASSERT(ctx->isnull || ctx->notnull || ctx->isrange);
if (!ctx->isnull && !ctx->notnull && !ctx->isrange) {
fltError("filterAddGroupUnitFromCtx get invalid ctx : isnull %d, notnull %d, isrange %d",
ctx->isnull, ctx->notnull, ctx->isrange);
FLT_ERR_RET(TSDB_CODE_APP_ERROR);
}
if (ctx->isnull) {
FLT_ERR_RET(filterAddUnit(dst, OP_TYPE_IS_NULL, &left, NULL, &uidx));
@ -1500,7 +1534,11 @@ int32_t filterAddGroupUnitFromCtx(SFilterInfo *dst, SFilterInfo *src, SFilterRan
}
if (ctx->notnull) {
ASSERT(!ctx->isrange);
if (ctx->isrange) {
fltError("filterAddGroupUnitFromCtx get invalid ctx : isnull %d, notnull %d, isrange %d",
ctx->isnull, ctx->notnull, ctx->isrange);
FLT_ERR_RET(TSDB_CODE_QRY_FILTER_RANGE_ERROR);
}
(void)memset(g, 0, sizeof(*g));
FLT_ERR_RET(filterAddUnit(dst, OP_TYPE_IS_NOT_NULL, &left, NULL, &uidx));
@ -1511,7 +1549,11 @@ int32_t filterAddGroupUnitFromCtx(SFilterInfo *dst, SFilterInfo *src, SFilterRan
}
if (!ctx->isrange) {
ASSERT(ctx->isnull || ctx->notnull);
if (!ctx->isnull && !ctx->notnull) {
fltError("filterAddGroupUnitFromCtx get invalid ctx : isnull %d, notnull %d, isrange %d",
ctx->isnull, ctx->notnull, ctx->isrange);
FLT_ERR_RET(TSDB_CODE_QRY_FILTER_RANGE_ERROR);
}
g->unitNum = 0;
return TSDB_CODE_SUCCESS;
}
@ -1586,7 +1628,10 @@ int32_t filterAddGroupUnitFromCtx(SFilterInfo *dst, SFilterInfo *src, SFilterRan
FLT_ERR_RET(filterAddUnitToGroup(g, uidx));
}
ASSERT(g->unitNum > 0);
if (g->unitNum <= 0) {
fltError("filterAddGroupUnitFromCtx get invalid filter group unit num %d", g->unitNum);
FLT_ERR_RET(TSDB_CODE_QRY_FILTER_RANGE_ERROR);
}
if (NULL == taosArrayPush(res,g)) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
@ -2066,7 +2111,10 @@ void filterFreeInfo(SFilterInfo *info) {
}
int32_t filterHandleValueExtInfo(SFilterUnit *unit, char extInfo) {
ASSERT(extInfo > 0 || extInfo < 0);
if (extInfo == 0) {
fltError("filterHandleValueExtInfo get invalid extInfo : %d", extInfo);
return TSDB_CODE_APP_ERROR;
}
uint8_t optr = FILTER_UNIT_OPTR(unit);
switch (optr) {
@ -2093,13 +2141,20 @@ int32_t fltInitValFieldData(SFilterInfo *info) {
for (uint32_t i = 0; i < info->unitNum; ++i) {
SFilterUnit *unit = &info->units[i];
if (unit->right.type != FLD_TYPE_VALUE) {
ASSERT(unit->compare.optr == FILTER_DUMMY_EMPTY_OPTR || scalarGetOperatorParamNum(unit->compare.optr) == 1);
if (unit->compare.optr != FILTER_DUMMY_EMPTY_OPTR && scalarGetOperatorParamNum(unit->compare.optr) != 1) {
fltError("filterInitValFieldData get invalid operator param num : %d and invalid compare optr %d",
scalarGetOperatorParamNum(unit->compare.optr), unit->compare.optr);
return TSDB_CODE_APP_ERROR;
}
continue;
}
SFilterField *right = FILTER_UNIT_RIGHT_FIELD(info, unit);
ASSERT(FILTER_GET_FLAG(right->flag, FLD_TYPE_VALUE));
if (!FILTER_GET_FLAG(right->flag, FLD_TYPE_VALUE)) {
fltError("filterInitValFieldData get invalid field flag : %d", right->flag);
return TSDB_CODE_APP_ERROR;
}
uint32_t type = FILTER_UNIT_DATA_TYPE(unit);
int8_t precision = FILTER_UNIT_DATA_PRECISION(unit);
@ -2107,7 +2162,10 @@ int32_t fltInitValFieldData(SFilterInfo *info) {
SValueNode *var = (SValueNode *)fi->desc;
if (var == NULL) {
ASSERT(fi->data != NULL);
if (!fi->data) {
fltError("filterInitValFieldData get invalid field data : NULL");
return TSDB_CODE_APP_ERROR;
}
continue;
}
@ -2257,7 +2315,10 @@ int32_t filterAddUnitRange(SFilterInfo *info, SFilterUnit *u, SFilterRangeCtx *c
FILTER_SET_FLAG(ra.sflag, RANGE_FLG_NULL);
break;
case OP_TYPE_NOT_EQUAL:
ASSERT(type == TSDB_DATA_TYPE_BOOL);
if (type != TSDB_DATA_TYPE_BOOL) {
fltError("filterAddUnitRange get invalid type : %d", type);
return TSDB_CODE_QRY_FILTER_INVALID_TYPE;
}
if (GET_INT8_VAL(val)) {
SIMPLE_COPY_VALUES(&ra.s, &tmp);
SIMPLE_COPY_VALUES(&ra.e, &tmp);
@ -2273,7 +2334,7 @@ int32_t filterAddUnitRange(SFilterInfo *info, SFilterUnit *u, SFilterRangeCtx *c
break;
default:
fltError("unsupported operator type");
return TSDB_CODE_APP_ERROR;
return TSDB_CODE_QRY_FILTER_NOT_SUPPORT_TYPE;
}
FLT_ERR_RET(filterAddRange(ctx, &ra, optr));
@ -2547,8 +2608,11 @@ int32_t filterMergeTwoGroupsImpl(SFilterInfo *info, SFilterRangeCtx **ctx, int32
FLT_ERR_RET(filterReuseRangeCtx(*ctx, type, 0));
}
ASSERT(gRes2->colInfo[cidx].type == RANGE_TYPE_MR_CTX);
ASSERT(gRes1->colInfo[cidx].type == RANGE_TYPE_MR_CTX);
if (gRes2->colInfo[cidx].type != RANGE_TYPE_MR_CTX || gRes1->colInfo[cidx].type != RANGE_TYPE_MR_CTX) {
fltError("filterMergeTwoGroupsImpl get invalid col type : %d and %d",
gRes2->colInfo[cidx].type, gRes1->colInfo[cidx].type);
return TSDB_CODE_QRY_FILTER_NOT_SUPPORT_TYPE;
}
FLT_ERR_RET(filterCopyRangeCtx(*ctx, gRes2->colInfo[cidx].info));
FLT_ERR_RET(filterSourceRangeFromCtx(*ctx, gRes1->colInfo[cidx].info, optr, empty, all));
@ -2588,7 +2652,10 @@ int32_t filterMergeTwoGroups(SFilterInfo *info, SFilterGroupCtx **gRes1, SFilter
continue;
}
ASSERT(idx1 == idx2);
if (idx1 != idx2) {
fltError("filterMergeTwoGroups get invalid idx : %d and %d", idx1, idx2);
FLT_ERR_JRET(TSDB_CODE_APP_ERROR);
}
++merNum;
@ -2644,16 +2711,19 @@ int32_t filterMergeTwoGroups(SFilterInfo *info, SFilterGroupCtx **gRes1, SFilter
}
}
ASSERT(merNum > 0);
SFilterColInfo *colInfo = NULL;
ASSERT(merNum == equal1 || merNum == equal2);
if (merNum == 0 || (equal1 != merNum && equal2 != merNum)) {
fltError("filterMergeTwoGroups get invalid merge num : %d, equal1 : %d, equal2 : %d", merNum, equal1, equal2);
FLT_ERR_JRET(TSDB_CODE_APP_ERROR);
}
filterFreeGroupCtx(*gRes2);
*gRes2 = NULL;
ASSERT(colCtxs && taosArrayGetSize(colCtxs) > 0);
if (!colCtxs || taosArrayGetSize(colCtxs) <= 0) {
fltError("filterMergeTwoGroups get invalid colCtxs with size %zu", taosArrayGetSize(colCtxs));
FLT_ERR_JRET(TSDB_CODE_APP_ERROR);
}
SFilterColInfo *colInfo = NULL;
int32_t ctxSize = (int32_t)taosArrayGetSize(colCtxs);
SFilterColCtx *pctx = NULL;
@ -2713,7 +2783,10 @@ int32_t filterMergeGroups(SFilterInfo *info, SFilterGroupCtx **gRes, int32_t *gR
if (pColNum > 0) {
for (int32_t m = 0; m <= pEnd; ++m) {
for (int32_t n = cStart; n <= cEnd; ++n) {
ASSERT(m < n);
if (m >= n) {
fltError("filterMergeGroups get invalid m : %d and n : %d", m, n);
FLT_ERR_JRET(TSDB_CODE_APP_ERROR);
}
FLT_ERR_JRET(filterMergeTwoGroups(info, &gRes[m], &gRes[n], &all));
FLT_CHK_JMP(all);
@ -2734,7 +2807,10 @@ int32_t filterMergeGroups(SFilterInfo *info, SFilterGroupCtx **gRes, int32_t *gR
for (int32_t m = cStart; m < cEnd; ++m) {
for (int32_t n = m + 1; n <= cEnd; ++n) {
ASSERT(m < n);
if (m >= n) {
fltError("filterMergeGroups get invalid m : %d and n : %d", m, n);
FLT_ERR_JRET(TSDB_CODE_APP_ERROR);
}
FLT_ERR_JRET(filterMergeTwoGroups(info, &gRes[m], &gRes[n], &all));
FLT_CHK_JMP(all);
@ -2844,7 +2920,10 @@ int32_t filterRewrite(SFilterInfo *info, SFilterGroupCtx **gRes, int32_t gResNum
for (uint32_t m = 0; m < res->colNum; ++m) {
colInfo = &res->colInfo[res->colIdx[m]];
if (FILTER_NO_MERGE_DATA_TYPE(colInfo->dataType)) {
ASSERT(colInfo->type == RANGE_TYPE_UNIT);
if (colInfo->type != RANGE_TYPE_UNIT) {
fltError("filterRewrite get invalid col type : %d", colInfo->type);
FLT_ERR_JRET(TSDB_CODE_QRY_FILTER_INVALID_TYPE);
}
int32_t usize = (int32_t)taosArrayGetSize((SArray *)colInfo->info);
for (int32_t n = 0; n < usize; ++n) {
@ -2859,7 +2938,10 @@ int32_t filterRewrite(SFilterInfo *info, SFilterGroupCtx **gRes, int32_t gResNum
continue;
}
ASSERT(colInfo->type == RANGE_TYPE_MR_CTX);
if (colInfo->type != RANGE_TYPE_MR_CTX) {
fltError("filterRewrite get invalid col type : %d", colInfo->type);
FLT_ERR_JRET(TSDB_CODE_QRY_FILTER_INVALID_TYPE);
}
FLT_ERR_JRET(filterAddGroupUnitFromCtx(info, &oinfo, colInfo->info, res->colIdx[m], &ng, optr, group));
}
@ -2905,7 +2987,10 @@ int32_t filterGenerateColRange(SFilterInfo *info, SFilterGroupCtx **gRes, int32_
continue;
}
ASSERT(idxNum[i] == gResNum);
if (idxNum[i] != gResNum) {
fltError("filterGenerateColRange get invalid idxNum : %d and gResNum : %d", idxNum[i], gResNum);
FLT_ERR_JRET(TSDB_CODE_APP_ERROR);
}
if (idxs == NULL) {
idxs = taosMemoryCalloc(info->fields[FLD_TYPE_COLUMN].num, sizeof(*idxs));
@ -2936,7 +3021,10 @@ int32_t filterGenerateColRange(SFilterInfo *info, SFilterGroupCtx **gRes, int32_
continue;
}
ASSERT(res->colIdx[n] == idxs[m]);
if (res->colIdx[n] != idxs[m]) {
fltError("filterGenerateColRange get invalid colIdx : %d and idxs : %d", res->colIdx[n], idxs[m]);
SCL_ERR_JRET(TSDB_CODE_APP_ERROR);
}
SFilterColInfo *colInfo = &res->colInfo[res->colIdx[n]];
if (info->colRange[m] == NULL) {
@ -2945,7 +3033,10 @@ int32_t filterGenerateColRange(SFilterInfo *info, SFilterGroupCtx **gRes, int32_
info->colRange[m]->colId = FILTER_GET_COL_FIELD_ID(fi);
}
ASSERT(colInfo->type == RANGE_TYPE_MR_CTX);
if (colInfo->type != RANGE_TYPE_MR_CTX) {
fltError("filterGenerateColRange get invalid col type : %d", colInfo->type);
FLT_ERR_JRET(TSDB_CODE_QRY_FILTER_INVALID_TYPE);
}
bool all = false;
FLT_ERR_JRET(filterSourceRangeFromCtx(info->colRange[m], colInfo->info, LOGIC_COND_TYPE_OR, NULL, &all));
@ -3195,7 +3286,10 @@ int32_t filterRmUnitByRange(SFilterInfo *info, SColumnDataAgg *pDataStatis, int3
unitIdx = pGroupIdx;
--info->blkGroupNum;
ASSERT(empty || all);
if (!empty && !all) {
fltError("filterRmUnitByRange get invalid empty and all : %d and %d", empty, all);
FLT_ERR_RET(TSDB_CODE_APP_ERROR);
}
if (empty) {
FILTER_SET_FLAG(info->blkFlag, FI_STATUS_BLK_EMPTY);
@ -3303,7 +3397,10 @@ int32_t filterExecuteBasedOnStatis(SFilterInfo *info, int32_t numOfRows, SColumn
goto _return;
}
ASSERT(info->unitNum > 1);
if (info->unitNum <= 1) {
fltError("filterExecuteBasedOnStatis get invalid unit num : %d", info->unitNum);
FLT_ERR_JRET(TSDB_CODE_APP_ERROR);
}
*all = filterExecuteBasedOnStatisImpl(info, numOfRows, p, statis, numOfCols);
goto _return;
@ -5161,7 +5258,6 @@ int32_t filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p,
return TSDB_CODE_SUCCESS;
}
ASSERT(false == info->scalarMode);
*p = output.columnData;
output.numOfRows = pSrc->info.rows;

7
source/libs/scalar/src/sclfunc.c
View File

@ -297,7 +297,8 @@ static int32_t doScalarFunctionUnique(SScalarParam *pInput, int32_t inputNum, SS
SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData;
_getDoubleValue_fn_t getValueFn = getVectorDoubleValueFn(type);
_getDoubleValue_fn_t getValueFn;
SCL_ERR_RET(getVectorDoubleValueFn(type, &getValueFn));
double *out = (double *)pOutputData->pData;
@ -328,7 +329,7 @@ static int32_t doScalarFunctionUnique2(SScalarParam *pInput, int32_t inputNum, S
for (int32_t i = 0; i < inputNum; ++i) {
pInputData[i] = pInput[i].columnData;
getValueFn[i] = getVectorDoubleValueFn(GET_PARAM_TYPE(&pInput[i]));
SCL_ERR_RET(getVectorDoubleValueFn(GET_PARAM_TYPE(&pInput[i]), &getValueFn[i]));
}
double *out = (double *)pOutputData->pData;
@ -2918,7 +2919,7 @@ static int32_t doScalarFunction2(SScalarParam *pInput, int32_t inputNum, SScalar
for (int32_t i = 0; i < inputNum; ++i) {
pInputData[i] = pInput[i].columnData;
getValueFn[i] = getVectorDoubleValueFn(GET_PARAM_TYPE(&pInput[i]));
SCL_ERR_RET(getVectorDoubleValueFn(GET_PARAM_TYPE(&pInput[i]), &getValueFn[i]));
}
bool hasNullType = (IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[0])) || IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[1])));

168
source/libs/scalar/src/sclvector.c
View File

@ -43,7 +43,7 @@ bool noConvertBeforeCompare(int32_t leftType, int32_t rightType, int32_t optr) {
(optr >= OP_TYPE_GREATER_THAN && optr <= OP_TYPE_NOT_EQUAL);
}
void convertNumberToNumber(const void *inData, void *outData, int8_t inType, int8_t outType) {
int32_t convertNumberToNumber(const void *inData, void *outData, int8_t inType, int8_t outType) {
switch (outType) {
case TSDB_DATA_TYPE_BOOL: {
GET_TYPED_DATA(*((bool *)outData), bool, inType, inData);
@ -91,9 +91,10 @@ void convertNumberToNumber(const void *inData, void *outData, int8_t inType, int
break;
}
default: {
ASSERT(0);
return TSDB_CODE_SCALAR_CONVERT_ERROR;
}
}
return TSDB_CODE_SUCCESS;
}
int32_t convertNcharToDouble(const void *inData, void *outData) {
@ -180,7 +181,10 @@ int32_t getVectorBigintValue_BOOL(void *src, int32_t index, int64_t *res) {
}
int32_t getVectorBigintValue_JSON(void *src, int32_t index, int64_t *res) {
ASSERT(!colDataIsNull_var(((SColumnInfoData *)src), index));
if (colDataIsNull_var(((SColumnInfoData *)src), index)) {
sclError("getVectorBigintValue_JSON get json data null with index %d", index);
SCL_ERR_RET(TSDB_CODE_SCALAR_CONVERT_ERROR);
}
char *data = colDataGetVarData((SColumnInfoData *)src, index);
double out = 0;
if (*data == TSDB_DATA_TYPE_NULL) {
@ -192,46 +196,47 @@ int32_t getVectorBigintValue_JSON(void *src, int32_t index, int64_t *res) {
*res = 0;
SCL_ERR_RET(TSDB_CODE_QRY_JSON_NOT_SUPPORT_ERROR);
} else {
convertNumberToNumber(data + CHAR_BYTES, &out, *data, TSDB_DATA_TYPE_DOUBLE);
SCL_ERR_RET(convertNumberToNumber(data + CHAR_BYTES, &out, *data, TSDB_DATA_TYPE_DOUBLE));
}
*res = (int64_t)out;
SCL_RET(TSDB_CODE_SUCCESS);
}
_getBigintValue_fn_t getVectorBigintValueFn(int32_t srcType) {
_getBigintValue_fn_t p = NULL;
int32_t getVectorBigintValueFn(int32_t srcType, _getBigintValue_fn_t *p) {
*p = NULL;
if (srcType == TSDB_DATA_TYPE_TINYINT) {
p = getVectorBigintValue_TINYINT;
*p = getVectorBigintValue_TINYINT;
} else if (srcType == TSDB_DATA_TYPE_UTINYINT) {
p = getVectorBigintValue_UTINYINT;
*p = getVectorBigintValue_UTINYINT;
} else if (srcType == TSDB_DATA_TYPE_SMALLINT) {
p = getVectorBigintValue_SMALLINT;
*p = getVectorBigintValue_SMALLINT;
} else if (srcType == TSDB_DATA_TYPE_USMALLINT) {
p = getVectorBigintValue_USMALLINT;
*p = getVectorBigintValue_USMALLINT;
} else if (srcType == TSDB_DATA_TYPE_INT) {
p = getVectorBigintValue_INT;
*p = getVectorBigintValue_INT;
} else if (srcType == TSDB_DATA_TYPE_UINT) {
p = getVectorBigintValue_UINT;
*p = getVectorBigintValue_UINT;
} else if (srcType == TSDB_DATA_TYPE_BIGINT) {
p = getVectorBigintValue_BIGINT;
*p = getVectorBigintValue_BIGINT;
} else if (srcType == TSDB_DATA_TYPE_UBIGINT) {
p = getVectorBigintValue_UBIGINT;
*p = getVectorBigintValue_UBIGINT;
} else if (srcType == TSDB_DATA_TYPE_FLOAT) {
p = getVectorBigintValue_FLOAT;
*p = getVectorBigintValue_FLOAT;
} else if (srcType == TSDB_DATA_TYPE_DOUBLE) {
p = getVectorBigintValue_DOUBLE;
*p = getVectorBigintValue_DOUBLE;
} else if (srcType == TSDB_DATA_TYPE_TIMESTAMP) {
p = getVectorBigintValue_BIGINT;
*p = getVectorBigintValue_BIGINT;
} else if (srcType == TSDB_DATA_TYPE_BOOL) {
p = getVectorBigintValue_BOOL;
*p = getVectorBigintValue_BOOL;
} else if (srcType == TSDB_DATA_TYPE_JSON) {
p = getVectorBigintValue_JSON;
*p = getVectorBigintValue_JSON;
} else if (srcType == TSDB_DATA_TYPE_NULL) {
p = NULL;
*p = NULL;
} else {
ASSERT(0);
sclError("getVectorBigintValueFn invalid srcType : %d", srcType);
return TSDB_CODE_SCALAR_CONVERT_ERROR;
}
return p;
return TSDB_CODE_SUCCESS;
}
static FORCE_INLINE int32_t varToTimestamp(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
@ -467,7 +472,7 @@ static FORCE_INLINE int32_t varToGeometry(char *buf, SScalarParam *pOut, int32_t
_return:
taosMemoryFree(output);
geosFreeBuffer(t);
ASSERT(t == NULL && len == 0);
t = NULL;
VarDataLenT dummyHeader = 0;
SCL_ERR_RET(colDataSetVal(pOut->columnData, rowIndex, (const char *)&dummyHeader, false));
SCL_RET(code);
@ -525,7 +530,7 @@ int32_t vectorConvertFromVarData(SSclVectorConvCtx *pCtx, int32_t *overflow) {
} else if (tTagIsJson(data) || *data == TSDB_DATA_TYPE_NULL) {
SCL_ERR_JRET(TSDB_CODE_QRY_JSON_NOT_SUPPORT_ERROR);
} else {
convertNumberToNumber(data + CHAR_BYTES, colDataGetNumData(pCtx->pOut->columnData, i), *data, pCtx->outType);
SCL_ERR_JRET(convertNumberToNumber(data + CHAR_BYTES, colDataGetNumData(pCtx->pOut->columnData, i), *data, pCtx->outType));
continue;
}
}
@ -582,7 +587,7 @@ int32_t getVectorDoubleValue_JSON(void *src, int32_t index, double *out) {
} else if (tTagIsJson(data)) {
SCL_ERR_RET(TSDB_CODE_QRY_JSON_NOT_SUPPORT_ERROR);
} else {
convertNumberToNumber(data + CHAR_BYTES, out, *data, TSDB_DATA_TYPE_DOUBLE);
SCL_ERR_RET(convertNumberToNumber(data + CHAR_BYTES, out, *data, TSDB_DATA_TYPE_DOUBLE));
}
SCL_RET(TSDB_CODE_SUCCESS);
}
@ -673,7 +678,7 @@ int32_t convertJsonValue(__compar_fn_t *fp, int32_t optr, int8_t typeLeft, int8_
*result = false;
return TSDB_CODE_SUCCESS;
} else if (typeLeft != type) {
convertNumberToNumber(*pLeftData, pLeftOut, typeLeft, type);
SCL_ERR_RET(convertNumberToNumber(*pLeftData, pLeftOut, typeLeft, type));
*pLeftData = pLeftOut;
}
@ -683,7 +688,7 @@ int32_t convertJsonValue(__compar_fn_t *fp, int32_t optr, int8_t typeLeft, int8_
*result = false;
return TSDB_CODE_SUCCESS;
} else if (typeRight != type) {
convertNumberToNumber(*pRightData, pRightOut, typeRight, type);
SCL_ERR_RET(convertNumberToNumber(*pRightData, pRightOut, typeRight, type));
*pRightData = pRightOut;
}
} else if (type == TSDB_DATA_TYPE_BINARY ||
@ -1130,8 +1135,10 @@ enum {
// TODO not correct for descending order scan
static int32_t vectorMathAddHelper(SColumnInfoData *pLeftCol, SColumnInfoData *pRightCol, SColumnInfoData *pOutputCol,
int32_t numOfRows, int32_t step, int32_t i) {
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
_getDoubleValue_fn_t getVectorDoubleValueFnRight;
SCL_ERR_RET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
SCL_ERR_RET(getVectorDoubleValueFn(pRightCol->info.type, &getVectorDoubleValueFnRight));
double *output = (double *)pOutputCol->pData;
@ -1155,9 +1162,10 @@ static int32_t vectorMathAddHelper(SColumnInfoData *pLeftCol, SColumnInfoData *p
static int32_t vectorMathTsAddHelper(SColumnInfoData *pLeftCol, SColumnInfoData *pRightCol, SColumnInfoData *pOutputCol,
int32_t numOfRows, int32_t step, int32_t i) {
_getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnLeft;
_getBigintValue_fn_t getVectorBigintValueFnRight;
SCL_ERR_RET(getVectorBigintValueFn(pLeftCol->info.type, &getVectorBigintValueFnLeft));
SCL_ERR_RET(getVectorBigintValueFn(pRightCol->info.type, &getVectorBigintValueFnRight));
int64_t *output = (int64_t *)pOutputCol->pData;
if (IS_HELPER_NULL(pRightCol, 0)) { // Set pLeft->numOfRows NULL value
@ -1230,8 +1238,10 @@ int32_t vectorMathAdd(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *p
(GET_PARAM_TYPE(pRight) == TSDB_DATA_TYPE_TIMESTAMP &&
GET_PARAM_TYPE(pLeft) == TSDB_DATA_TYPE_BOOL)) { // timestamp plus duration
int64_t *output = (int64_t *)pOutputCol->pData;
_getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnLeft;
_getBigintValue_fn_t getVectorBigintValueFnRight;
SCL_ERR_JRET(getVectorBigintValueFn(pLeftCol->info.type, &getVectorBigintValueFnLeft));
SCL_ERR_JRET(getVectorBigintValueFn(pRightCol->info.type, &getVectorBigintValueFnRight));
if (pLeft->numOfRows == 1 && pRight->numOfRows == 1) {
if (GET_PARAM_TYPE(pLeft) == TSDB_DATA_TYPE_TIMESTAMP) {
@ -1258,9 +1268,10 @@ int32_t vectorMathAdd(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *p
}
} else {
double *output = (double *)pOutputCol->pData;
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
_getDoubleValue_fn_t getVectorDoubleValueFnRight;
SCL_ERR_JRET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
SCL_ERR_JRET(getVectorDoubleValueFn(pRightCol->info.type, &getVectorDoubleValueFnRight));
if (pLeft->numOfRows == pRight->numOfRows) {
for (; i < pRight->numOfRows && i >= 0; i += step, output += 1) {
if (IS_NULL) {
@ -1289,8 +1300,10 @@ _return:
// TODO not correct for descending order scan
static int32_t vectorMathSubHelper(SColumnInfoData *pLeftCol, SColumnInfoData *pRightCol, SColumnInfoData *pOutputCol,
int32_t numOfRows, int32_t step, int32_t factor, int32_t i) {
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
_getDoubleValue_fn_t getVectorDoubleValueFnRight;
SCL_ERR_RET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
SCL_ERR_RET(getVectorDoubleValueFn(pRightCol->info.type, &getVectorDoubleValueFnRight));
double *output = (double *)pOutputCol->pData;
@ -1314,8 +1327,10 @@ static int32_t vectorMathSubHelper(SColumnInfoData *pLeftCol, SColumnInfoData *p
static int32_t vectorMathTsSubHelper(SColumnInfoData *pLeftCol, SColumnInfoData *pRightCol, SColumnInfoData *pOutputCol,
int32_t numOfRows, int32_t step, int32_t factor, int32_t i) {
_getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnLeft;
_getBigintValue_fn_t getVectorBigintValueFnRight;
SCL_ERR_RET(getVectorBigintValueFn(pLeftCol->info.type, &getVectorBigintValueFnLeft));
SCL_ERR_RET(getVectorBigintValueFn(pRightCol->info.type, &getVectorBigintValueFnRight));
int64_t *output = (int64_t *)pOutputCol->pData;
@ -1357,8 +1372,10 @@ int32_t vectorMathSub(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *p
(GET_PARAM_TYPE(pRight) == TSDB_DATA_TYPE_TIMESTAMP &&
GET_PARAM_TYPE(pLeft) == TSDB_DATA_TYPE_BIGINT)) { // timestamp minus duration
int64_t *output = (int64_t *)pOutputCol->pData;
_getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnLeft;
_getBigintValue_fn_t getVectorBigintValueFnRight;
SCL_ERR_JRET(getVectorBigintValueFn(pLeftCol->info.type, &getVectorBigintValueFnLeft));
SCL_ERR_JRET(getVectorBigintValueFn(pRightCol->info.type, &getVectorBigintValueFnRight));
if (pLeft->numOfRows == 1 && pRight->numOfRows == 1) {
SCL_ERR_JRET(vectorMathTsSubHelper(pLeftCol, pRightCol, pOutputCol, pLeft->numOfRows, step, 1, i));
@ -1381,8 +1398,10 @@ int32_t vectorMathSub(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *p
}
} else {
double *output = (double *)pOutputCol->pData;
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
_getDoubleValue_fn_t getVectorDoubleValueFnRight;
SCL_ERR_JRET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
SCL_ERR_JRET(getVectorDoubleValueFn(pRightCol->info.type, &getVectorDoubleValueFnRight));
if (pLeft->numOfRows == pRight->numOfRows) {
for (; i < pRight->numOfRows && i >= 0; i += step, output += 1) {
@ -1412,8 +1431,10 @@ _return:
// TODO not correct for descending order scan
static int32_t vectorMathMultiplyHelper(SColumnInfoData *pLeftCol, SColumnInfoData *pRightCol, SColumnInfoData *pOutputCol,
int32_t numOfRows, int32_t step, int32_t i) {
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
_getDoubleValue_fn_t getVectorDoubleValueFnRight;
SCL_ERR_RET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
SCL_ERR_RET(getVectorDoubleValueFn(pRightCol->info.type, &getVectorDoubleValueFnRight));
double *output = (double *)pOutputCol->pData;
@ -1449,8 +1470,10 @@ int32_t vectorMathMultiply(SScalarParam *pLeft, SScalarParam *pRight, SScalarPar
SCL_ERR_JRET(vectorConvertVarToDouble(pLeft, &leftConvert, &pLeftCol));
SCL_ERR_JRET(vectorConvertVarToDouble(pRight, &rightConvert, &pRightCol));
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
_getDoubleValue_fn_t getVectorDoubleValueFnRight;
SCL_ERR_JRET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
SCL_ERR_JRET(getVectorDoubleValueFn(pRightCol->info.type, &getVectorDoubleValueFnRight));
double *output = (double *)pOutputCol->pData;
if (pLeft->numOfRows == pRight->numOfRows) {
@ -1491,8 +1514,10 @@ int32_t vectorMathDivide(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam
SCL_ERR_JRET(vectorConvertVarToDouble(pLeft, &leftConvert, &pLeftCol));
SCL_ERR_JRET(vectorConvertVarToDouble(pRight, &rightConvert, &pRightCol));
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
_getDoubleValue_fn_t getVectorDoubleValueFnRight;
SCL_ERR_JRET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
SCL_ERR_JRET(getVectorDoubleValueFn(pRightCol->info.type, &getVectorDoubleValueFnRight));
double *output = (double *)pOutputCol->pData;
if (pLeft->numOfRows == pRight->numOfRows) {
@ -1573,8 +1598,10 @@ int32_t vectorMathRemainder(SScalarParam *pLeft, SScalarParam *pRight, SScalarPa
SCL_ERR_JRET(vectorConvertVarToDouble(pLeft, &leftConvert, &pLeftCol));
SCL_ERR_JRET(vectorConvertVarToDouble(pRight, &rightConvert, &pRightCol));
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
_getDoubleValue_fn_t getVectorDoubleValueFnRight;
SCL_ERR_JRET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
SCL_ERR_JRET(getVectorDoubleValueFn(pRightCol->info.type, &getVectorDoubleValueFnRight));
double *output = (double *)pOutputCol->pData;
@ -1661,7 +1688,8 @@ int32_t vectorMathMinus(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam
SColumnInfoData *pLeftCol = NULL;
SCL_ERR_JRET(vectorConvertVarToDouble(pLeft, &leftConvert, &pLeftCol));
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeftCol->info.type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft;
SCL_ERR_JRET(getVectorDoubleValueFn(pLeftCol->info.type, &getVectorDoubleValueFnLeft));
double *output = (double *)pOutputCol->pData;
for (; i < pLeft->numOfRows && i >= 0; i += step, output += 1) {
@ -1692,15 +1720,20 @@ int32_t vectorAssign(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *pO
}
}
ASSERT(pRight->numOfQualified == 1 || pRight->numOfQualified == 0);
if (pRight->numOfQualified != 1 && pRight->numOfQualified != 0) {
sclError("vectorAssign: invalid qualified number %d", pRight->numOfQualified);
SCL_ERR_RET(TSDB_CODE_APP_ERROR);
}
pOut->numOfQualified = pRight->numOfQualified * pOut->numOfRows;
return TSDB_CODE_SUCCESS;
}
static int32_t vectorBitAndHelper(SColumnInfoData *pLeftCol, SColumnInfoData *pRightCol, SColumnInfoData *pOutputCol,
int32_t numOfRows, int32_t step, int32_t i) {
_getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnLeft;
_getBigintValue_fn_t getVectorBigintValueFnRight;
SCL_ERR_RET(getVectorBigintValueFn(pLeftCol->info.type, &getVectorBigintValueFnLeft));
SCL_ERR_RET(getVectorBigintValueFn(pRightCol->info.type, &getVectorBigintValueFnRight));
int64_t *output = (int64_t *)pOutputCol->pData;
@ -1736,8 +1769,10 @@ int32_t vectorBitAnd(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *pO
SCL_ERR_JRET(vectorConvertVarToDouble(pLeft, &leftConvert, &pLeftCol));
SCL_ERR_JRET(vectorConvertVarToDouble(pRight, &rightConvert, &pRightCol));
_getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnLeft;
_getBigintValue_fn_t getVectorBigintValueFnRight;
SCL_ERR_JRET(getVectorBigintValueFn(pLeftCol->info.type, &getVectorBigintValueFnLeft));
SCL_ERR_JRET(getVectorBigintValueFn(pRightCol->info.type, &getVectorBigintValueFnRight));
int64_t *output = (int64_t *)pOutputCol->pData;
if (pLeft->numOfRows == pRight->numOfRows) {
@ -1766,8 +1801,10 @@ _return:
static int32_t vectorBitOrHelper(SColumnInfoData *pLeftCol, SColumnInfoData *pRightCol, SColumnInfoData *pOutputCol,
int32_t numOfRows, int32_t step, int32_t i) {
_getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnLeft;
_getBigintValue_fn_t getVectorBigintValueFnRight;
SCL_ERR_RET(getVectorBigintValueFn(pLeftCol->info.type, &getVectorBigintValueFnLeft));
SCL_ERR_RET(getVectorBigintValueFn(pRightCol->info.type, &getVectorBigintValueFnRight));
int64_t *output = (int64_t *)pOutputCol->pData;
@ -1803,8 +1840,10 @@ int32_t vectorBitOr(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *pOu
SCL_ERR_JRET(vectorConvertVarToDouble(pLeft, &leftConvert, &pLeftCol));
SCL_ERR_JRET(vectorConvertVarToDouble(pRight, &rightConvert, &pRightCol));
_getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
_getBigintValue_fn_t getVectorBigintValueFnLeft;
_getBigintValue_fn_t getVectorBigintValueFnRight;
SCL_ERR_JRET(getVectorBigintValueFn(pLeftCol->info.type, &getVectorBigintValueFnLeft));
SCL_ERR_JRET(getVectorBigintValueFn(pRightCol->info.type, &getVectorBigintValueFnRight));
int64_t *output = (int64_t *)pOutputCol->pData;
if (pLeft->numOfRows == pRight->numOfRows) {
@ -1892,7 +1931,8 @@ int32_t doVectorCompareImpl(SScalarParam *pLeft, SScalarParam *pRight, SScalarPa
&leftOut, &rightOut, &isJsonnull, &freeLeft, &freeRight, &result));
if (isJsonnull) {
ASSERT(0);
sclError("doVectorCompareImpl: invalid json null value");
SCL_ERR_RET(TSDB_CODE_APP_ERROR);
}
if (!pLeftData || !pRightData) {

2
source/libs/scalar/test/filter/CMakeLists.txt
View File

@ -1,7 +1,7 @@
MESSAGE(STATUS "build filter unit test")
IF(NOT TD_DARWIN)
IF(TD_DARWIN)
# GoogleTest requires at least C++11
SET(CMAKE_CXX_STANDARD 11)
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)

3
source/libs/scalar/test/filter/filterTests.cpp
View File

@ -113,8 +113,7 @@ int32_t flttMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType,
if (NULL == *block) {
SSDataBlock *res = NULL;
int32_t code = createDataBlock(&res);
ASSERT(code == 0);
FLT_ERR_RET(createDataBlock(&res));
for (int32_t i = 0; i < 2; ++i) {
SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_NULL, 10, 1 + i);

14
source/libs/scalar/test/scalar/scalarTests.cpp
View File

@ -115,7 +115,10 @@ int32_t scltAppendReservedSlot(SArray *pBlockList, int16_t *dataBlockId, int16_t
res->info.capacity = rows;
res->info.rows = rows;
SColumnInfoData *p = static_cast<SColumnInfoData *>(taosArrayGet(res->pDataBlock, 0));
ASSERT(p->pData != NULL && p->nullbitmap != NULL);
if (p->pData == NULL || p->nullbitmap == NULL) {
sclError("data block is not initialized since pData or nullbitmap is NULL");
SCL_ERR_RET(TSDB_CODE_APP_ERROR);
}
(void)taosArrayPush(pBlockList, &res);
*dataBlockId = taosArrayGetSize(pBlockList) - 1;
@ -189,8 +192,7 @@ int32_t scltMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType,
if (NULL == *block) {
SSDataBlock *res = NULL;
int32_t code = createDataBlock(&res);
ASSERT(code == 0);
SCL_ERR_RET(createDataBlock(&res));
for (int32_t i = 0; i < 2; ++i) {
SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_INT, 10, i + 1);
@ -1422,8 +1424,10 @@ int32_t makeCalculate(void *json, void *key, int32_t rightType, void *rightData,
opType == OP_TYPE_NMATCH) {
(void)printf("op:%s,3result:%d,except:%f\n", operatorTypeStr(opType), *((bool *)colDataGetData(column, 0)),
exceptValue);
assert(*(bool *)colDataGetData(column, 0) == exceptValue);
// ASSERT_EQ((int) *((bool *)colDataGetData(column, 0)), (int)exceptValue);
if(*(bool *)colDataGetData(column, 0) != exceptValue) {
(void)printf("expect value %d, but got %d\n", *((bool *)colDataGetData(column, 0)), exceptValue);
SCL_ERR_RET(TSDB_CODE_FAILED);
}
}
taosArrayDestroyEx(blockList, scltFreeDataBlock);

4
source/libs/stream/src/streamMeta.c
View File

@ -1260,7 +1260,9 @@ void streamMetaUpdateStageRole(SStreamMeta* pMeta, int64_t stage, bool isLeader)
pMeta->stage = stage;
// mark the sign to send msg before close all tasks
if ((!isLeader) && (pMeta->role == NODE_ROLE_LEADER)) {
// 1. for leader vnode, always send msg before closing
// 2. for follower vnode, if it's is changed from leader, also sending msg before closing.
if (pMeta->role == NODE_ROLE_LEADER) {
pMeta->sendMsgBeforeClosing = true;
}

5
source/util/src/terror.c
View File

@ -478,6 +478,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_WINDOW_CONDITION, "The time pseudo colum
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, "Executor internal error")
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_JOIN_CONDITION, "Not supported join on condition")
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_FILTER_NOT_SUPPORT_TYPE, "Not supported range type")
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_FILTER_WRONG_OPTR_TYPE, "Wrong operator type")
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_FILTER_RANGE_ERROR, "Wrong filter range")
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_FILTER_INVALID_TYPE, "Invalid filter type")
// grant
TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED, "License expired")
@ -724,6 +727,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_TIME_UNIT_TOO_SMALL, "Function time unit c
TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_INVALID_VALUE_RANGE, "Function got invalid value range")
TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_SETUP_ERROR, "Function set up failed")
TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_INVALID_RES_LENGTH, "Function result exceed max length")
TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_HISTOGRAM_ERROR, "Function failed to calculate histogram")
TAOS_DEFINE_ERROR(TSDB_CODE_FUNC_PERCENTILE_ERROR, "Function failed to calculate percentile")
//udf
TAOS_DEFINE_ERROR(TSDB_CODE_UDF_STOPPING, "udf is stopping")

44
tests/script/tsim/show/showalive.sim
View File

@ -21,6 +21,30 @@ sql create dnode $hostname4 port 7500
sleep 1000
$loop_count = 0
loop00:
sleep 1000
$loop_count = $loop_count + 1
if $loop_count == 20 then
return -1
endi
print 0 show cluster alive;
sql show cluster alive;
print res------------------------
print $data00 $data01
print $data10 $data11
if $data00 != 1 then
print =====data00=$data00
goto loop00
endi
print =============== create database, stable, table
sql create database test vgroups 6;
sql use test;
@ -46,6 +70,15 @@ endi
print show cluster alive;
sql show cluster alive;
print res------------------------
print $data00 $data01
print $data10 $data11
if $rows != 1 then
print =====rows=$rows
goto loop0
endi
if $data00 != 1 then
print =====data00=$data00
goto loop0
@ -54,6 +87,15 @@ endi
print show test.alive;
sql show test.alive;
print res------------------------
print $data00 $data01
print $data10 $data11
if $rows != 1 then
print =====rows=$rows
goto loop0
endi
if $data00 != 1 then
print =====data00=$data00
goto loop0
@ -164,4 +206,4 @@ endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT
system sh/exec.sh -n dnode3 -s stop -x SIGINT
system sh/exec.sh -n dnode4 -s stop -x SIGINT
system sh/exec.sh -n dnode4 -s stop -x SIGINT