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

Merge pull request #26598 from taosdata/enh/TD-30998 

Enh/td 30998
This commit is contained in:
dapan1121 2024-07-19 13:41:08 +08:00 committed by GitHub
parent 5a2093837f e1bcd84426
commit fc1753c1f1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 3067 additions and 1624 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/libs/scalar/filter.h
View File

@ -58,12 +58,13 @@ extern int32_t filterGetTimeRange(SNode *pNode, STimeWindow *win, bool *isStrict
extern int32_t filterConverNcharColumns(SFilterInfo *pFilterInfo, int32_t rows, bool *gotNchar); extern int32_t filterConverNcharColumns(SFilterInfo *pFilterInfo, int32_t rows, bool *gotNchar);
extern int32_t filterFreeNcharColumns(SFilterInfo *pFilterInfo); extern int32_t filterFreeNcharColumns(SFilterInfo *pFilterInfo);
extern void filterFreeInfo(SFilterInfo *info); extern void filterFreeInfo(SFilterInfo *info);
extern bool filterRangeExecute(SFilterInfo *info, SColumnDataAgg *pColsAgg, int32_t numOfCols, int32_t numOfRows); extern int32_t filterRangeExecute(SFilterInfo *info, SColumnDataAgg *pDataStatis, int32_t numOfCols, int32_t numOfRows,
bool *keep);
/* condition split interface */ /* condition split interface */
int32_t filterPartitionCond(SNode **pCondition, SNode **pPrimaryKeyCond, SNode **pTagIndexCond, SNode **pTagCond, int32_t filterPartitionCond(SNode **pCondition, SNode **pPrimaryKeyCond, SNode **pTagIndexCond, SNode **pTagCond,
SNode **pOtherCond); SNode **pOtherCond);
bool filterIsMultiTableColsCond(SNode *pCond); int32_t filterIsMultiTableColsCond(SNode *pCond, bool *res);
EConditionType filterClassifyCondition(SNode *pNode); EConditionType filterClassifyCondition(SNode *pNode);
#ifdef __cplusplus #ifdef __cplusplus

1
include/util/taoserror.h
View File

@ -597,6 +597,7 @@ int32_t taosGetErrSize();
#define TSDB_CODE_QRY_GEO_NOT_SUPPORT_ERROR TAOS_DEF_ERROR_CODE(0, 0x0731) #define TSDB_CODE_QRY_GEO_NOT_SUPPORT_ERROR TAOS_DEF_ERROR_CODE(0, 0x0731)
#define TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x0732) #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_INVALID_JOIN_CONDITION TAOS_DEF_ERROR_CODE(0, 0x0733)
#define TSDB_CODE_QRY_FILTER_NOT_SUPPORT_TYPE TAOS_DEF_ERROR_CODE(0, 0x0734)
// grant // grant
#define TSDB_CODE_GRANT_EXPIRED TAOS_DEF_ERROR_CODE(0, 0x0800) #define TSDB_CODE_GRANT_EXPIRED TAOS_DEF_ERROR_CODE(0, 0x0800)

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

@ -220,14 +220,14 @@ static int32_t doDynamicPruneDataBlock(SOperatorInfo* pOperator, SDataBlockInfo*
return code; return code;
} }
static bool doFilterByBlockSMA(SFilterInfo* pFilterInfo, SColumnDataAgg* pColsAgg, int32_t numOfCols, static int32_t doFilterByBlockSMA(SFilterInfo* pFilterInfo, SColumnDataAgg* pColsAgg, int32_t numOfCols,
int32_t numOfRows) { int32_t numOfRows, bool *keep) {
if (pColsAgg == NULL || pFilterInfo == NULL) { if (pColsAgg == NULL || pFilterInfo == NULL) {
return true; *keep = true;
return TSDB_CODE_SUCCESS;
} }
bool keep = filterRangeExecute(pFilterInfo, pColsAgg, numOfCols, numOfRows); return filterRangeExecute(pFilterInfo, pColsAgg, numOfCols, numOfRows, keep);
return keep;
} }
static bool doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) { static bool doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) {
@ -351,7 +351,11 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
bool success = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo); bool success = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo);
if (success) { if (success) {
size_t size = taosArrayGetSize(pBlock->pDataBlock); size_t size = taosArrayGetSize(pBlock->pDataBlock);
bool keep = doFilterByBlockSMA(pOperator->exprSupp.pFilterInfo, pBlock->pBlockAgg, size, pBlockInfo->rows); bool keep = false;
int32_t code = doFilterByBlockSMA(pOperator->exprSupp.pFilterInfo, pBlock->pBlockAgg, size, pBlockInfo->rows, &keep);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
if (!keep) { if (!keep) {
qDebug("%s data block filter out by block SMA, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64, qDebug("%s data block filter out by block SMA, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64,
GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);

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

@ -143,12 +143,24 @@ int32_t executeMakePointFunc(SColumnInfoData *pInputData[], int32_t iLeft, int32
getDoubleValueFn[1]= getVectorDoubleValueFn(pInputData[1]->info.type); getDoubleValueFn[1]= getVectorDoubleValueFn(pInputData[1]->info.type);
unsigned char *output = NULL; unsigned char *output = NULL;
code = doMakePointFunc(getDoubleValueFn[0](pInputData[0]->pData, iLeft), getDoubleValueFn[1](pInputData[1]->pData, iRight), &output); double leftRes = 0;
if (code != TSDB_CODE_SUCCESS) { double rightRes = 0;
code = getDoubleValueFn[0](pInputData[0]->pData, iLeft, &leftRes);
if (TSDB_CODE_SUCCESS != code) {
goto _exit;
}
code = getDoubleValueFn[1](pInputData[1]->pData, iRight, &rightRes);
if (TSDB_CODE_SUCCESS != code) {
goto _exit;
}
code = doMakePointFunc(leftRes, rightRes, &output);
if (TSDB_CODE_SUCCESS != code) {
goto _exit;
}
code = colDataSetVal(pOutputData, TMAX(iLeft, iRight), output, (output == NULL));
if (TSDB_CODE_SUCCESS != code) {
goto _exit; goto _exit;
} }
colDataSetVal(pOutputData, TMAX(iLeft, iRight), output, (output == NULL));
_exit: _exit:
if (output) { if (output) {

15
source/libs/planner/src/planOptimizer.c
View File

@ -5045,7 +5045,13 @@ static int32_t hashJoinOptSplitPrimFromLogicCond(SNode **pCondition, SNode **pPr
SNodeList *pPrimaryKeyConds = NULL; SNodeList *pPrimaryKeyConds = NULL;
SNode *pCond = NULL; SNode *pCond = NULL;
WHERE_EACH(pCond, pLogicCond->pParameterList) { WHERE_EACH(pCond, pLogicCond->pParameterList) {
if (filterIsMultiTableColsCond(pCond) || COND_TYPE_PRIMARY_KEY != filterClassifyCondition(pCond)) { bool result = false;
code = filterIsMultiTableColsCond(pCond, &result);
if (TSDB_CODE_SUCCESS != code) {
break;
}
if (result || COND_TYPE_PRIMARY_KEY != filterClassifyCondition(pCond)) {
WHERE_NEXT; WHERE_NEXT;
continue; continue;
} }
@ -5089,7 +5095,12 @@ int32_t hashJoinOptSplitPrimCond(SNode **pCondition, SNode **pPrimaryKeyCond) {
} }
bool needOutput = false; bool needOutput = false;
if (filterIsMultiTableColsCond(*pCondition)) { bool result = false;
int32_t code = filterIsMultiTableColsCond(*pCondition, &result);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
if (result) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

22
source/libs/scalar/inc/filterInt.h
View File

@ -98,7 +98,7 @@ typedef struct SFilterRange {
typedef bool (*rangeCompFunc)(const void *, const void *, const void *, const void *, __compar_fn_t); typedef bool (*rangeCompFunc)(const void *, const void *, const void *, const void *, __compar_fn_t);
typedef int32_t (*filter_desc_compare_func)(const void *, const void *); typedef int32_t (*filter_desc_compare_func)(const void *, const void *);
typedef bool (*filter_exec_func)(void *, int32_t, SColumnInfoData *, SColumnDataAgg *, int16_t, int32_t *); typedef int32_t (*filter_exec_func)(void *, int32_t, SColumnInfoData *, SColumnDataAgg *, int16_t, int32_t *, bool *);
typedef int32_t (*filer_get_col_from_name)(void *, int32_t, char *, void **); typedef int32_t (*filer_get_col_from_name)(void *, int32_t, char *, void **);
typedef struct SFilterDataInfo { typedef struct SFilterDataInfo {
@ -363,7 +363,8 @@ struct SFilterInfo {
} while (0) } while (0)
#define INSERT_RANGE(ctx, r, ra) \ #define INSERT_RANGE(ctx, r, ra) \
do { \ do { \
SFilterRangeNode *n = filterNewRange(ctx, ra); \ SFilterRangeNode *n = NULL; \
FLT_ERR_RET(filterNewRange(ctx, ra, &n)); \
n->prev = (r)->prev; \ n->prev = (r)->prev; \
if ((r)->prev) { \ if ((r)->prev) { \
(r)->prev->next = n; \ (r)->prev->next = n; \
@ -375,7 +376,8 @@ struct SFilterInfo {
} while (0) } while (0)
#define APPEND_RANGE(ctx, r, ra) \ #define APPEND_RANGE(ctx, r, ra) \
do { \ do { \
SFilterRangeNode *n = filterNewRange(ctx, ra); \ SFilterRangeNode *n = NULL; \
FLT_ERR_RET(filterNewRange(ctx, ra, &n)); \
n->prev = (r); \ n->prev = (r); \
if (r) { \ if (r) { \
(r)->next = n; \ (r)->next = n; \
@ -458,11 +460,13 @@ struct SFilterInfo {
#define FILTER_UNIT_GET_R(i, idx) ((i)->unitRes[idx]) #define FILTER_UNIT_GET_R(i, idx) ((i)->unitRes[idx])
#define FILTER_UNIT_SET_R(i, idx, v) (i)->unitRes[idx] = (v) #define FILTER_UNIT_SET_R(i, idx, v) (i)->unitRes[idx] = (v)
#define FILTER_PUSH_UNIT(colInfo, u) \ #define FILTER_PUSH_UNIT(colInfo, u) \
do { \ do { \
(colInfo).type = RANGE_TYPE_UNIT; \ (colInfo).type = RANGE_TYPE_UNIT; \
(colInfo).dataType = FILTER_UNIT_DATA_TYPE(u); \ (colInfo).dataType = FILTER_UNIT_DATA_TYPE(u); \
taosArrayPush((SArray *)((colInfo).info), &u); \ if (taosArrayPush((SArray *)((colInfo).info), &u) == NULL) { \
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); \
} \
} while (0) } while (0)
#define FILTER_PUSH_VAR_HASH(colInfo, ha) \ #define FILTER_PUSH_VAR_HASH(colInfo, ha) \
do { \ do { \
@ -494,7 +498,7 @@ struct SFilterInfo {
#define FILTER_EMPTY_RES(i) FILTER_GET_FLAG((i)->status, FI_STATUS_EMPTY) #define FILTER_EMPTY_RES(i) FILTER_GET_FLAG((i)->status, FI_STATUS_EMPTY)
extern bool filterDoCompare(__compar_fn_t func, uint8_t optr, void *left, void *right); extern bool filterDoCompare(__compar_fn_t func, uint8_t optr, void *left, void *right);
extern __compar_fn_t filterGetCompFunc(int32_t type, int32_t optr); extern int32_t filterGetCompFunc(__compar_fn_t *func, int32_t type, int32_t optr);
extern __compar_fn_t filterGetCompFuncEx(int32_t lType, int32_t rType, int32_t optr); extern __compar_fn_t filterGetCompFuncEx(int32_t lType, int32_t rType, int32_t optr);
#ifdef __cplusplus #ifdef __cplusplus

8
source/libs/scalar/inc/sclInt.h
View File

@ -95,11 +95,11 @@ int32_t sclConvertToTsValueNode(int8_t precision, SValueNode* valueNode);
#define GET_PARAM_PRECISON(_c) ((_c)->columnData->info.precision) #define GET_PARAM_PRECISON(_c) ((_c)->columnData->info.precision)
void sclFreeParam(SScalarParam* param); void sclFreeParam(SScalarParam* param);
void doVectorCompare(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t startIndex, int32_t numOfRows, int32_t doVectorCompare(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t startIndex, int32_t numOfRows,
int32_t _ord, int32_t optr); int32_t _ord, int32_t optr);
void vectorCompareImpl(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t startIndex, int32_t numOfRows, int32_t vectorCompareImpl(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t startIndex, int32_t numOfRows,
int32_t _ord, int32_t optr); int32_t _ord, int32_t optr);
void vectorCompare(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t _ord, int32_t optr); int32_t vectorCompare(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t _ord, int32_t optr);
#ifdef __cplusplus #ifdef __cplusplus
} }

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

@ -29,43 +29,54 @@ typedef struct SSclVectorConvCtx {
int16_t outType; int16_t outType;
} SSclVectorConvCtx; } SSclVectorConvCtx;
typedef double (*_getDoubleValue_fn_t)(void *src, int32_t index); typedef int32_t (*_getDoubleValue_fn_t)(void *src, int32_t index, double *out);
static FORCE_INLINE double getVectorDoubleValue_TINYINT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_TINYINT(void *src, int32_t index, double *out) {
return (double)*((int8_t *)src + index); *out = (double)*((int8_t *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_UTINYINT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_UTINYINT(void *src, int32_t index, double *out) {
return (double)*((uint8_t *)src + index); *out = (double)*((uint8_t *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_SMALLINT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_SMALLINT(void *src, int32_t index, double *out) {
return (double)*((int16_t *)src + index); *out = (double)*((int16_t *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_USMALLINT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_USMALLINT(void *src, int32_t index, double *out) {
return (double)*((uint16_t *)src + index); *out = (double)*((uint16_t *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_INT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_INT(void *src, int32_t index, double *out) {
return (double)*((int32_t *)src + index); *out = (double)*((int32_t *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_UINT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_UINT(void *src, int32_t index, double *out) {
return (double)*((uint32_t *)src + index); *out = (double)*((uint32_t *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_BIGINT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_BIGINT(void *src, int32_t index, double *out) {
return (double)*((int64_t *)src + index); *out = (double)*((int64_t *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_UBIGINT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_UBIGINT(void *src, int32_t index, double *out) {
return (double)*((uint64_t *)src + index); *out = (double)*((uint64_t *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_FLOAT(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_FLOAT(void *src, int32_t index, double *out) {
return (double)*((float *)src + index); *out = (double)*((float *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_DOUBLE(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_DOUBLE(void *src, int32_t index, double *out) {
return (double)*((double *)src + index); *out = (double)*((double *)src + index);
return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE double getVectorDoubleValue_BOOL(void *src, int32_t index) { static FORCE_INLINE int32_t getVectorDoubleValue_BOOL(void *src, int32_t index, double *out) {
return (double)*((bool *)src + index); *out = (double)*((bool *)src + index);
return TSDB_CODE_SUCCESS;
} }
double getVectorDoubleValue_JSON(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) { static FORCE_INLINE _getDoubleValue_fn_t getVectorDoubleValueFn(int32_t srcType) {
_getDoubleValue_fn_t p = NULL; _getDoubleValue_fn_t p = NULL;
@ -103,8 +114,8 @@ static FORCE_INLINE _getDoubleValue_fn_t getVectorDoubleValueFn(int32_t srcType)
return p; return p;
} }
typedef void (*_bufConverteFunc)(char *buf, SScalarParam *pOut, int32_t outType, int32_t *overflow); typedef int32_t (*_bufConverteFunc)(char *buf, SScalarParam *pOut, int32_t outType, int32_t *overflow);
typedef void (*_bin_scalar_fn_t)(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *output, int32_t order); typedef int32_t (*_bin_scalar_fn_t)(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *output, int32_t order);
_bin_scalar_fn_t getBinScalarOperatorFn(int32_t binOperator); _bin_scalar_fn_t getBinScalarOperatorFn(int32_t binOperator);
#ifdef __cplusplus #ifdef __cplusplus

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

File diff suppressed because it is too large Load Diff

202
source/libs/scalar/src/scalar.c
View File

@ -72,9 +72,16 @@ int32_t sclConvertValueToSclParam(SValueNode *pValueNode, SScalarParam *out, int
return code; return code;
} }
colDataSetVal(in.columnData, 0, nodesGetValueFromNode(pValueNode), false); code = colDataSetVal(in.columnData, 0, nodesGetValueFromNode(pValueNode), false);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
code = colInfoDataEnsureCapacity(out->columnData, 1, true);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
colInfoDataEnsureCapacity(out->columnData, 1, true);
code = vectorConvertSingleColImpl(&in, out, overflow, -1, -1); code = vectorConvertSingleColImpl(&in, out, overflow, -1, -1);
sclFreeParam(&in); sclFreeParam(&in);
@ -84,6 +91,7 @@ int32_t sclConvertValueToSclParam(SValueNode *pValueNode, SScalarParam *out, int
int32_t sclExtendResRows(SScalarParam *pDst, SScalarParam *pSrc, SArray *pBlockList) { int32_t sclExtendResRows(SScalarParam *pDst, SScalarParam *pSrc, SArray *pBlockList) {
SSDataBlock *pb = taosArrayGetP(pBlockList, 0); SSDataBlock *pb = taosArrayGetP(pBlockList, 0);
SScalarParam *pLeft = taosMemoryCalloc(1, sizeof(SScalarParam)); SScalarParam *pLeft = taosMemoryCalloc(1, sizeof(SScalarParam));
int32_t code = TSDB_CODE_SUCCESS;
if (NULL == pLeft) { if (NULL == pLeft) {
sclError("calloc %d failed", (int32_t)sizeof(SScalarParam)); sclError("calloc %d failed", (int32_t)sizeof(SScalarParam));
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
@ -92,15 +100,16 @@ int32_t sclExtendResRows(SScalarParam *pDst, SScalarParam *pSrc, SArray *pBlockL
pLeft->numOfRows = pb->info.rows; pLeft->numOfRows = pb->info.rows;
if (pDst->numOfRows < pb->info.rows) { if (pDst->numOfRows < pb->info.rows) {
colInfoDataEnsureCapacity(pDst->columnData, pb->info.rows, true); SCL_ERR_JRET(colInfoDataEnsureCapacity(pDst->columnData, pb->info.rows, true));
} }
_bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(OP_TYPE_ASSIGN); _bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(OP_TYPE_ASSIGN);
OperatorFn(pLeft, pSrc, pDst, TSDB_ORDER_ASC); SCL_ERR_JRET(OperatorFn(pLeft, pSrc, pDst, TSDB_ORDER_ASC));
_return:
taosMemoryFree(pLeft); taosMemoryFree(pLeft);
return TSDB_CODE_SUCCESS; SCL_RET(code);
} }
int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) { int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) {
@ -116,7 +125,9 @@ int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) {
SNodeListNode *nodeList = (SNodeListNode *)pNode; SNodeListNode *nodeList = (SNodeListNode *)pNode;
SListCell *cell = nodeList->pNodeList->pHead; SListCell *cell = nodeList->pNodeList->pHead;
SScalarParam out = {.columnData = taosMemoryCalloc(1, sizeof(SColumnInfoData))}; SScalarParam out = {.columnData = taosMemoryCalloc(1, sizeof(SColumnInfoData))};
if (out.columnData == NULL) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
int32_t len = 0; int32_t len = 0;
void *buf = NULL; void *buf = NULL;
@ -228,7 +239,7 @@ int32_t sclCopyValueNodeValue(SValueNode *pNode, void **res) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
memcpy(*res, nodesGetValueFromNode(pNode), pNode->node.resType.bytes); (void)memcpy(*res, nodesGetValueFromNode(pNode), pNode->node.resType.bytes);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -343,7 +354,7 @@ int32_t sclInitParam(SNode *node, SScalarParam *param, SScalarCtx *ctx, int32_t
if (TSDB_DATA_TYPE_NULL == valueNode->node.resType.type || valueNode->isNull) { if (TSDB_DATA_TYPE_NULL == valueNode->node.resType.type || valueNode->isNull) {
colDataSetNULL(param->columnData, 0); colDataSetNULL(param->columnData, 0);
} else { } else {
colDataSetVal(param->columnData, 0, nodesGetValueFromNode(valueNode), false); SCL_ERR_RET(colDataSetVal(param->columnData, 0, nodesGetValueFromNode(valueNode), false));
} }
break; break;
} }
@ -502,23 +513,27 @@ _return:
SCL_RET(code); SCL_RET(code);
} }
int32_t sclGetNodeType(SNode *pNode, SScalarCtx *ctx) { int32_t sclGetNodeType(SNode *pNode, SScalarCtx *ctx, int32_t *type) {
if (NULL == pNode) { if (NULL == pNode) {
return -1; *type = -1;
return TSDB_CODE_SUCCESS;
} }
switch ((int)nodeType(pNode)) { switch ((int)nodeType(pNode)) {
case QUERY_NODE_VALUE: { case QUERY_NODE_VALUE: {
SValueNode *valueNode = (SValueNode *)pNode; SValueNode *valueNode = (SValueNode *)pNode;
return valueNode->node.resType.type; *type = valueNode->node.resType.type;
return TSDB_CODE_SUCCESS;
} }
case QUERY_NODE_NODE_LIST: { case QUERY_NODE_NODE_LIST: {
SNodeListNode *nodeList = (SNodeListNode *)pNode; SNodeListNode *nodeList = (SNodeListNode *)pNode;
return nodeList->node.resType.type; *type = nodeList->node.resType.type;
return TSDB_CODE_SUCCESS;
} }
case QUERY_NODE_COLUMN: { case QUERY_NODE_COLUMN: {
SColumnNode *colNode = (SColumnNode *)pNode; SColumnNode *colNode = (SColumnNode *)pNode;
return colNode->node.resType.type; *type = colNode->node.resType.type;
return TSDB_CODE_SUCCESS;
} }
case QUERY_NODE_FUNCTION: case QUERY_NODE_FUNCTION:
case QUERY_NODE_OPERATOR: case QUERY_NODE_OPERATOR:
@ -526,19 +541,22 @@ int32_t sclGetNodeType(SNode *pNode, SScalarCtx *ctx) {
SScalarParam *res = (SScalarParam *)taosHashGet(ctx->pRes, &pNode, POINTER_BYTES); SScalarParam *res = (SScalarParam *)taosHashGet(ctx->pRes, &pNode, POINTER_BYTES);
if (NULL == res) { if (NULL == res) {
sclError("no result for node, type:%d, node:%p", nodeType(pNode), pNode); sclError("no result for node, type:%d, node:%p", nodeType(pNode), pNode);
return -1; SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
} }
return res->columnData->info.type; *type = (int32_t)(res->columnData->info.type);
return TSDB_CODE_SUCCESS;
} }
} }
return -1; *type = -1;
return TSDB_CODE_SUCCESS;
} }
void sclSetOperatorValueType(SOperatorNode *node, SScalarCtx *ctx) { int32_t sclSetOperatorValueType(SOperatorNode *node, SScalarCtx *ctx) {
ctx->type.opResType = node->node.resType.type; ctx->type.opResType = node->node.resType.type;
ctx->type.selfType = sclGetNodeType(node->pLeft, ctx); SCL_ERR_RET(sclGetNodeType(node->pLeft, ctx, &(ctx->type.selfType)));
ctx->type.peerType = sclGetNodeType(node->pRight, ctx); SCL_ERR_RET(sclGetNodeType(node->pRight, ctx, &(ctx->type.peerType)));
SCL_RET(TSDB_CODE_SUCCESS);
} }
int32_t sclInitOperatorParams(SScalarParam **pParams, SOperatorNode *node, SScalarCtx *ctx, int32_t *rowNum) { int32_t sclInitOperatorParams(SScalarParam **pParams, SOperatorNode *node, SScalarCtx *ctx, int32_t *rowNum) {
@ -555,7 +573,7 @@ int32_t sclInitOperatorParams(SScalarParam **pParams, SOperatorNode *node, SScal
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
sclSetOperatorValueType(node, ctx); SCL_ERR_JRET(sclSetOperatorValueType(node, ctx));
SCL_ERR_JRET(sclInitParam(node->pLeft, &paramList[0], ctx, rowNum)); SCL_ERR_JRET(sclInitParam(node->pLeft, &paramList[0], ctx, rowNum));
if (paramNum > 1) { if (paramNum > 1) {
@ -603,13 +621,13 @@ int32_t sclWalkCaseWhenList(SScalarCtx *ctx, SNodeList *pList, struct SListCell
SCL_ERR_RET(sclGetNodeRes(pWhenThen->pWhen, ctx, &pWhen)); SCL_ERR_RET(sclGetNodeRes(pWhenThen->pWhen, ctx, &pWhen));
SCL_ERR_RET(sclGetNodeRes(pWhenThen->pThen, ctx, &pThen)); SCL_ERR_RET(sclGetNodeRes(pWhenThen->pThen, ctx, &pThen));
vectorCompareImpl(pCase, pWhen, pComp, rowIdx, 1, TSDB_ORDER_ASC, OP_TYPE_EQUAL); SCL_ERR_JRET(vectorCompareImpl(pCase, pWhen, pComp, rowIdx, 1, TSDB_ORDER_ASC, OP_TYPE_EQUAL));
bool *equal = (bool *)colDataGetData(pComp->columnData, rowIdx); bool *equal = (bool *)colDataGetData(pComp->columnData, rowIdx);
if (*equal) { if (*equal) {
bool isNull = colDataIsNull_s(pThen->columnData, (pThen->numOfRows > 1 ? rowIdx : 0)); bool isNull = colDataIsNull_s(pThen->columnData, (pThen->numOfRows > 1 ? rowIdx : 0));
char *pData = isNull ? NULL : colDataGetData(pThen->columnData, (pThen->numOfRows > 1 ? rowIdx : 0)); char *pData = isNull ? NULL : colDataGetData(pThen->columnData, (pThen->numOfRows > 1 ? rowIdx : 0));
colDataSetVal(output->columnData, rowIdx, pData, isNull); SCL_ERR_JRET(colDataSetVal(output->columnData, rowIdx, pData, isNull));
if (0 == rowIdx && 1 == pCase->numOfRows && 1 == pWhen->numOfRows && 1 == pThen->numOfRows && totalRows > 1) { if (0 == rowIdx && 1 == pCase->numOfRows && 1 == pWhen->numOfRows && 1 == pThen->numOfRows && totalRows > 1) {
SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList)); SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList));
@ -623,7 +641,7 @@ int32_t sclWalkCaseWhenList(SScalarCtx *ctx, SNodeList *pList, struct SListCell
if (pElse) { if (pElse) {
bool isNull = colDataIsNull_s(pElse->columnData, (pElse->numOfRows > 1 ? rowIdx : 0)); bool isNull = colDataIsNull_s(pElse->columnData, (pElse->numOfRows > 1 ? rowIdx : 0));
char *pData = isNull ? NULL : colDataGetData(pElse->columnData, (pElse->numOfRows > 1 ? rowIdx : 0)); char *pData = isNull ? NULL : colDataGetData(pElse->columnData, (pElse->numOfRows > 1 ? rowIdx : 0));
colDataSetVal(output->columnData, rowIdx, pData, isNull); SCL_ERR_JRET(colDataSetVal(output->columnData, rowIdx, pData, isNull));
if (0 == rowIdx && 1 == pCase->numOfRows && 1 == pElse->numOfRows && totalRows > 1) { if (0 == rowIdx && 1 == pCase->numOfRows && 1 == pElse->numOfRows && totalRows > 1) {
SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList)); SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList));
@ -633,7 +651,7 @@ int32_t sclWalkCaseWhenList(SScalarCtx *ctx, SNodeList *pList, struct SListCell
goto _return; goto _return;
} }
colDataSetVal(output->columnData, rowIdx, NULL, true); SCL_ERR_JRET(colDataSetVal(output->columnData, rowIdx, NULL, true));
if (0 == rowIdx && 1 == pCase->numOfRows && totalRows > 1) { if (0 == rowIdx && 1 == pCase->numOfRows && totalRows > 1) {
SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList)); SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList));
@ -672,7 +690,7 @@ int32_t sclWalkWhenList(SScalarCtx *ctx, SNodeList *pList, struct SListCell *pCe
if (*whenValue) { if (*whenValue) {
bool isNull = colDataIsNull_s(pThen->columnData, (pThen->numOfRows > 1 ? rowIdx : 0)); bool isNull = colDataIsNull_s(pThen->columnData, (pThen->numOfRows > 1 ? rowIdx : 0));
char *pData = isNull ? NULL : colDataGetData(pThen->columnData, (pThen->numOfRows > 1 ? rowIdx : 0)); char *pData = isNull ? NULL : colDataGetData(pThen->columnData, (pThen->numOfRows > 1 ? rowIdx : 0));
colDataSetVal(output->columnData, rowIdx, pData, isNull); SCL_ERR_JRET(colDataSetVal(output->columnData, rowIdx, pData, isNull));
if (preSingle && 0 == rowIdx && 1 == pWhen->numOfRows && 1 == pThen->numOfRows && totalRows > 1) { if (preSingle && 0 == rowIdx && 1 == pWhen->numOfRows && 1 == pThen->numOfRows && totalRows > 1) {
SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList)); SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList));
@ -691,7 +709,7 @@ int32_t sclWalkWhenList(SScalarCtx *ctx, SNodeList *pList, struct SListCell *pCe
if (pElse) { if (pElse) {
bool isNull = colDataIsNull_s(pElse->columnData, (pElse->numOfRows > 1 ? rowIdx : 0)); bool isNull = colDataIsNull_s(pElse->columnData, (pElse->numOfRows > 1 ? rowIdx : 0));
char *pData = isNull ? NULL : colDataGetData(pElse->columnData, (pElse->numOfRows > 1 ? rowIdx : 0)); char *pData = isNull ? NULL : colDataGetData(pElse->columnData, (pElse->numOfRows > 1 ? rowIdx : 0));
colDataSetVal(output->columnData, rowIdx, pData, isNull); SCL_ERR_JRET(colDataSetVal(output->columnData, rowIdx, pData, isNull));
if (preSingle && 0 == rowIdx && 1 == pElse->numOfRows && totalRows > 1) { if (preSingle && 0 == rowIdx && 1 == pElse->numOfRows && totalRows > 1) {
SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList)); SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList));
@ -701,7 +719,7 @@ int32_t sclWalkWhenList(SScalarCtx *ctx, SNodeList *pList, struct SListCell *pCe
goto _return; goto _return;
} }
colDataSetVal(output->columnData, rowIdx, NULL, true); SCL_ERR_JRET(colDataSetVal(output->columnData, rowIdx, NULL, true));
if (preSingle && 0 == rowIdx && totalRows > 1) { if (preSingle && 0 == rowIdx && totalRows > 1) {
SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList)); SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList));
@ -827,7 +845,7 @@ int32_t sclExecLogic(SLogicConditionNode *node, SScalarCtx *ctx, SScalarParam *o
} }
if (complete) { if (complete) {
colDataSetVal(output->columnData, i, (char *)&value, false); SCL_ERR_JRET(colDataSetVal(output->columnData, i, (char *)&value, false));
if (value) { if (value) {
numOfQualified++; numOfQualified++;
} }
@ -876,9 +894,8 @@ int32_t sclExecOperator(SOperatorNode *node, SScalarCtx *ctx, SScalarParam *outp
SScalarParam *pRight = paramNum > 1 ? &params[1] : NULL; SScalarParam *pRight = paramNum > 1 ? &params[1] : NULL;
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
OperatorFn(pLeft, pRight, output, TSDB_ORDER_ASC); SCL_ERR_JRET(OperatorFn(pLeft, pRight, output, TSDB_ORDER_ASC));
code = terrno; SCL_ERR_JRET(terrno);
_return: _return:
sclFreeParamList(params, paramNum); sclFreeParamList(params, paramNum);
@ -928,13 +945,14 @@ int32_t sclExecCaseWhen(SCaseWhenNode *node, SScalarCtx *ctx, SScalarParam *outp
} }
if (pCase) { if (pCase) {
vectorCompare(pCase, pWhen, &comp, TSDB_ORDER_ASC, OP_TYPE_EQUAL); SCL_ERR_JRET(vectorCompare(pCase, pWhen, &comp, TSDB_ORDER_ASC, OP_TYPE_EQUAL));
for (int32_t i = 0; i < rowNum; ++i) { for (int32_t i = 0; i < rowNum; ++i) {
bool *equal = (bool *)colDataGetData(comp.columnData, (comp.numOfRows > 1 ? i : 0)); bool *equal = (bool *)colDataGetData(comp.columnData, (comp.numOfRows > 1 ? i : 0));
if (*equal) { if (*equal) {
colDataSetVal(output->columnData, i, colDataGetData(pThen->columnData, (pThen->numOfRows > 1 ? i : 0)), SCL_ERR_JRET(colDataSetVal(output->columnData, i,
colDataIsNull_s(pThen->columnData, (pThen->numOfRows > 1 ? i : 0))); colDataGetData(pThen->columnData, (pThen->numOfRows > 1 ? i : 0)),
colDataIsNull_s(pThen->columnData, (pThen->numOfRows > 1 ? i : 0))));
if (0 == i && 1 == pCase->numOfRows && 1 == pWhen->numOfRows && 1 == pThen->numOfRows && rowNum > 1) { if (0 == i && 1 == pCase->numOfRows && 1 == pWhen->numOfRows && 1 == pThen->numOfRows && rowNum > 1) {
SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList)); SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList));
break; break;
@ -951,8 +969,9 @@ int32_t sclExecCaseWhen(SCaseWhenNode *node, SScalarCtx *ctx, SScalarParam *outp
for (int32_t i = 0; i < rowNum; ++i) { for (int32_t i = 0; i < rowNum; ++i) {
bool *whenValue = (bool *)colDataGetData(pWhen->columnData, (pWhen->numOfRows > 1 ? i : 0)); bool *whenValue = (bool *)colDataGetData(pWhen->columnData, (pWhen->numOfRows > 1 ? i : 0));
if (*whenValue) { if (*whenValue) {
colDataSetVal(output->columnData, i, colDataGetData(pThen->columnData, (pThen->numOfRows > 1 ? i : 0)), SCL_ERR_JRET(colDataSetVal(output->columnData, i,
colDataIsNull_s(pThen->columnData, (pThen->numOfRows > 1 ? i : 0))); colDataGetData(pThen->columnData, (pThen->numOfRows > 1 ? i : 0)),
colDataIsNull_s(pThen->columnData, (pThen->numOfRows > 1 ? i : 0))));
if (0 == i && 1 == pWhen->numOfRows && 1 == pThen->numOfRows && rowNum > 1) { if (0 == i && 1 == pWhen->numOfRows && 1 == pThen->numOfRows && rowNum > 1) {
SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList)); SCL_ERR_JRET(sclExtendResRows(output, output, ctx->pBlockList));
break; break;
@ -1063,42 +1082,34 @@ static uint8_t sclGetOpValueNodeTsPrecision(SNode *pLeft, SNode *pRight) {
return 0; return 0;
} }
int32_t sclConvertOpValueNodeTs(SOperatorNode *node, SScalarCtx *ctx) {
int32_t code = 0;
int32_t sclConvertOpValueNodeTs(SOperatorNode *node) {
if (node->pLeft && SCL_IS_VAR_VALUE_NODE(node->pLeft)) { if (node->pLeft && SCL_IS_VAR_VALUE_NODE(node->pLeft)) {
if (node->pRight && (TSDB_DATA_TYPE_TIMESTAMP == ((SExprNode *)node->pRight)->resType.type)) { if (node->pRight && (TSDB_DATA_TYPE_TIMESTAMP == ((SExprNode *)node->pRight)->resType.type)) {
SCL_ERR_JRET( SCL_ERR_RET(
sclConvertToTsValueNode(sclGetOpValueNodeTsPrecision(node->pLeft, node->pRight), (SValueNode *)node->pLeft)); sclConvertToTsValueNode(sclGetOpValueNodeTsPrecision(node->pLeft, node->pRight), (SValueNode *)node->pLeft));
} }
} else if (node->pRight && SCL_IS_NOTNULL_CONST_NODE(node->pRight)) { } else if (node->pRight && SCL_IS_NOTNULL_CONST_NODE(node->pRight)) {
if (node->pLeft && (TSDB_DATA_TYPE_TIMESTAMP == ((SExprNode *)node->pLeft)->resType.type)) { if (node->pLeft && (TSDB_DATA_TYPE_TIMESTAMP == ((SExprNode *)node->pLeft)->resType.type)) {
if (SCL_IS_VAR_VALUE_NODE(node->pRight)) { if (SCL_IS_VAR_VALUE_NODE(node->pRight)) {
SCL_ERR_JRET(sclConvertToTsValueNode(sclGetOpValueNodeTsPrecision(node->pLeft, node->pRight), SCL_ERR_RET(sclConvertToTsValueNode(sclGetOpValueNodeTsPrecision(node->pLeft, node->pRight),
(SValueNode *)node->pRight)); (SValueNode *)node->pRight));
} else if (QUERY_NODE_NODE_LIST == node->pRight->type) { } else if (QUERY_NODE_NODE_LIST == node->pRight->type) {
SNode *pNode; SNode *pNode;
FOREACH(pNode, ((SNodeListNode *)node->pRight)->pNodeList) { FOREACH(pNode, ((SNodeListNode *)node->pRight)->pNodeList) {
if (SCL_IS_VAR_VALUE_NODE(pNode)) { if (SCL_IS_VAR_VALUE_NODE(pNode)) {
SCL_ERR_JRET( SCL_ERR_RET(
sclConvertToTsValueNode(sclGetOpValueNodeTsPrecision(node->pLeft, pNode), (SValueNode *)pNode)); sclConvertToTsValueNode(sclGetOpValueNodeTsPrecision(node->pLeft, pNode), (SValueNode *)pNode));
} }
} }
} }
} }
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
_return:
ctx->code = code;
return DEAL_RES_ERROR;
} }
int32_t sclConvertCaseWhenValueNodeTs(SCaseWhenNode *node, SScalarCtx *ctx) {
int32_t code = 0;
int32_t sclConvertCaseWhenValueNodeTs(SCaseWhenNode *node) {
if (NULL == node->pCase) { if (NULL == node->pCase) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1108,7 +1119,7 @@ int32_t sclConvertCaseWhenValueNodeTs(SCaseWhenNode *node, SScalarCtx *ctx) {
FOREACH(pNode, node->pWhenThenList) { FOREACH(pNode, node->pWhenThenList) {
SExprNode *pExpr = (SExprNode *)((SWhenThenNode *)pNode)->pWhen; SExprNode *pExpr = (SExprNode *)((SWhenThenNode *)pNode)->pWhen;
if (TSDB_DATA_TYPE_TIMESTAMP == pExpr->resType.type) { if (TSDB_DATA_TYPE_TIMESTAMP == pExpr->resType.type) {
SCL_ERR_JRET(sclConvertToTsValueNode(pExpr->resType.precision, (SValueNode *)node->pCase)); SCL_ERR_RET(sclConvertToTsValueNode(pExpr->resType.precision, (SValueNode *)node->pCase));
break; break;
} }
} }
@ -1116,18 +1127,13 @@ int32_t sclConvertCaseWhenValueNodeTs(SCaseWhenNode *node, SScalarCtx *ctx) {
SNode *pNode; SNode *pNode;
FOREACH(pNode, node->pWhenThenList) { FOREACH(pNode, node->pWhenThenList) {
if (SCL_IS_VAR_VALUE_NODE(((SWhenThenNode *)pNode)->pWhen)) { if (SCL_IS_VAR_VALUE_NODE(((SWhenThenNode *)pNode)->pWhen)) {
SCL_ERR_JRET(sclConvertToTsValueNode(((SExprNode *)node->pCase)->resType.precision, SCL_ERR_RET(sclConvertToTsValueNode(((SExprNode *)node->pCase)->resType.precision,
(SValueNode *)((SWhenThenNode *)pNode)->pWhen)); (SValueNode *)((SWhenThenNode *)pNode)->pWhen));
} }
} }
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
_return:
ctx->code = code;
return DEAL_RES_ERROR;
} }
EDealRes sclRewriteNonConstOperator(SNode **pNode, SScalarCtx *ctx) { EDealRes sclRewriteNonConstOperator(SNode **pNode, SScalarCtx *ctx) {
@ -1213,7 +1219,7 @@ EDealRes sclRewriteFunction(SNode **pNode, SScalarCtx *ctx) {
res->translate = true; res->translate = true;
strcpy(res->node.aliasName, node->node.aliasName); (void)strcpy(res->node.aliasName, node->node.aliasName);
res->node.resType.type = output.columnData->info.type; res->node.resType.type = output.columnData->info.type;
res->node.resType.bytes = output.columnData->info.bytes; res->node.resType.bytes = output.columnData->info.bytes;
res->node.resType.scale = output.columnData->info.scale; res->node.resType.scale = output.columnData->info.scale;
@ -1225,14 +1231,33 @@ EDealRes sclRewriteFunction(SNode **pNode, SScalarCtx *ctx) {
if (type == TSDB_DATA_TYPE_JSON) { if (type == TSDB_DATA_TYPE_JSON) {
int32_t len = getJsonValueLen(output.columnData->pData); int32_t len = getJsonValueLen(output.columnData->pData);
res->datum.p = taosMemoryCalloc(len, 1); res->datum.p = taosMemoryCalloc(len, 1);
memcpy(res->datum.p, output.columnData->pData, len); if (NULL == res->datum.p) {
sclError("calloc %d failed", len);
sclFreeParam(&output);
nodesDestroyNode((SNode *)res);
ctx->code = TSDB_CODE_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
(void)memcpy(res->datum.p, output.columnData->pData, len);
} else if (IS_VAR_DATA_TYPE(type)) { } else if (IS_VAR_DATA_TYPE(type)) {
// res->datum.p = taosMemoryCalloc(res->node.resType.bytes + VARSTR_HEADER_SIZE + 1, 1); // res->datum.p = taosMemoryCalloc(res->node.resType.bytes + VARSTR_HEADER_SIZE + 1, 1);
res->datum.p = taosMemoryCalloc(varDataTLen(output.columnData->pData) + 1, 1); res->datum.p = taosMemoryCalloc(varDataTLen(output.columnData->pData) + 1, 1);
if (NULL == res->datum.p) {
sclError("calloc %d failed", (int)(varDataTLen(output.columnData->pData) + 1));
sclFreeParam(&output);
nodesDestroyNode((SNode *)res);
ctx->code = TSDB_CODE_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
res->node.resType.bytes = varDataTLen(output.columnData->pData); res->node.resType.bytes = varDataTLen(output.columnData->pData);
memcpy(res->datum.p, output.columnData->pData, varDataTLen(output.columnData->pData)); (void)memcpy(res->datum.p, output.columnData->pData, varDataTLen(output.columnData->pData));
} else { } else {
nodesSetValueNodeValue(res, output.columnData->pData); ctx->code = nodesSetValueNodeValue(res, output.columnData->pData);
if (ctx->code) {
sclFreeParam(&output);
nodesDestroyNode((SNode *)res);
return DEAL_RES_ERROR;
}
} }
} }
@ -1269,13 +1294,17 @@ EDealRes sclRewriteLogic(SNode **pNode, SScalarCtx *ctx) {
res->node.resType = node->node.resType; res->node.resType = node->node.resType;
res->translate = true; res->translate = true;
strcpy(res->node.aliasName, node->node.aliasName); (void)strcpy(res->node.aliasName, node->node.aliasName);
int32_t type = output.columnData->info.type; int32_t type = output.columnData->info.type;
if (IS_VAR_DATA_TYPE(type)) { if (IS_VAR_DATA_TYPE(type)) {
res->datum.p = output.columnData->pData; res->datum.p = output.columnData->pData;
output.columnData->pData = NULL; output.columnData->pData = NULL;
} else { } else {
nodesSetValueNodeValue(res, output.columnData->pData); ctx->code = nodesSetValueNodeValue(res, output.columnData->pData);
if (ctx->code) {
sclFreeParam(&output);
return DEAL_RES_ERROR;
}
} }
nodesDestroyNode(*pNode); nodesDestroyNode(*pNode);
@ -1288,7 +1317,10 @@ EDealRes sclRewriteLogic(SNode **pNode, SScalarCtx *ctx) {
EDealRes sclRewriteOperator(SNode **pNode, SScalarCtx *ctx) { EDealRes sclRewriteOperator(SNode **pNode, SScalarCtx *ctx) {
SOperatorNode *node = (SOperatorNode *)*pNode; SOperatorNode *node = (SOperatorNode *)*pNode;
SCL_ERR_RET(sclConvertOpValueNodeTs(node, ctx)); ctx->code = sclConvertOpValueNodeTs(node);
if (ctx->code) {
return DEAL_RES_ERROR;
}
if ((!SCL_IS_CONST_NODE(node->pLeft)) || (!SCL_IS_CONST_NODE(node->pRight))) { if ((!SCL_IS_CONST_NODE(node->pLeft)) || (!SCL_IS_CONST_NODE(node->pRight))) {
return sclRewriteNonConstOperator(pNode, ctx); return sclRewriteNonConstOperator(pNode, ctx);
@ -1311,7 +1343,7 @@ EDealRes sclRewriteOperator(SNode **pNode, SScalarCtx *ctx) {
res->translate = true; res->translate = true;
strcpy(res->node.aliasName, node->node.aliasName); (void)strcpy(res->node.aliasName, node->node.aliasName);
res->node.resType = node->node.resType; res->node.resType = node->node.resType;
if (colDataIsNull_s(output.columnData, 0)) { if (colDataIsNull_s(output.columnData, 0)) {
res->isNull = true; res->isNull = true;
@ -1322,7 +1354,11 @@ EDealRes sclRewriteOperator(SNode **pNode, SScalarCtx *ctx) {
res->datum.p = output.columnData->pData; res->datum.p = output.columnData->pData;
output.columnData->pData = NULL; output.columnData->pData = NULL;
} else { } else {
nodesSetValueNodeValue(res, output.columnData->pData); ctx->code = nodesSetValueNodeValue(res, output.columnData->pData);
if (ctx->code) {
sclFreeParam(&output);
return DEAL_RES_ERROR;
}
} }
} }
@ -1336,7 +1372,10 @@ EDealRes sclRewriteOperator(SNode **pNode, SScalarCtx *ctx) {
EDealRes sclRewriteCaseWhen(SNode **pNode, SScalarCtx *ctx) { EDealRes sclRewriteCaseWhen(SNode **pNode, SScalarCtx *ctx) {
SCaseWhenNode *node = (SCaseWhenNode *)*pNode; SCaseWhenNode *node = (SCaseWhenNode *)*pNode;
SCL_ERR_RET(sclConvertCaseWhenValueNodeTs(node, ctx)); ctx->code = sclConvertCaseWhenValueNodeTs(node);
if (ctx->code) {
return DEAL_RES_ERROR;
}
if ((!SCL_IS_CONST_NODE(node->pCase)) || (!SCL_IS_CONST_NODE(node->pElse))) { if ((!SCL_IS_CONST_NODE(node->pCase)) || (!SCL_IS_CONST_NODE(node->pElse))) {
return DEAL_RES_CONTINUE; return DEAL_RES_CONTINUE;
@ -1367,7 +1406,7 @@ EDealRes sclRewriteCaseWhen(SNode **pNode, SScalarCtx *ctx) {
res->translate = true; res->translate = true;
strcpy(res->node.aliasName, node->node.aliasName); (void)strcpy(res->node.aliasName, node->node.aliasName);
res->node.resType = node->node.resType; res->node.resType = node->node.resType;
if (colDataIsNull_s(output.columnData, 0)) { if (colDataIsNull_s(output.columnData, 0)) {
res->isNull = true; res->isNull = true;
@ -1376,9 +1415,21 @@ EDealRes sclRewriteCaseWhen(SNode **pNode, SScalarCtx *ctx) {
int32_t type = output.columnData->info.type; int32_t type = output.columnData->info.type;
if (IS_VAR_DATA_TYPE(type)) { // todo refactor if (IS_VAR_DATA_TYPE(type)) { // todo refactor
res->datum.p = taosMemoryCalloc(varDataTLen(output.columnData->pData) + 1, sizeof(char)); // add \0 to the end for print json value res->datum.p = taosMemoryCalloc(varDataTLen(output.columnData->pData) + 1, sizeof(char)); // add \0 to the end for print json value
memcpy(res->datum.p, output.columnData->pData, varDataTLen(output.columnData->pData)); if (NULL == res->datum.p) {
sclError("calloc %d failed", (int)(varDataTLen(output.columnData->pData) + 1));
sclFreeParam(&output);
nodesDestroyNode((SNode *)res);
ctx->code = TSDB_CODE_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
(void)memcpy(res->datum.p, output.columnData->pData, varDataTLen(output.columnData->pData));
} else { } else {
nodesSetValueNodeValue(res, output.columnData->pData); ctx->code = nodesSetValueNodeValue(res, output.columnData->pData);
if (ctx->code) {
sclFreeParam(&output);
nodesDestroyNode((SNode *)res);
return DEAL_RES_ERROR;
}
} }
} }
@ -1513,11 +1564,14 @@ EDealRes sclWalkTarget(SNode *pNode, SScalarCtx *ctx) {
return DEAL_RES_ERROR; return DEAL_RES_ERROR;
} }
colDataAssign(col, res->columnData, res->numOfRows, NULL); ctx->code = colDataAssign(col, res->columnData, res->numOfRows, NULL);
if (ctx->code) {
return DEAL_RES_ERROR;
}
block->info.rows = res->numOfRows; block->info.rows = res->numOfRows;
sclFreeParam(res); sclFreeParam(res);
taosHashRemove(ctx->pRes, (void *)&target->pExpr, POINTER_BYTES); (void)taosHashRemove(ctx->pRes, (void *)&target->pExpr, POINTER_BYTES);
return DEAL_RES_CONTINUE; return DEAL_RES_CONTINUE;
} }
@ -1733,14 +1787,14 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
if (1 == res->numOfRows && pb->info.rows > 0) { if (1 == res->numOfRows && pb->info.rows > 0) {
SCL_ERR_JRET(sclExtendResRows(pDst, res, pBlockList)); SCL_ERR_JRET(sclExtendResRows(pDst, res, pBlockList));
} else { } else {
colInfoDataEnsureCapacity(pDst->columnData, res->numOfRows, true); SCL_ERR_JRET(colInfoDataEnsureCapacity(pDst->columnData, res->numOfRows, true));
colDataAssign(pDst->columnData, res->columnData, res->numOfRows, NULL); SCL_ERR_JRET(colDataAssign(pDst->columnData, res->columnData, res->numOfRows, NULL));
pDst->numOfRows = res->numOfRows; pDst->numOfRows = res->numOfRows;
pDst->numOfQualified = res->numOfQualified; pDst->numOfQualified = res->numOfQualified;
} }
sclFreeParam(res); sclFreeParam(res);
taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES); (void)taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES);
} }
_return: _return:

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

@ -124,7 +124,7 @@ int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
} }
default: { default: {
colDataAssign(pOutputData, pInputData, pInput->numOfRows, NULL); SCL_ERR_RET(colDataAssign(pOutputData, pInputData, pInput->numOfRows, NULL));
} }
} }
@ -147,7 +147,9 @@ static int32_t doScalarFunctionUnique(SScalarParam *pInput, int32_t inputNum, SS
colDataSetNULL(pOutputData, i); colDataSetNULL(pOutputData, i);
continue; continue;
} }
double result = valFn(getValueFn(pInputData->pData, i)); double tmp = 0;
SCL_ERR_RET(getValueFn(pInputData->pData, i, &tmp));
double result = valFn(tmp);
if (isinf(result) || isnan(result)) { if (isinf(result) || isnan(result)) {
colDataSetNULL(pOutputData, i); colDataSetNULL(pOutputData, i);
} else { } else {
@ -182,7 +184,11 @@ static int32_t doScalarFunctionUnique2(SScalarParam *pInput, int32_t inputNum, S
colDataSetNULL(pOutputData, i); colDataSetNULL(pOutputData, i);
continue; continue;
} }
result = valFn(getValueFn[0](pInputData[0]->pData, i), getValueFn[1](pInputData[1]->pData, i)); double val1 = 0;
double val2 = 0;
SCL_ERR_RET(getValueFn[0](pInputData[0]->pData, i, &val1));
SCL_ERR_RET(getValueFn[1](pInputData[1]->pData, i, &val2));
result = valFn(val1, val2);
if (isinf(result) || isnan(result)) { if (isinf(result) || isnan(result)) {
colDataSetNULL(pOutputData, i); colDataSetNULL(pOutputData, i);
} else { } else {
@ -198,8 +204,11 @@ static int32_t doScalarFunctionUnique2(SScalarParam *pInput, int32_t inputNum, S
colDataSetNULL(pOutputData, i); colDataSetNULL(pOutputData, i);
continue; continue;
} }
double val1 = 0;
result = valFn(getValueFn[0](pInputData[0]->pData, 0), getValueFn[1](pInputData[1]->pData, i)); double val2 = 0;
SCL_ERR_RET(getValueFn[0](pInputData[0]->pData, 0, &val1));
SCL_ERR_RET(getValueFn[1](pInputData[1]->pData, i, &val2));
result = valFn(val1, val2);
if (isinf(result) || isnan(result)) { if (isinf(result) || isnan(result)) {
colDataSetNULL(pOutputData, i); colDataSetNULL(pOutputData, i);
continue; continue;
@ -217,8 +226,11 @@ static int32_t doScalarFunctionUnique2(SScalarParam *pInput, int32_t inputNum, S
colDataSetNULL(pOutputData, i); colDataSetNULL(pOutputData, i);
continue; continue;
} }
double val1 = 0;
result = valFn(getValueFn[0](pInputData[0]->pData, i), getValueFn[1](pInputData[1]->pData, 0)); double val2 = 0;
SCL_ERR_RET(getValueFn[0](pInputData[0]->pData, i, &val1));
SCL_ERR_RET(getValueFn[1](pInputData[1]->pData, 0, &val2));
result = valFn(val1, val2);
if (isinf(result) || isnan(result)) { if (isinf(result) || isnan(result)) {
colDataSetNULL(pOutputData, i); colDataSetNULL(pOutputData, i);
continue; continue;
@ -277,7 +289,7 @@ static int32_t doScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
} }
default: { default: {
colDataAssign(pOutputData, pInputData, pInput->numOfRows, NULL); SCL_ERR_RET(colDataAssign(pOutputData, pInputData, pInput->numOfRows, NULL));
} }
} }
@ -380,6 +392,9 @@ static int32_t doLengthFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
static int32_t concatCopyHelper(const char *input, char *output, bool hasNchar, int32_t type, VarDataLenT *dataLen) { static int32_t concatCopyHelper(const char *input, char *output, bool hasNchar, int32_t type, VarDataLenT *dataLen) {
if (hasNchar && type == TSDB_DATA_TYPE_VARCHAR) { if (hasNchar && type == TSDB_DATA_TYPE_VARCHAR) {
TdUcs4 *newBuf = taosMemoryCalloc((varDataLen(input) + 1) * TSDB_NCHAR_SIZE, 1); TdUcs4 *newBuf = taosMemoryCalloc((varDataLen(input) + 1) * TSDB_NCHAR_SIZE, 1);
if (NULL == newBuf) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t len = varDataLen(input); int32_t len = varDataLen(input);
bool ret = taosMbsToUcs4(varDataVal(input), len, newBuf, (varDataLen(input) + 1) * TSDB_NCHAR_SIZE, &len); bool ret = taosMbsToUcs4(varDataVal(input), len, newBuf, (varDataLen(input) + 1) * TSDB_NCHAR_SIZE, &len);
if (!ret) { if (!ret) {
@ -410,12 +425,19 @@ static int32_t getNumOfNullEntries(SColumnInfoData *pColumnInfoData, int32_t num
} }
int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) { int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
int32_t ret = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SColumnInfoData **pInputData = taosMemoryCalloc(inputNum, sizeof(SColumnInfoData *)); SColumnInfoData **pInputData = taosMemoryCalloc(inputNum, sizeof(SColumnInfoData *));
SColumnInfoData *pOutputData = pOutput->columnData; SColumnInfoData *pOutputData = pOutput->columnData;
char **input = taosMemoryCalloc(inputNum, POINTER_BYTES); char **input = taosMemoryCalloc(inputNum, POINTER_BYTES);
char *outputBuf = NULL; char *outputBuf = NULL;
if (NULL == pInputData) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
if (NULL == input) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
int32_t inputLen = 0; int32_t inputLen = 0;
int32_t numOfRows = 0; int32_t numOfRows = 0;
bool hasNchar = (GET_PARAM_TYPE(pOutput) == TSDB_DATA_TYPE_NCHAR) ? true : false; bool hasNchar = (GET_PARAM_TYPE(pOutput) == TSDB_DATA_TYPE_NCHAR) ? true : false;
@ -441,6 +463,9 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
int32_t outputLen = inputLen + numOfRows * VARSTR_HEADER_SIZE; int32_t outputLen = inputLen + numOfRows * VARSTR_HEADER_SIZE;
outputBuf = taosMemoryCalloc(outputLen, 1); outputBuf = taosMemoryCalloc(outputLen, 1);
if (NULL == outputBuf) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
char *output = outputBuf; char *output = outputBuf;
for (int32_t k = 0; k < numOfRows; ++k) { for (int32_t k = 0; k < numOfRows; ++k) {
@ -462,33 +487,37 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
int32_t rowIdx = (pInput[i].numOfRows == 1) ? 0 : k; int32_t rowIdx = (pInput[i].numOfRows == 1) ? 0 : k;
input[i] = colDataGetData(pInputData[i], rowIdx); input[i] = colDataGetData(pInputData[i], rowIdx);
ret = concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen); SCL_ERR_JRET(concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen));
if (ret != TSDB_CODE_SUCCESS) {
goto DONE;
}
} }
varDataSetLen(output, dataLen); varDataSetLen(output, dataLen);
colDataSetVal(pOutputData, k, output, false); SCL_ERR_JRET(colDataSetVal(pOutputData, k, output, false));
output += varDataTLen(output); output += varDataTLen(output);
} }
pOutput->numOfRows = numOfRows; pOutput->numOfRows = numOfRows;
DONE: _return:
taosMemoryFree(input); taosMemoryFree(input);
taosMemoryFree(outputBuf); taosMemoryFree(outputBuf);
taosMemoryFree(pInputData); taosMemoryFree(pInputData);
return ret; SCL_RET(code);
} }
int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) { int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
int32_t ret = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SColumnInfoData **pInputData = taosMemoryCalloc(inputNum, sizeof(SColumnInfoData *)); SColumnInfoData **pInputData = taosMemoryCalloc(inputNum, sizeof(SColumnInfoData *));
SColumnInfoData *pOutputData = pOutput->columnData; SColumnInfoData *pOutputData = pOutput->columnData;
char **input = taosMemoryCalloc(inputNum, POINTER_BYTES); char **input = taosMemoryCalloc(inputNum, POINTER_BYTES);
char *outputBuf = NULL; char *outputBuf = NULL;
if (NULL == pInputData) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
if (NULL == input) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
int32_t inputLen = 0; int32_t inputLen = 0;
int32_t numOfRows = 0; int32_t numOfRows = 0;
bool hasNchar = (GET_PARAM_TYPE(pOutput) == TSDB_DATA_TYPE_NCHAR) ? true : false; bool hasNchar = (GET_PARAM_TYPE(pOutput) == TSDB_DATA_TYPE_NCHAR) ? true : false;
@ -518,6 +547,10 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
int32_t outputLen = inputLen + numOfRows * VARSTR_HEADER_SIZE; int32_t outputLen = inputLen + numOfRows * VARSTR_HEADER_SIZE;
outputBuf = taosMemoryCalloc(outputLen, 1); outputBuf = taosMemoryCalloc(outputLen, 1);
if (NULL == outputBuf) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
char *output = outputBuf; char *output = outputBuf;
for (int32_t k = 0; k < numOfRows; ++k) { for (int32_t k = 0; k < numOfRows; ++k) {
@ -537,18 +570,12 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
int32_t rowIdx = (pInput[i].numOfRows == 1) ? 0 : k; int32_t rowIdx = (pInput[i].numOfRows == 1) ? 0 : k;
input[i] = colDataGetData(pInputData[i], rowIdx); input[i] = colDataGetData(pInputData[i], rowIdx);
ret = concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen); SCL_ERR_JRET(concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen));
if (ret != TSDB_CODE_SUCCESS) {
goto DONE;
}
if (i < inputNum - 1) { if (i < inputNum - 1) {
// insert the separator // insert the separator
char *sep = (pInput[0].numOfRows == 1) ? colDataGetData(pInputData[0], 0) : colDataGetData(pInputData[0], k); char *sep = (pInput[0].numOfRows == 1) ? colDataGetData(pInputData[0], 0) : colDataGetData(pInputData[0], k);
ret = concatCopyHelper(sep, output, hasNchar, GET_PARAM_TYPE(&pInput[0]), &dataLen); SCL_ERR_JRET(concatCopyHelper(sep, output, hasNchar, GET_PARAM_TYPE(&pInput[0]), &dataLen));
if (ret != TSDB_CODE_SUCCESS) {
goto DONE;
}
} }
} }
@ -557,19 +584,19 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
memset(output, 0, dataLen); memset(output, 0, dataLen);
} else { } else {
varDataSetLen(output, dataLen); varDataSetLen(output, dataLen);
colDataSetVal(pOutputData, k, output, false); SCL_ERR_JRET(colDataSetVal(pOutputData, k, output, false));
output += varDataTLen(output); output += varDataTLen(output);
} }
} }
pOutput->numOfRows = numOfRows; pOutput->numOfRows = numOfRows;
DONE: _return:
taosMemoryFree(input); taosMemoryFree(input);
taosMemoryFree(outputBuf); taosMemoryFree(outputBuf);
taosMemoryFree(pInputData); taosMemoryFree(pInputData);
return ret; return code;
} }
static int32_t doCaseConvFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput, _conv_fn convFn) { static int32_t doCaseConvFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput, _conv_fn convFn) {
@ -580,6 +607,10 @@ static int32_t doCaseConvFunction(SScalarParam *pInput, int32_t inputNum, SScala
int32_t outputLen = pInputData->varmeta.length; int32_t outputLen = pInputData->varmeta.length;
char *outputBuf = taosMemoryCalloc(outputLen, 1); char *outputBuf = taosMemoryCalloc(outputLen, 1);
if (outputBuf == NULL) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
char *output = outputBuf; char *output = outputBuf;
for (int32_t i = 0; i < pInput->numOfRows; ++i) { for (int32_t i = 0; i < pInput->numOfRows; ++i) {
@ -600,7 +631,11 @@ static int32_t doCaseConvFunction(SScalarParam *pInput, int32_t inputNum, SScala
} }
} }
varDataSetLen(output, len); varDataSetLen(output, len);
colDataSetVal(pOutputData, i, output, false); int32_t code = colDataSetVal(pOutputData, i, output, false);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFree(outputBuf);
SCL_ERR_RET(code);
}
output += varDataTLen(output); output += varDataTLen(output);
} }
@ -618,6 +653,10 @@ static int32_t doTrimFunction(SScalarParam *pInput, int32_t inputNum, SScalarPar
int32_t outputLen = pInputData->varmeta.length; int32_t outputLen = pInputData->varmeta.length;
char *outputBuf = taosMemoryCalloc(outputLen, 1); char *outputBuf = taosMemoryCalloc(outputLen, 1);
if (outputBuf == NULL) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
char *output = outputBuf; char *output = outputBuf;
for (int32_t i = 0; i < pInput->numOfRows; ++i) { for (int32_t i = 0; i < pInput->numOfRows; ++i) {
@ -631,7 +670,11 @@ static int32_t doTrimFunction(SScalarParam *pInput, int32_t inputNum, SScalarPar
int32_t charLen = (type == TSDB_DATA_TYPE_VARCHAR) ? len : len / TSDB_NCHAR_SIZE; int32_t charLen = (type == TSDB_DATA_TYPE_VARCHAR) ? len : len / TSDB_NCHAR_SIZE;
trimFn(input, output, type, charLen); trimFn(input, output, type, charLen);
colDataSetVal(pOutputData, i, output, false); int32_t code = colDataSetVal(pOutputData, i, output, false);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFree(outputBuf);
SCL_ERR_RET(code);
}
output += varDataTLen(output); output += varDataTLen(output);
} }
@ -688,7 +731,11 @@ int32_t substrFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
varDataSetLen(output, 0); varDataSetLen(output, 0);
} }
colDataSetVal(pOutputData, i, output, false); int32_t code = colDataSetVal(pOutputData, i, output, false);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFree(outputBuf);
SCL_ERR_RET(code);
}
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
@ -715,12 +762,21 @@ int32_t md5Function(SScalarParam* pInput, int32_t inputNum, SScalarParam* pOutpu
if (bufLen < varDataLen(input) + VARSTR_HEADER_SIZE) { if (bufLen < varDataLen(input) + VARSTR_HEADER_SIZE) {
bufLen = varDataLen(input) + VARSTR_HEADER_SIZE; bufLen = varDataLen(input) + VARSTR_HEADER_SIZE;
pOutputBuf = taosMemoryRealloc(pOutputBuf, bufLen); pOutputBuf = taosMemoryRealloc(pOutputBuf, bufLen);
if (!pOutputBuf) {
qError("md5 function alloc memory failed");
return TSDB_CODE_OUT_OF_MEMORY;
}
} }
char *output = pOutputBuf; char *output = pOutputBuf;
memcpy(varDataVal(output), varDataVal(input), varDataLen(input)); memcpy(varDataVal(output), varDataVal(input), varDataLen(input));
int32_t len = taosCreateMD5Hash(varDataVal(output), varDataLen(input)); int32_t len = taosCreateMD5Hash(varDataVal(output), varDataLen(input));
varDataSetLen(output, len); varDataSetLen(output, len);
colDataSetVal(pOutputData, i, output, false); int32_t code = colDataSetVal(pOutputData, i, output, false);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFree(pOutputBuf);
SCL_ERR_RET(code);
}
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
taosMemoryFree(pOutputBuf); taosMemoryFree(pOutputBuf);
@ -1062,7 +1118,10 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
} }
} }
colDataSetVal(pOutput->columnData, i, output, false); code = colDataSetVal(pOutput->columnData, i, output, false);
if (TSDB_CODE_SUCCESS != code) {
goto _end;
}
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
@ -1167,7 +1226,7 @@ _end:
memmove(buf + VARSTR_HEADER_SIZE, buf, len); memmove(buf + VARSTR_HEADER_SIZE, buf, len);
varDataSetLen(buf, len); varDataSetLen(buf, len);
colDataSetVal(pOutput->columnData, i, buf, false); SCL_ERR_RET(colDataSetVal(pOutput->columnData, i, buf, false));
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
@ -1193,7 +1252,7 @@ int32_t toUnixtimestampFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
colDataSetNULL(pOutput->columnData, i); colDataSetNULL(pOutput->columnData, i);
} else { } else {
colDataSetVal(pOutput->columnData, i, (char *)&timeVal, false); SCL_ERR_RET(colDataSetVal(pOutput->columnData, i, (char *)&timeVal, false));
} }
} }
@ -1206,12 +1265,21 @@ int32_t toJsonFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
int32_t type = GET_PARAM_TYPE(pInput); int32_t type = GET_PARAM_TYPE(pInput);
char tmp[TSDB_MAX_JSON_TAG_LEN] = {0}; char tmp[TSDB_MAX_JSON_TAG_LEN] = {0};
int32_t code = TSDB_CODE_SUCCESS;
for (int32_t i = 0; i < pInput[0].numOfRows; ++i) { for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
SArray *pTagVals = taosArrayInit(8, sizeof(STagVal)); SArray *pTagVals = taosArrayInit(8, sizeof(STagVal));
if (NULL == pTagVals) {
return TSDB_CODE_OUT_OF_MEMORY;
}
STag *pTag = NULL; STag *pTag = NULL;
if (colDataIsNull_s(pInput[0].columnData, i)) { if (colDataIsNull_s(pInput[0].columnData, i)) {
tTagNew(pTagVals, 1, true, &pTag); code = tTagNew(pTagVals, 1, true, &pTag);
if (TSDB_CODE_SUCCESS != code) {
tTagFree(pTag);
taosArrayDestroy(pTagVals);
SCL_ERR_RET(code);
}
} else { } else {
char *input = pInput[0].columnData->pData + pInput[0].columnData->varmeta.offset[i]; char *input = pInput[0].columnData->pData + pInput[0].columnData->varmeta.offset[i];
if (varDataLen(input) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) { if (varDataLen(input) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
@ -1221,13 +1289,21 @@ int32_t toJsonFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
memcpy(tmp, varDataVal(input), varDataLen(input)); memcpy(tmp, varDataVal(input), varDataLen(input));
tmp[varDataLen(input)] = 0; tmp[varDataLen(input)] = 0;
if (parseJsontoTagData(tmp, pTagVals, &pTag, NULL)) { if (parseJsontoTagData(tmp, pTagVals, &pTag, NULL)) {
tTagNew(pTagVals, 1, true, &pTag); code = tTagNew(pTagVals, 1, true, &pTag);
if (TSDB_CODE_SUCCESS != code) {
tTagFree(pTag);
taosArrayDestroy(pTagVals);
SCL_ERR_RET(code);
}
} }
} }
colDataSetVal(pOutput->columnData, i, (const char *)pTag, false); code = colDataSetVal(pOutput->columnData, i, (const char *)pTag, false);
tTagFree(pTag); tTagFree(pTag);
taosArrayDestroy(pTagVals); taosArrayDestroy(pTagVals);
if (TSDB_CODE_SUCCESS != code ) {
SCL_ERR_RET(code);
}
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
@ -1242,6 +1318,9 @@ int32_t toTimestampFunction(SScalarParam* pInput, int32_t inputNum, SScalarParam
int32_t len, code = TSDB_CODE_SUCCESS; int32_t len, code = TSDB_CODE_SUCCESS;
SArray *formats = NULL; SArray *formats = NULL;
if (tsStr == NULL || format == NULL) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
for (int32_t i = 0; i < pInput[0].numOfRows; ++i) { for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
if (colDataIsNull_s(pInput[1].columnData, i) || colDataIsNull_s(pInput[0].columnData, i)) { if (colDataIsNull_s(pInput[1].columnData, i) || colDataIsNull_s(pInput[0].columnData, i)) {
colDataSetNULL(pOutput->columnData, i); colDataSetNULL(pOutput->columnData, i);
@ -1251,11 +1330,11 @@ int32_t toTimestampFunction(SScalarParam* pInput, int32_t inputNum, SScalarParam
char *tsData = colDataGetData(pInput[0].columnData, i); char *tsData = colDataGetData(pInput[0].columnData, i);
char *formatData = colDataGetData(pInput[1].columnData, pInput[1].numOfRows > 1 ? i : 0); char *formatData = colDataGetData(pInput[1].columnData, pInput[1].numOfRows > 1 ? i : 0);
len = TMIN(TS_FORMAT_MAX_LEN - 1, varDataLen(tsData)); len = TMIN(TS_FORMAT_MAX_LEN - 1, varDataLen(tsData));
strncpy(tsStr, varDataVal(tsData), len); (void)strncpy(tsStr, varDataVal(tsData), len); // No need to handle the return value.
tsStr[len] = '\0'; tsStr[len] = '\0';
len = TMIN(TS_FORMAT_MAX_LEN - 1, varDataLen(formatData)); len = TMIN(TS_FORMAT_MAX_LEN - 1, varDataLen(formatData));
if (pInput[1].numOfRows > 1 || i == 0) { if (pInput[1].numOfRows > 1 || i == 0) {
strncpy(format, varDataVal(formatData), len); (void)strncpy(format, varDataVal(formatData), len); // No need to handle the return value.
format[len] = '\0'; format[len] = '\0';
if (formats) { if (formats) {
taosArrayDestroy(formats); taosArrayDestroy(formats);
@ -1267,10 +1346,12 @@ int32_t toTimestampFunction(SScalarParam* pInput, int32_t inputNum, SScalarParam
code = taosChar2Ts(format, &formats, tsStr, &ts, precision, errMsg, 128); code = taosChar2Ts(format, &formats, tsStr, &ts, precision, errMsg, 128);
if (code) { if (code) {
qError("func to_timestamp failed %s", errMsg); qError("func to_timestamp failed %s", errMsg);
break; SCL_ERR_JRET(code);
} }
colDataSetVal(pOutput->columnData, i, (char *)&ts, false); SCL_ERR_JRET(colDataSetVal(pOutput->columnData, i, (char *)&ts, false));
} }
_return:
if (formats) taosArrayDestroy(formats); if (formats) taosArrayDestroy(formats);
taosMemoryFree(tsStr); taosMemoryFree(tsStr);
taosMemoryFree(format); taosMemoryFree(format);
@ -1283,6 +1364,10 @@ int32_t toCharFunction(SScalarParam* pInput, int32_t inputNum, SScalarParam* pOu
int32_t len; int32_t len;
SArray *formats = NULL; SArray *formats = NULL;
int32_t code = 0; int32_t code = 0;
if (format == NULL || out == NULL) {
SCL_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
for (int32_t i = 0; i < pInput[0].numOfRows; ++i) { for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
if (colDataIsNull_s(pInput[1].columnData, i) || colDataIsNull_s(pInput[0].columnData, i)) { if (colDataIsNull_s(pInput[1].columnData, i) || colDataIsNull_s(pInput[0].columnData, i)) {
colDataSetNULL(pOutput->columnData, i); colDataSetNULL(pOutput->columnData, i);
@ -1293,7 +1378,7 @@ int32_t toCharFunction(SScalarParam* pInput, int32_t inputNum, SScalarParam* pOu
char *formatData = colDataGetData(pInput[1].columnData, pInput[1].numOfRows > 1 ? i : 0); char *formatData = colDataGetData(pInput[1].columnData, pInput[1].numOfRows > 1 ? i : 0);
len = TMIN(TS_FORMAT_MAX_LEN - 1, varDataLen(formatData)); len = TMIN(TS_FORMAT_MAX_LEN - 1, varDataLen(formatData));
if (pInput[1].numOfRows > 1 || i == 0) { if (pInput[1].numOfRows > 1 || i == 0) {
strncpy(format, varDataVal(formatData), len); (void)strncpy(format, varDataVal(formatData), len);
format[len] = '\0'; format[len] = '\0';
if (formats) { if (formats) {
taosArrayDestroy(formats); taosArrayDestroy(formats);
@ -1301,11 +1386,12 @@ int32_t toCharFunction(SScalarParam* pInput, int32_t inputNum, SScalarParam* pOu
} }
} }
int32_t precision = pInput[0].columnData->info.precision; int32_t precision = pInput[0].columnData->info.precision;
code = taosTs2Char(format, &formats, *(int64_t *)ts, precision, varDataVal(out), TS_FORMAT_MAX_LEN); SCL_ERR_JRET(taosTs2Char(format, &formats, *(int64_t *)ts, precision, varDataVal(out), TS_FORMAT_MAX_LEN));
if (code) break;
varDataSetLen(out, strlen(varDataVal(out))); varDataSetLen(out, strlen(varDataVal(out)));
colDataSetVal(pOutput->columnData, i, out, false); SCL_ERR_JRET(colDataSetVal(pOutput->columnData, i, out, false));
} }
_return:
if (formats) taosArrayDestroy(formats); if (formats) taosArrayDestroy(formats);
taosMemoryFree(format); taosMemoryFree(format);
taosMemoryFree(out); taosMemoryFree(out);
@ -1341,7 +1427,7 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
} }
GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[timePrecIdx]), pInput[timePrecIdx].columnData->pData); GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[timePrecIdx]), pInput[timePrecIdx].columnData->pData);
memcpy(timezone, varDataVal(pInput[timeZoneIdx].columnData->pData), varDataLen(pInput[timeZoneIdx].columnData->pData)); (void)memcpy(timezone, varDataVal(pInput[timeZoneIdx].columnData->pData), varDataLen(pInput[timeZoneIdx].columnData->pData));
for (int32_t i = 0; i < pInput[0].numOfRows; ++i) { for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
if (colDataIsNull_s(pInput[0].columnData, i)) { if (colDataIsNull_s(pInput[0].columnData, i)) {
@ -1373,7 +1459,7 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
} else { } else {
timeVal = timeVal / timeUnit * timeUnit; timeVal = timeVal / timeUnit * timeUnit;
} }
colDataSetVal(pOutput->columnData, i, (char *)&timeVal, false); SCL_ERR_RET(colDataSetVal(pOutput->columnData, i, (char *)&timeVal, false));
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
@ -1507,7 +1593,7 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
} }
} }
colDataSetVal(pOutput->columnData, i, (char *)&result, false); SCL_ERR_RET(colDataSetVal(pOutput->columnData, i, (char *)&result, false));
} }
pOutput->numOfRows = numOfRows; pOutput->numOfRows = numOfRows;
@ -1541,10 +1627,10 @@ int32_t todayFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOut
int32_t timezoneFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) { int32_t timezoneFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
char output[TD_TIMEZONE_LEN + VARSTR_HEADER_SIZE] = {0}; char output[TD_TIMEZONE_LEN + VARSTR_HEADER_SIZE] = {0};
memcpy(varDataVal(output), tsTimezoneStr, TD_TIMEZONE_LEN); (void)memcpy(varDataVal(output), tsTimezoneStr, TD_TIMEZONE_LEN);
varDataSetLen(output, strlen(tsTimezoneStr)); varDataSetLen(output, strlen(tsTimezoneStr));
for (int32_t i = 0; i < pInput->numOfRows; ++i) { for (int32_t i = 0; i < pInput->numOfRows; ++i) {
colDataSetVal(pOutput->columnData, i, output, false); SCL_ERR_RET(colDataSetVal(pOutput->columnData, i, output, false));
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -2259,16 +2345,16 @@ int32_t leastSQRScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarPa
char interceptBuf[64] = {0}; char interceptBuf[64] = {0};
int n = snprintf(slopBuf, 64, "%.6lf", matrix02); int n = snprintf(slopBuf, 64, "%.6lf", matrix02);
if (n > LEASTSQUARES_DOUBLE_ITEM_LENGTH) { if (n > LEASTSQUARES_DOUBLE_ITEM_LENGTH) {
snprintf(slopBuf, 64, "%." DOUBLE_PRECISION_DIGITS, matrix02); (void)snprintf(slopBuf, 64, "%." DOUBLE_PRECISION_DIGITS, matrix02);
} }
n = snprintf(interceptBuf, 64, "%.6lf", matrix12); n = snprintf(interceptBuf, 64, "%.6lf", matrix12);
if (n > LEASTSQUARES_DOUBLE_ITEM_LENGTH) { if (n > LEASTSQUARES_DOUBLE_ITEM_LENGTH) {
snprintf(interceptBuf, 64, "%." DOUBLE_PRECISION_DIGITS, matrix12); (void) snprintf(interceptBuf, 64, "%." DOUBLE_PRECISION_DIGITS, matrix12);
} }
size_t len = size_t len =
snprintf(varDataVal(buf), sizeof(buf) - VARSTR_HEADER_SIZE, "{slop:%s, intercept:%s}", slopBuf, interceptBuf); snprintf(varDataVal(buf), sizeof(buf) - VARSTR_HEADER_SIZE, "{slop:%s, intercept:%s}", slopBuf, interceptBuf);
varDataSetLen(buf, len); varDataSetLen(buf, len);
colDataSetVal(pOutputData, 0, buf, false); SCL_ERR_RET(colDataSetVal(pOutputData, 0, buf, false));
} }
pOutput->numOfRows = 1; pOutput->numOfRows = 1;
@ -2295,7 +2381,7 @@ int32_t percentileScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalar
if (hasNull) { if (hasNull) {
colDataSetNULL(pOutputData, 0); colDataSetNULL(pOutputData, 0);
} else { } else {
colDataSetVal(pOutputData, 0, (char *)&val, false); SCL_ERR_RET(colDataSetVal(pOutputData, 0, (char *)&val, false));
} }
pOutput->numOfRows = 1; pOutput->numOfRows = 1;
@ -2341,7 +2427,7 @@ int32_t spreadScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
colDataSetNULL(pOutputData, 0); colDataSetNULL(pOutputData, 0);
} else { } else {
double result = max - min; double result = max - min;
colDataSetVal(pOutputData, 0, (char *)&result, false); SCL_ERR_RET(colDataSetVal(pOutputData, 0, (char *)&result, false));
} }
pOutput->numOfRows = 1; pOutput->numOfRows = 1;
@ -2544,7 +2630,7 @@ int32_t stateCountScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalar
} else { } else {
count = 0; count = 0;
} }
colDataSetVal(pOutputData, i, (char *)&out, false); SCL_ERR_RET(colDataSetVal(pOutputData, i, (char *)&out, false));
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
@ -2568,7 +2654,7 @@ int32_t stateDurationScalarFunction(SScalarParam *pInput, int32_t inputNum, SSca
if (ret) { if (ret) {
out = 0; out = 0;
} }
colDataSetVal(pOutputData, i, (char *)&out, false); SCL_ERR_RET(colDataSetVal(pOutputData, i, (char *)&out, false));
} }
pOutput->numOfRows = pInput->numOfRows; pOutput->numOfRows = pInput->numOfRows;
@ -2599,7 +2685,7 @@ typedef struct SHistoFuncBin {
double percentage; double percentage;
} SHistoFuncBin; } SHistoFuncBin;
static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *binDescStr, int8_t binType, static int32_t getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *binDescStr, int8_t binType,
bool normalized) { bool normalized) {
cJSON *binDesc = cJSON_Parse(binDescStr); cJSON *binDesc = cJSON_Parse(binDescStr);
int32_t numOfBins; int32_t numOfBins;
@ -2609,7 +2695,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
int32_t startIndex; int32_t startIndex;
if (numOfParams != 4) { if (numOfParams != 4) {
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
cJSON *start = cJSON_GetObjectItem(binDesc, "start"); cJSON *start = cJSON_GetObjectItem(binDesc, "start");
@ -2620,18 +2706,18 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) { if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) {
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_RET(TSDB_CODE_SUCCESS);
} }
if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000 if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) || if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) ||
(factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) { (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) {
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
int32_t counter = (int32_t)count->valueint; int32_t counter = (int32_t)count->valueint;
@ -2644,19 +2730,23 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
} }
intervals = taosMemoryCalloc(numOfBins, sizeof(double)); intervals = taosMemoryCalloc(numOfBins, sizeof(double));
if (NULL == intervals) {
cJSON_Delete(binDesc);
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
if (cJSON_IsNumber(width) && factor == NULL && binType == LINEAR_BIN) { if (cJSON_IsNumber(width) && factor == NULL && binType == LINEAR_BIN) {
// linear bin process // linear bin process
if (width->valuedouble == 0) { if (width->valuedouble == 0) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
for (int i = 0; i < counter + 1; ++i) { for (int i = 0; i < counter + 1; ++i) {
intervals[startIndex] = start->valuedouble + i * width->valuedouble; intervals[startIndex] = start->valuedouble + i * width->valuedouble;
if (isinf(intervals[startIndex])) { if (isinf(intervals[startIndex])) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
startIndex++; startIndex++;
} }
@ -2665,26 +2755,26 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
if (start->valuedouble == 0) { if (start->valuedouble == 0) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) { if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
for (int i = 0; i < counter + 1; ++i) { for (int i = 0; i < counter + 1; ++i) {
intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0); intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0);
if (isinf(intervals[startIndex])) { if (isinf(intervals[startIndex])) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
startIndex++; startIndex++;
} }
} else { } else {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
if (infinity->valueint == true) { if (infinity->valueint == true) {
@ -2692,7 +2782,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
intervals[numOfBins - 1] = INFINITY; intervals[numOfBins - 1] = INFINITY;
// in case of desc bin orders, -inf/inf should be swapped // in case of desc bin orders, -inf/inf should be swapped
if (numOfBins < 4) { if (numOfBins < 4) {
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
if (intervals[1] > intervals[numOfBins - 2]) { if (intervals[1] > intervals[numOfBins - 2]) {
TSWAP(intervals[0], intervals[numOfBins - 1]); TSWAP(intervals[0], intervals[numOfBins - 1]);
@ -2701,15 +2791,19 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
} else if (cJSON_IsArray(binDesc)) { /* user input bins */ } else if (cJSON_IsArray(binDesc)) { /* user input bins */
if (binType != USER_INPUT_BIN) { if (binType != USER_INPUT_BIN) {
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
numOfBins = cJSON_GetArraySize(binDesc); numOfBins = cJSON_GetArraySize(binDesc);
intervals = taosMemoryCalloc(numOfBins, sizeof(double)); intervals = taosMemoryCalloc(numOfBins, sizeof(double));
if (NULL == intervals) {
cJSON_Delete(binDesc);
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
cJSON *bin = binDesc->child; cJSON *bin = binDesc->child;
if (bin == NULL) { if (bin == NULL) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
int i = 0; int i = 0;
while (bin) { while (bin) {
@ -2717,23 +2811,26 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
if (!cJSON_IsNumber(bin)) { if (!cJSON_IsNumber(bin)) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
if (i != 0 && intervals[i] <= intervals[i - 1]) { if (i != 0 && intervals[i] <= intervals[i - 1]) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_ERR_RET(TSDB_CODE_FAILED);
} }
bin = bin->next; bin = bin->next;
i++; i++;
} }
} else { } else {
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return false; SCL_RET(TSDB_CODE_FAILED);
} }
*binNum = numOfBins - 1; *binNum = numOfBins - 1;
*bins = taosMemoryCalloc(numOfBins, sizeof(SHistoFuncBin)); *bins = taosMemoryCalloc(numOfBins, sizeof(SHistoFuncBin));
if (NULL == bins) {
SCL_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
for (int32_t i = 0; i < *binNum; ++i) { for (int32_t i = 0; i < *binNum; ++i) {
(*bins)[i].lower = intervals[i] < intervals[i + 1] ? intervals[i] : intervals[i + 1]; (*bins)[i].lower = intervals[i] < intervals[i + 1] ? intervals[i] : intervals[i + 1];
(*bins)[i].upper = intervals[i + 1] > intervals[i] ? intervals[i + 1] : intervals[i]; (*bins)[i].upper = intervals[i + 1] > intervals[i] ? intervals[i + 1] : intervals[i];
@ -2743,7 +2840,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc); cJSON_Delete(binDesc);
return true; SCL_RET(TSDB_CODE_SUCCESS);
} }
int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) { int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
@ -2762,9 +2859,10 @@ int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
int64_t normalized = *(int64_t *)(pInput[3].columnData->pData); int64_t normalized = *(int64_t *)(pInput[3].columnData->pData);
int32_t type = GET_PARAM_TYPE(pInput); int32_t type = GET_PARAM_TYPE(pInput);
if (!getHistogramBinDesc(&bins, &numOfBins, binDesc, binType, (bool)normalized)) { int32_t code = getHistogramBinDesc(&bins, &numOfBins, binDesc, binType, (bool)normalized);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFree(binDesc); taosMemoryFree(binDesc);
return TSDB_CODE_FAILED; return code;
} }
taosMemoryFree(binDesc); taosMemoryFree(binDesc);
@ -2796,7 +2894,7 @@ int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
} }
} }
colInfoDataEnsureCapacity(pOutputData, numOfBins, false); SCL_ERR_JRET(colInfoDataEnsureCapacity(pOutputData, numOfBins, false));
for (int32_t k = 0; k < numOfBins; ++k) { for (int32_t k = 0; k < numOfBins; ++k) {
int32_t len; int32_t len;
@ -2809,11 +2907,12 @@ int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
bins[k].upper, bins[k].percentage); bins[k].upper, bins[k].percentage);
} }
varDataSetLen(buf, len); varDataSetLen(buf, len);
colDataSetVal(pOutputData, k, buf, false); SCL_ERR_JRET(colDataSetVal(pOutputData, k, buf, false));
} }
taosMemoryFree(bins);
pOutput->numOfRows = numOfBins; pOutput->numOfRows = numOfBins;
_return:
taosMemoryFree(bins);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -2830,7 +2929,7 @@ int32_t selectScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
} }
char *data = colDataGetData(pInputData, i); char *data = colDataGetData(pInputData, i);
colDataSetVal(pOutputData, i, data, false); SCL_ERR_RET(colDataSetVal(pOutputData, i, data, false));
} }
pOutput->numOfRows = 1; pOutput->numOfRows = 1;

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

File diff suppressed because it is too large Load Diff

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

@ -55,21 +55,27 @@ void flttInitLogFile() {
tsAsyncLog = 0; tsAsyncLog = 0;
qDebugFlag = 159; qDebugFlag = 159;
strcpy(tsLogDir, TD_LOG_DIR_PATH); (void)strcpy(tsLogDir, TD_LOG_DIR_PATH);
if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) { if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) {
printf("failed to open log file in directory:%s\n", tsLogDir); printf("failed to open log file in directory:%s\n", tsLogDir);
} }
} }
void flttMakeValueNode(SNode **pNode, int32_t dataType, void *value) { int32_t flttMakeValueNode(SNode **pNode, int32_t dataType, void *value) {
SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_VALUE); SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == node) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
SValueNode *vnode = (SValueNode *)node; SValueNode *vnode = (SValueNode *)node;
vnode->node.resType.type = dataType; vnode->node.resType.type = dataType;
if (IS_VAR_DATA_TYPE(dataType)) { if (IS_VAR_DATA_TYPE(dataType)) {
vnode->datum.p = (char *)taosMemoryMalloc(varDataTLen(value)); vnode->datum.p = (char *)taosMemoryMalloc(varDataTLen(value));
varDataCopy(vnode->datum.p, value); if (NULL == vnode->datum.p) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
(void)varDataCopy(vnode->datum.p, value);
vnode->node.resType.bytes = varDataLen(value); vnode->node.resType.bytes = varDataLen(value);
} else { } else {
vnode->node.resType.bytes = tDataTypes[dataType].bytes; vnode->node.resType.bytes = tDataTypes[dataType].bytes;
@ -77,13 +83,17 @@ void flttMakeValueNode(SNode **pNode, int32_t dataType, void *value) {
} }
*pNode = (SNode *)vnode; *pNode = (SNode *)vnode;
FLT_RET(TSDB_CODE_SUCCESS);
} }
void flttMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, int32_t dataBytes, int32_t rowNum, int32_t flttMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, int32_t dataBytes, int32_t rowNum,
void *value) { void *value) {
static uint64_t dbidx = 0; static uint64_t dbidx = 0;
SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_COLUMN); SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == node) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
SColumnNode *rnode = (SColumnNode *)node; SColumnNode *rnode = (SColumnNode *)node;
rnode->node.resType.type = dataType; rnode->node.resType.type = dataType;
rnode->node.resType.bytes = dataBytes; rnode->node.resType.bytes = dataBytes;
@ -96,23 +106,27 @@ void flttMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, in
rnode->colId = 3; rnode->colId = 3;
*pNode = (SNode *)rnode; *pNode = (SNode *)rnode;
return; FLT_RET(TSDB_CODE_SUCCESS);
} }
if (NULL == *block) { if (NULL == *block) {
SSDataBlock *res = createDataBlock(); SSDataBlock *res = createDataBlock();
for (int32_t i = 0; i < 2; ++i) { for (int32_t i = 0; i < 2; ++i) {
SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_NULL, 10, 1 + i); SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_NULL, 10, 1 + i);
blockDataAppendColInfo(res, &idata); FLT_ERR_RET(blockDataAppendColInfo(res, &idata));
} }
SColumnInfoData idata = createColumnInfoData(dataType, dataBytes, 3); SColumnInfoData idata = createColumnInfoData(dataType, dataBytes, 3);
blockDataAppendColInfo(res, &idata); FLT_ERR_RET(blockDataAppendColInfo(res, &idata));
blockDataEnsureCapacity(res, rowNum); FLT_ERR_RET(blockDataEnsureCapacity(res, rowNum));
SColumnInfoData *pColumn = (SColumnInfoData *)taosArrayGetLast(res->pDataBlock); SColumnInfoData *pColumn = (SColumnInfoData *)taosArrayGetLast(res->pDataBlock);
if (NULL == pColumn) {
fltError("fail to get the last task, num:%d", (int32_t)taosArrayGetSize(res->pDataBlock));
FLT_ERR_RET(TSDB_CODE_QRY_SYS_ERROR);
}
for (int32_t i = 0; i < rowNum; ++i) { for (int32_t i = 0; i < rowNum; ++i) {
colDataSetVal(pColumn, i, (const char *)value, false); FLT_ERR_RET(colDataSetVal(pColumn, i, (const char *)value, false));
if (IS_VAR_DATA_TYPE(dataType)) { if (IS_VAR_DATA_TYPE(dataType)) {
value = (char *)value + varDataTLen(value); value = (char *)value + varDataTLen(value);
} else { } else {
@ -130,13 +144,16 @@ void flttMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, in
int32_t idx = taosArrayGetSize(res->pDataBlock); int32_t idx = taosArrayGetSize(res->pDataBlock);
SColumnInfoData idata = createColumnInfoData(dataType, dataBytes, 1 + idx); SColumnInfoData idata = createColumnInfoData(dataType, dataBytes, 1 + idx);
blockDataAppendColInfo(res, &idata); FLT_ERR_RET(blockDataAppendColInfo(res, &idata));
blockDataEnsureCapacity(res, rowNum); FLT_ERR_RET(blockDataEnsureCapacity(res, rowNum));
SColumnInfoData *pColumn = (SColumnInfoData *)taosArrayGetLast(res->pDataBlock); SColumnInfoData *pColumn = (SColumnInfoData *)taosArrayGetLast(res->pDataBlock);
if (NULL == pColumn) {
fltError("fail to get the last task, num:%d", (int32_t)taosArrayGetSize(res->pDataBlock));
FLT_ERR_RET(TSDB_CODE_QRY_SYS_ERROR);
}
for (int32_t i = 0; i < rowNum; ++i) { for (int32_t i = 0; i < rowNum; ++i) {
colDataSetVal(pColumn, i, (const char *)value, false); FLT_ERR_RET(colDataSetVal(pColumn, i, (const char *)value, false));
if (IS_VAR_DATA_TYPE(dataType)) { if (IS_VAR_DATA_TYPE(dataType)) {
value = (char *)value + varDataTLen(value); value = (char *)value + varDataTLen(value);
} else { } else {
@ -149,10 +166,14 @@ void flttMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, in
} }
*pNode = (SNode *)rnode; *pNode = (SNode *)rnode;
FLT_RET(TSDB_CODE_SUCCESS);
} }
void flttMakeOpNode(SNode **pNode, EOperatorType opType, int32_t resType, SNode *pLeft, SNode *pRight) { int32_t flttMakeOpNode(SNode **pNode, EOperatorType opType, int32_t resType, SNode *pLeft, SNode *pRight) {
SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_OPERATOR); SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_OPERATOR);
if (NULL == node) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
SOperatorNode *onode = (SOperatorNode *)node; SOperatorNode *onode = (SOperatorNode *)node;
onode->node.resType.type = resType; onode->node.resType.type = resType;
onode->node.resType.bytes = tDataTypes[resType].bytes; onode->node.resType.bytes = tDataTypes[resType].bytes;
@ -162,10 +183,14 @@ void flttMakeOpNode(SNode **pNode, EOperatorType opType, int32_t resType, SNode
onode->pRight = pRight; onode->pRight = pRight;
*pNode = (SNode *)onode; *pNode = (SNode *)onode;
FLT_RET(TSDB_CODE_SUCCESS);
} }
void flttMakeLogicNode(SNode **pNode, ELogicConditionType opType, SNode **nodeList, int32_t nodeNum) { int32_t flttMakeLogicNode(SNode **pNode, ELogicConditionType opType, SNode **nodeList, int32_t nodeNum) {
SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_LOGIC_CONDITION); SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_LOGIC_CONDITION);
if (NULL == node) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
SLogicConditionNode *onode = (SLogicConditionNode *)node; SLogicConditionNode *onode = (SLogicConditionNode *)node;
onode->condType = opType; onode->condType = opType;
onode->node.resType.type = TSDB_DATA_TYPE_BOOL; onode->node.resType.type = TSDB_DATA_TYPE_BOOL;
@ -173,14 +198,18 @@ void flttMakeLogicNode(SNode **pNode, ELogicConditionType opType, SNode **nodeLi
onode->pParameterList = nodesMakeList(); onode->pParameterList = nodesMakeList();
for (int32_t i = 0; i < nodeNum; ++i) { for (int32_t i = 0; i < nodeNum; ++i) {
nodesListAppend(onode->pParameterList, nodeList[i]); FLT_ERR_RET(nodesListAppend(onode->pParameterList, nodeList[i]));
} }
*pNode = (SNode *)onode; *pNode = (SNode *)onode;
FLT_RET(TSDB_CODE_SUCCESS);
} }
void flttMakeLogicNodeFromList(SNode **pNode, ELogicConditionType opType, SNodeList *nodeList) { int32_t flttMakeLogicNodeFromList(SNode **pNode, ELogicConditionType opType, SNodeList *nodeList) {
SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_LOGIC_CONDITION); SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_LOGIC_CONDITION);
if (NULL == node) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
SLogicConditionNode *onode = (SLogicConditionNode *)node; SLogicConditionNode *onode = (SLogicConditionNode *)node;
onode->condType = opType; onode->condType = opType;
onode->node.resType.type = TSDB_DATA_TYPE_BOOL; onode->node.resType.type = TSDB_DATA_TYPE_BOOL;
@ -189,15 +218,20 @@ void flttMakeLogicNodeFromList(SNode **pNode, ELogicConditionType opType, SNodeL
onode->pParameterList = nodeList; onode->pParameterList = nodeList;
*pNode = (SNode *)onode; *pNode = (SNode *)onode;
FLT_RET(TSDB_CODE_SUCCESS);
} }
void flttMakeListNode(SNode **pNode, SNodeList *list, int32_t resType) { int32_t flttMakeListNode(SNode **pNode, SNodeList *list, int32_t resType) {
SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_NODE_LIST); SNode *node = (SNode *)nodesMakeNode(QUERY_NODE_NODE_LIST);
if (NULL == node) {
FLT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
SNodeListNode *lnode = (SNodeListNode *)node; SNodeListNode *lnode = (SNodeListNode *)node;
lnode->node.resType.type = resType; lnode->node.resType.type = resType;
lnode->pNodeList = list; lnode->pNodeList = list;
*pNode = (SNode *)lnode; *pNode = (SNode *)lnode;
FLT_RET(TSDB_CODE_SUCCESS);
} }
void initScalarParam(SScalarParam *pParam) { void initScalarParam(SScalarParam *pParam) {
@ -213,17 +247,21 @@ TEST(timerangeTest, greater) {
SScalarParam res; SScalarParam res;
initScalarParam(&res); initScalarParam(&res);
int32_t code = TSDB_CODE_SUCCESS;
int64_t tsmall = 222, tbig = 333; int64_t tsmall = 222, tbig = 333;
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall); ASSERT_EQ(code, 0);
flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval); code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall);
ASSERT_EQ(code, 0);
code = flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval);
ASSERT_EQ(code, 0);
// SFilterInfo *filter = NULL; // SFilterInfo *filter = NULL;
// int32_t code = filterInitFromNode(opNode1, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP); // int32_t code = filterInitFromNode(opNode1, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP);
// ASSERT_EQ(code, 0); // ASSERT_EQ(code, 0);
STimeWindow win = {0}; STimeWindow win = {0};
bool isStrict = false; bool isStrict = false;
int32_t code = filterGetTimeRange(opNode1, &win, &isStrict); code = filterGetTimeRange(opNode1, &win, &isStrict);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
ASSERT_EQ(isStrict, true); ASSERT_EQ(isStrict, true);
ASSERT_EQ(win.skey, tsmall + 1); ASSERT_EQ(win.skey, tsmall + 1);
@ -237,25 +275,33 @@ TEST(timerangeTest, greater_and_lower) {
bool eRes[5] = {false, false, true, true, true}; bool eRes[5] = {false, false, true, true, true};
SScalarParam res; SScalarParam res;
initScalarParam(&res); initScalarParam(&res);
int32_t code = TSDB_CODE_SUCCESS;
int64_t tsmall = 222, tbig = 333; int64_t tsmall = 222, tbig = 333;
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall); ASSERT_EQ(code, 0);
flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval); code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall);
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); ASSERT_EQ(code, 0);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tbig); code = flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval);
flttMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval); ASSERT_EQ(code, 0);
code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
ASSERT_EQ(code, 0);
code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tbig);
ASSERT_EQ(code, 0);
code = flttMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval);
ASSERT_EQ(code, 0);
SNode *list[2] = {0}; SNode *list[2] = {0};
list[0] = opNode1; list[0] = opNode1;
list[1] = opNode2; list[1] = opNode2;
flttMakeLogicNode(&logicNode, LOGIC_COND_TYPE_AND, list, 2); code = flttMakeLogicNode(&logicNode, LOGIC_COND_TYPE_AND, list, 2);
ASSERT_EQ(code, 0);
// SFilterInfo *filter = NULL; // SFilterInfo *filter = NULL;
// int32_t code = filterInitFromNode(logicNode, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP); // int32_t code = filterInitFromNode(logicNode, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP);
// ASSERT_EQ(code, 0); // ASSERT_EQ(code, 0);
STimeWindow win = {0}; STimeWindow win = {0};
bool isStrict = false; bool isStrict = false;
int32_t code = filterGetTimeRange(logicNode, &win, &isStrict); code = filterGetTimeRange(logicNode, &win, &isStrict);
ASSERT_EQ(isStrict, true); ASSERT_EQ(isStrict, true);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
ASSERT_EQ(win.skey, tsmall + 1); ASSERT_EQ(win.skey, tsmall + 1);
@ -269,25 +315,33 @@ TEST(timerangeTest, greater_equal_and_lower_equal) {
bool eRes[5] = {false, false, true, true, true}; bool eRes[5] = {false, false, true, true, true};
SScalarParam res; SScalarParam res;
initScalarParam(&res); initScalarParam(&res);
int32_t code = TSDB_CODE_SUCCESS;
int64_t tsmall = 222, tbig = 333; int64_t tsmall = 222, tbig = 333;
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall); ASSERT_EQ(code, 0);
flttMakeOpNode(&opNode1, OP_TYPE_GREATER_EQUAL, TSDB_DATA_TYPE_BOOL, pcol, pval); code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall);
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); ASSERT_EQ(code, 0);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tbig); code = flttMakeOpNode(&opNode1, OP_TYPE_GREATER_EQUAL, TSDB_DATA_TYPE_BOOL, pcol, pval);
flttMakeOpNode(&opNode2, OP_TYPE_LOWER_EQUAL, TSDB_DATA_TYPE_BOOL, pcol, pval); ASSERT_EQ(code, 0);
code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
ASSERT_EQ(code, 0);
code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tbig);
ASSERT_EQ(code, 0);
code = flttMakeOpNode(&opNode2, OP_TYPE_LOWER_EQUAL, TSDB_DATA_TYPE_BOOL, pcol, pval);
ASSERT_EQ(code, 0);
SNode *list[2] = {0}; SNode *list[2] = {0};
list[0] = opNode1; list[0] = opNode1;
list[1] = opNode2; list[1] = opNode2;
flttMakeLogicNode(&logicNode, LOGIC_COND_TYPE_AND, list, 2); code = flttMakeLogicNode(&logicNode, LOGIC_COND_TYPE_AND, list, 2);
ASSERT_EQ(code, 0);
// SFilterInfo *filter = NULL; // SFilterInfo *filter = NULL;
// int32_t code = filterInitFromNode(logicNode, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP); // int32_t code = filterInitFromNode(logicNode, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP);
// ASSERT_EQ(code, 0); // ASSERT_EQ(code, 0);
STimeWindow win = {0}; STimeWindow win = {0};
bool isStrict = false; bool isStrict = false;
int32_t code = filterGetTimeRange(logicNode, &win, &isStrict); code = filterGetTimeRange(logicNode, &win, &isStrict);
ASSERT_EQ(isStrict, true); ASSERT_EQ(isStrict, true);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
ASSERT_EQ(win.skey, tsmall); ASSERT_EQ(win.skey, tsmall);
@ -301,42 +355,58 @@ TEST(timerangeTest, greater_and_lower_not_strict) {
bool eRes[5] = {false, false, true, true, true}; bool eRes[5] = {false, false, true, true, true};
SScalarParam res; SScalarParam res;
initScalarParam(&res); initScalarParam(&res);
int32_t code = TSDB_CODE_SUCCESS;
int64_t tsmall1 = 222, tbig1 = 333; int64_t tsmall1 = 222, tbig1 = 333;
int64_t tsmall2 = 444, tbig2 = 555; int64_t tsmall2 = 444, tbig2 = 555;
SNode *list[2] = {0}; SNode *list[2] = {0};
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall1); ASSERT_EQ(code, 0);
flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval); code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall1);
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); ASSERT_EQ(code, 0);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tbig1); code = flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval);
flttMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval); ASSERT_EQ(code, 0);
code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
ASSERT_EQ(code, 0);
code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tbig1);
ASSERT_EQ(code, 0);
code = flttMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval);
ASSERT_EQ(code, 0);
list[0] = opNode1; list[0] = opNode1;
list[1] = opNode2; list[1] = opNode2;
flttMakeLogicNode(&logicNode1, LOGIC_COND_TYPE_AND, list, 2); code = flttMakeLogicNode(&logicNode1, LOGIC_COND_TYPE_AND, list, 2);
ASSERT_EQ(code, 0);
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall2); ASSERT_EQ(code, 0);
flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval); code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tsmall2);
flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL); ASSERT_EQ(code, 0);
flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tbig2); code = flttMakeOpNode(&opNode1, OP_TYPE_GREATER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval);
flttMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval); ASSERT_EQ(code, 0);
code = flttMakeColumnNode(&pcol, NULL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 0, NULL);
ASSERT_EQ(code, 0);
code = flttMakeValueNode(&pval, TSDB_DATA_TYPE_TIMESTAMP, &tbig2);
ASSERT_EQ(code, 0);
code = flttMakeOpNode(&opNode2, OP_TYPE_LOWER_THAN, TSDB_DATA_TYPE_BOOL, pcol, pval);
ASSERT_EQ(code, 0);
list[0] = opNode1; list[0] = opNode1;
list[1] = opNode2; list[1] = opNode2;
flttMakeLogicNode(&logicNode2, LOGIC_COND_TYPE_AND, list, 2); code = flttMakeLogicNode(&logicNode2, LOGIC_COND_TYPE_AND, list, 2);
ASSERT_EQ(code, 0);
list[0] = logicNode1; list[0] = logicNode1;
list[1] = logicNode2; list[1] = logicNode2;
flttMakeLogicNode(&logicNode1, LOGIC_COND_TYPE_OR, list, 2); code = flttMakeLogicNode(&logicNode1, LOGIC_COND_TYPE_OR, list, 2);
ASSERT_EQ(code, 0);
// SFilterInfo *filter = NULL; // SFilterInfo *filter = NULL;
// int32_t code = filterInitFromNode(logicNode, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP); // int32_t code = filterInitFromNode(logicNode, &filter, FLT_OPTION_NO_REWRITE|FLT_OPTION_TIMESTAMP);
// ASSERT_EQ(code, 0); // ASSERT_EQ(code, 0);
STimeWindow win = {0}; STimeWindow win = {0};
bool isStrict = false; bool isStrict = false;
int32_t code = filterGetTimeRange(logicNode1, &win, &isStrict); code = filterGetTimeRange(logicNode1, &win, &isStrict);
ASSERT_EQ(isStrict, false); ASSERT_EQ(isStrict, false);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
ASSERT_EQ(win.skey, tsmall1 + 1); ASSERT_EQ(win.skey, tsmall1 + 1);

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

File diff suppressed because it is too large Load Diff

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

@ -471,6 +471,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_GEO_NOT_SUPPORT_ERROR, "Geometry not support
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_WINDOW_CONDITION, "The time pseudo column is illegally used in the condition of the event window.") TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_WINDOW_CONDITION, "The time pseudo column is illegally used in the condition of the event window.")
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, "Executor internal error") 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_INVALID_JOIN_CONDITION, "Not supported join on condition")
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_FILTER_NOT_SUPPORT_TYPE, "Not supported range type")
// grant // grant
TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED, "License expired") TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED, "License expired")