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

Merge pull request #10333 from taosdata/feature/qnode 

Feature/qnode
This commit is contained in:
dapan1121 2022-02-21 15:18:04 +08:00 committed by GitHub
parent c5854b83d1 7c1e4d319d
commit ac4c61004b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
38 changed files with 3200 additions and 1339 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
include/common/tcompare.h
View File

@ -23,8 +23,12 @@
extern "C" {
#endif
int32_t compareStrPatternComp(const void* pLeft, const void* pRight);
int32_t compareWStrPatternComp(const void* pLeft, const void* pRight);
int32_t compareStrPatternMatch(const void* pLeft, const void* pRight);
int32_t compareStrPatternNotMatch(const void* pLeft, const void* pRight);
int32_t compareWStrPatternMatch(const void* pLeft, const void* pRight);
int32_t compareWStrPatternNotMatch(const void* pLeft, const void* pRight);
__compar_fn_t getComparFunc(int32_t type, int32_t optr);
__compar_fn_t getKeyComparFunc(int32_t keyType, int32_t order);
int32_t doCompare(const char* a, const char* b, int32_t type, size_t size);
@ -33,4 +37,4 @@ int32_t doCompare(const char* a, const char* b, int32_t type, size_t size)
}
#endif
#endif /*_TD_TCOMPARE_H_*/
#endif /*_TD_TCOMPARE_H_*/

10
include/common/tname.h
View File

@ -16,6 +16,11 @@
#ifndef TDENGINE_TNAME_H
#define TDENGINE_TNAME_H
#ifdef __cplusplus
extern "C" {
#endif
#include "tdef.h"
#include "tmsg.h"
@ -59,4 +64,9 @@ int32_t tNameSetAcctId(SName* dst, int32_t acctId);
SSchema createSchema(uint8_t type, int32_t bytes, int32_t colId, const char* name);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_TNAME_H

13
include/common/ttypes.h
View File

@ -78,9 +78,12 @@ typedef struct {
case TSDB_DATA_TYPE_UINT: \
(_v) = (_finalType)GET_UINT32_VAL(_data); \
break; \
default: \
case TSDB_DATA_TYPE_INT: \
(_v) = (_finalType)GET_INT32_VAL(_data); \
break; \
default: \
(_v) = (_finalType)varDataLen(_data); \
break; \
} \
} while (0)
@ -115,9 +118,11 @@ typedef struct {
case TSDB_DATA_TYPE_UINT: \
*(uint32_t *)(_v) = (uint32_t)(_data); \
break; \
default: \
case TSDB_DATA_TYPE_INT: \
*(int32_t *)(_v) = (int32_t)(_data); \
break; \
default: \
break; \
} \
} while (0)
@ -138,6 +143,9 @@ typedef struct {
#define IS_VALID_FLOAT(_t) ((_t) >= -FLT_MAX && (_t) <= FLT_MAX)
#define IS_VALID_DOUBLE(_t) ((_t) >= -DBL_MAX && (_t) <= DBL_MAX)
#define IS_CONVERT_AS_SIGNED(_t) (IS_SIGNED_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
#define IS_CONVERT_AS_UNSIGNED(_t) (IS_UNSIGNED_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL))
static FORCE_INLINE bool isNull(const void *val, int32_t type) {
switch (type) {
case TSDB_DATA_TYPE_BOOL:
@ -205,6 +213,7 @@ void* getDataMax(int32_t type);
#define SET_DOUBLE_NULL(v) (*(uint64_t *)(v) = TSDB_DATA_DOUBLE_NULL)
#define SET_BIGINT_NULL(v) (*(uint64_t *)(v) = TSDB_DATA_BIGINT_NULL)
#ifdef __cplusplus
}

13
include/libs/function/function.h
View File

@ -226,13 +226,18 @@ typedef struct SAggFunctionInfo {
int32_t (*dataReqFunc)(SqlFunctionCtx *pCtx, STimeWindow* w, int32_t colId);
} SAggFunctionInfo;
struct SScalarFuncParam;
typedef struct SScalarParam {
void* data;
int32_t num;
int32_t type;
int32_t bytes;
} SScalarParam;
typedef struct SScalarFunctionInfo {
char name[FUNCTIONS_NAME_MAX_LENGTH];
int8_t type; // scalar function or aggregation function
uint32_t functionId; // index of scalar function
void (*process)(struct SScalarFuncParam* pOutput, size_t numOfInput, const struct SScalarFuncParam *pInput);
void (*process)(struct SScalarParam* pOutput, size_t numOfInput, const struct SScalarParam *pInput);
} SScalarFunctionInfo;
typedef struct SMultiFunctionsDesc {
@ -285,10 +290,6 @@ int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num);
bool isRowEntryCompleted(struct SResultRowEntryInfo* pEntry);
bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry);
struct SScalarFunctionSupport* createScalarFuncSupport(int32_t num);
void destroyScalarFuncSupport(struct SScalarFunctionSupport* pSupport, int32_t num);
struct SScalarFunctionSupport* getScalarFuncSupport(struct SScalarFunctionSupport* pSupport, int32_t index);
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// fill api
struct SFillInfo;

9
include/libs/function/functionMgt.h
View File

@ -21,6 +21,7 @@ extern "C" {
#endif
#include "querynodes.h"
#include "function.h"
typedef enum EFunctionType {
// aggregate function
@ -117,6 +118,13 @@ typedef struct SFuncExecFuncs {
FExecFinalize finalize;
} SFuncExecFuncs;
typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
typedef struct SScalarFuncExecFuncs {
FScalarExecProcess process;
} SScalarFuncExecFuncs;
int32_t fmFuncMgtInit();
int32_t fmGetFuncInfo(const char* pFuncName, int32_t* pFuncId, int32_t* pFuncType);
@ -134,6 +142,7 @@ bool fmIsTimeorderFunc(int32_t funcId);
int32_t fmFuncScanType(int32_t funcId);
int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet);
#ifdef __cplusplus
}

3
include/libs/nodes/querynodes.h
View File

@ -306,9 +306,10 @@ bool nodesIsJsonOp(const SOperatorNode* pOp);
bool nodesIsTimeorderQuery(const SNode* pQuery);
bool nodesIsTimelineQuery(const SNode* pQuery);
void *nodesGetValueFromNode(SValueNode *pNode);
#ifdef __cplusplus
}
#endif
#endif /*_TD_QUERY_NODES_H_*/
#endif /*_TD_QUERY_NODES_H_*/

34
include/libs/scalar/filter.h Normal file
View File

@ -0,0 +1,34 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_FILTER_H
#define TDENGINE_FILTER_H
#ifdef __cplusplus
extern "C" {
#endif
typedef struct SFilterInfo SFilterInfo;
typedef struct SFilterColumnParam{
int32_t numOfCols;
SArray* pDataBlock;
} SFilterColumnParam;
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_FILTER_H

41
include/libs/scalar/scalar.h Normal file
View File

@ -0,0 +1,41 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_SCALAR_H
#define TDENGINE_SCALAR_H
#ifdef __cplusplus
extern "C" {
#endif
#include "function.h"
#include "nodes.h"
#include "querynodes.h"
typedef struct SFilterInfo SFilterInfo;
int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes);
int32_t scalarCalculate(SNode *pNode, SSDataBlock *pSrc, SScalarParam *pDst);
int32_t scalarGetOperatorParamNum(EOperatorType type);
int32_t vectorGetConvertType(int32_t type1, int32_t type2);
int32_t vectorConvertImpl(SScalarParam* pIn, SScalarParam* pOut);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_SCALAR_H

21
include/util/compare.h
View File

@ -49,10 +49,19 @@ int32_t WCSPatternMatch(const wchar_t *pattern, const wchar_t *str, size_t size,
int32_t taosArrayCompareString(const void *a, const void *b);
int32_t setCompareBytes1(const void *pLeft, const void *pRight);
int32_t setCompareBytes2(const void *pLeft, const void *pRight);
int32_t setCompareBytes4(const void *pLeft, const void *pRight);
int32_t setCompareBytes8(const void *pLeft, const void *pRight);
int32_t setChkInBytes1(const void *pLeft, const void *pRight);
int32_t setChkInBytes2(const void *pLeft, const void *pRight);
int32_t setChkInBytes4(const void *pLeft, const void *pRight);
int32_t setChkInBytes8(const void *pLeft, const void *pRight);
int32_t setChkNotInBytes1(const void *pLeft, const void *pRight);
int32_t setChkNotInBytes2(const void *pLeft, const void *pRight);
int32_t setChkNotInBytes4(const void *pLeft, const void *pRight);
int32_t setChkNotInBytes8(const void *pLeft, const void *pRight);
int32_t compareChkInString(const void *pLeft, const void *pRight);
int32_t compareChkNotInString(const void *pLeft, const void *pRight);
int32_t compareInt8Val(const void *pLeft, const void *pRight);
int32_t compareInt16Val(const void *pLeft, const void *pRight);
@ -74,7 +83,6 @@ int32_t compareStrRegexComp(const void *pLeft, const void *pRight);
int32_t compareStrRegexCompMatch(const void *pLeft, const void *pRight);
int32_t compareStrRegexCompNMatch(const void *pLeft, const void *pRight);
int32_t compareFindItemInSet(const void *pLeft, const void *pRight);
int32_t compareInt8ValDesc(const void *pLeft, const void *pRight);
int32_t compareInt16ValDesc(const void *pLeft, const void *pRight);
@ -92,6 +100,9 @@ int32_t compareUint64ValDesc(const void *pLeft, const void *pRight);
int32_t compareLenPrefixedStrDesc(const void *pLeft, const void *pRight);
int32_t compareLenPrefixedWStrDesc(const void *pLeft, const void *pRight);
__compar_fn_t getComparFunc(int32_t type, int32_t optr);
#ifdef __cplusplus
}
#endif

18
include/util/tdef.h
View File

@ -118,16 +118,18 @@ do { \
#define TSDB_RELATION_GREATER_EQUAL 5
#define TSDB_RELATION_NOT_EQUAL 6
#define TSDB_RELATION_LIKE 7
#define TSDB_RELATION_ISNULL 8
#define TSDB_RELATION_NOTNULL 9
#define TSDB_RELATION_IN 10
#define TSDB_RELATION_NOT_LIKE 8
#define TSDB_RELATION_ISNULL 9
#define TSDB_RELATION_NOTNULL 10
#define TSDB_RELATION_IN 11
#define TSDB_RELATION_NOT_IN 12
#define TSDB_RELATION_AND 11
#define TSDB_RELATION_OR 12
#define TSDB_RELATION_NOT 13
#define TSDB_RELATION_AND 13
#define TSDB_RELATION_OR 14
#define TSDB_RELATION_NOT 15
#define TSDB_RELATION_MATCH 14
#define TSDB_RELATION_NMATCH 15
#define TSDB_RELATION_MATCH 16
#define TSDB_RELATION_NMATCH 17
#define TSDB_BINARY_OP_ADD 4000
#define TSDB_BINARY_OP_SUBTRACT 4001

4
include/util/thash.h
View File

@ -55,6 +55,8 @@ uint32_t taosIntHash_64(const char *key, uint32_t len);
_hash_fn_t taosGetDefaultHashFunction(int32_t type);
_equal_fn_t taosGetDefaultEqualFunction(int32_t type);
typedef struct SHashNode {
struct SHashNode *next;
uint32_t hashVal; // the hash value of key
@ -258,6 +260,8 @@ void* taosHashAcquire(SHashObj *pHashObj, const void *key, size_t keyLen);
*/
void taosHashRelease(SHashObj *pHashObj, void *p);
void taosHashSetEqualFp(SHashObj *pHashObj, _equal_fn_t fp);
#ifdef __cplusplus
}

1
include/util/types.h
View File

@ -66,6 +66,7 @@ static FORCE_INLINE double taos_align_get_double(const char *pBuf) {
// #else
#define GET_FLOAT_VAL(x) (*(float *)(x))
#define GET_DOUBLE_VAL(x) (*(double *)(x))
#define SET_BIGINT_VAL(x, y) { (*(int64_t *)(x)) = (int64_t)(y); }
#define SET_FLOAT_VAL(x, y) { (*(float *)(x)) = (float)(y); }
#define SET_DOUBLE_VAL(x, y) { (*(double *)(x)) = (double)(y); }
#define SET_FLOAT_PTR(x, y) { (*(float *)(x)) = (*(float *)(y)); }

112
source/common/src/tcompare.c
View File

@ -15,116 +15,6 @@
#include "tcompare.h"
int32_t compareStrPatternComp(const void* pLeft, const void* pRight) {
SPatternCompareInfo pInfo = {'%', '_'};
assert(varDataLen(pRight) <= TSDB_MAX_FIELD_LEN);
char *pattern = calloc(varDataLen(pRight) + 1, sizeof(char));
memcpy(pattern, varDataVal(pRight), varDataLen(pRight));
size_t sz = varDataLen(pLeft);
char *buf = malloc(sz + 1);
memcpy(buf, varDataVal(pLeft), sz);
buf[sz] = 0;
int32_t ret = patternMatch(pattern, buf, sz, &pInfo);
free(buf);
free(pattern);
return (ret == TSDB_PATTERN_MATCH) ? 0 : 1;
}
int32_t compareWStrPatternComp(const void* pLeft, const void* pRight) {
SPatternCompareInfo pInfo = {'%', '_'};
assert(varDataLen(pRight) <= TSDB_MAX_FIELD_LEN * TSDB_NCHAR_SIZE);
wchar_t *pattern = calloc(varDataLen(pRight) + 1, sizeof(wchar_t));
memcpy(pattern, varDataVal(pRight), varDataLen(pRight));
int32_t ret = WCSPatternMatch(pattern, varDataVal(pLeft), varDataLen(pLeft)/TSDB_NCHAR_SIZE, &pInfo);
free(pattern);
return (ret == TSDB_PATTERN_MATCH) ? 0 : 1;
}
__compar_fn_t getComparFunc(int32_t type, int32_t optr) {
__compar_fn_t comparFn = NULL;
if (optr == TSDB_RELATION_IN && (type != TSDB_DATA_TYPE_BINARY && type != TSDB_DATA_TYPE_NCHAR)) {
switch (type) {
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_UTINYINT:
return setCompareBytes1;
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_USMALLINT:
return setCompareBytes2;
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_FLOAT:
return setCompareBytes4;
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_UBIGINT:
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_TIMESTAMP:
return setCompareBytes8;
default:
assert(0);
}
}
switch (type) {
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT: comparFn = compareInt8Val; break;
case TSDB_DATA_TYPE_SMALLINT: comparFn = compareInt16Val; break;
case TSDB_DATA_TYPE_INT: comparFn = compareInt32Val; break;
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP: comparFn = compareInt64Val; break;
case TSDB_DATA_TYPE_FLOAT: comparFn = compareFloatVal; break;
case TSDB_DATA_TYPE_DOUBLE: comparFn = compareDoubleVal; break;
case TSDB_DATA_TYPE_BINARY: {
if (optr == TSDB_RELATION_MATCH) {
comparFn = compareStrRegexCompMatch;
} else if (optr == TSDB_RELATION_NMATCH) {
comparFn = compareStrRegexCompNMatch;
} else if (optr == TSDB_RELATION_LIKE) { /* wildcard query using like operator */
comparFn = compareStrPatternComp;
} else if (optr == TSDB_RELATION_IN) {
comparFn = compareFindItemInSet;
} else { /* normal relational comparFn */
comparFn = compareLenPrefixedStr;
}
break;
}
case TSDB_DATA_TYPE_NCHAR: {
if (optr == TSDB_RELATION_MATCH) {
comparFn = compareStrRegexCompMatch;
} else if (optr == TSDB_RELATION_NMATCH) {
comparFn = compareStrRegexCompNMatch;
} else if (optr == TSDB_RELATION_LIKE) {
comparFn = compareWStrPatternComp;
} else if (optr == TSDB_RELATION_IN) {
comparFn = compareFindItemInSet;
} else {
comparFn = compareLenPrefixedWStr;
}
break;
}
case TSDB_DATA_TYPE_UTINYINT: comparFn = compareUint8Val; break;
case TSDB_DATA_TYPE_USMALLINT: comparFn = compareUint16Val;break;
case TSDB_DATA_TYPE_UINT: comparFn = compareUint32Val;break;
case TSDB_DATA_TYPE_UBIGINT: comparFn = compareUint64Val;break;
default:
comparFn = compareInt32Val;
break;
}
return comparFn;
}
__compar_fn_t getKeyComparFunc(int32_t keyType, int32_t order) {
__compar_fn_t comparFn = NULL;
@ -218,4 +108,4 @@ int32_t doCompare(const char* f1, const char* f2, int32_t type, size_t size) {
}
}
}
}
}

3
source/libs/CMakeLists.txt
View File

@ -13,4 +13,5 @@ add_subdirectory(function)
add_subdirectory(qcom)
add_subdirectory(qworker)
add_subdirectory(tfs)
add_subdirectory(nodes)
add_subdirectory(nodes)
add_subdirectory(scalar)

2
source/libs/executor/CMakeLists.txt
View File

@ -13,7 +13,7 @@ add_library(executor STATIC ${EXECUTOR_SRC})
# INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_SOURCE_DIR}/include/libs/executor"
# )
target_link_libraries(executor
PRIVATE os util common function parser planner qcom vnode
PRIVATE os util common function parser planner qcom vnode scalar nodes
)
target_include_directories(

2
source/libs/executor/inc/executorimpl.h
View File

@ -29,9 +29,9 @@ extern "C" {
#include "executil.h"
#include "executor.h"
#include "planner.h"
#include "scalar.h"
#include "taosdef.h"
#include "tarray.h"
#include "tfilter.h"
#include "thash.h"
#include "tlockfree.h"
#include "tpagedbuf.h"

6
source/libs/executor/src/executorimpl.c
View File

@ -2148,7 +2148,7 @@ static int32_t setupQueryRuntimeEnv(STaskRuntimeEnv *pRuntimeEnv, int32_t numOfT
// NOTE: pTableCheckInfo need to update the query time range and the lastKey info
pRuntimeEnv->pTableRetrieveTsMap = taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
pRuntimeEnv->scalarSup = createScalarFuncSupport(pQueryAttr->numOfOutput);
//pRuntimeEnv->scalarSup = createScalarFuncSupport(pQueryAttr->numOfOutput);
if (pRuntimeEnv->scalarSup == NULL || pRuntimeEnv->pResultRowHashTable == NULL || pRuntimeEnv->keyBuf == NULL ||
pRuntimeEnv->prevRow == NULL || pRuntimeEnv->tagVal == NULL) {
@ -2174,7 +2174,7 @@ static int32_t setupQueryRuntimeEnv(STaskRuntimeEnv *pRuntimeEnv, int32_t numOfT
return TSDB_CODE_SUCCESS;
_clean:
destroyScalarFuncSupport(pRuntimeEnv->scalarSup, pRuntimeEnv->pQueryAttr->numOfOutput);
//destroyScalarFuncSupport(pRuntimeEnv->scalarSup, pRuntimeEnv->pQueryAttr->numOfOutput);
tfree(pRuntimeEnv->pResultRowHashTable);
tfree(pRuntimeEnv->keyBuf);
tfree(pRuntimeEnv->prevRow);
@ -2212,7 +2212,7 @@ static void teardownQueryRuntimeEnv(STaskRuntimeEnv *pRuntimeEnv) {
//qDebug("QInfo:0x%"PRIx64" teardown runtime env", pQInfo->qId);
destroyScalarFuncSupport(pRuntimeEnv->scalarSup, pQueryAttr->numOfOutput);
//destroyScalarFuncSupport(pRuntimeEnv->scalarSup, pQueryAttr->numOfOutput);
// destroyUdfInfo(pRuntimeEnv->pUdfInfo);
destroyResultBuf(pRuntimeEnv->pResultBuf);
doFreeQueryHandle(pRuntimeEnv);

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

@ -8,7 +8,7 @@ AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
ADD_EXECUTABLE(executorTest ${SOURCE_LIST})
TARGET_LINK_LIBRARIES(
executorTest
PUBLIC os util common transport gtest taos qcom executor function planner
PRIVATE os util common transport gtest taos qcom executor function planner scalar nodes
)
TARGET_INCLUDE_DIRECTORIES(

2
source/libs/executor/test/executorTests.cpp
View File

@ -32,6 +32,8 @@
#include "trpc.h"
#include "stub.h"
#include "executor.h"
#include "tmsg.h"
#include "tname.h"
namespace {

6
source/libs/function/inc/tunaryoperator.h
View File

@ -20,10 +20,10 @@
extern "C" {
#endif
#include "tscalarfunction.h"
//#include "tscalarfunction.h"
typedef void (*_unary_scalar_fn_t)(SScalarFuncParam *pLeft, SScalarFuncParam* pOutput);
_unary_scalar_fn_t getUnaryScalarOperatorFn(int32_t binOperator);
//typedef void (*_unary_scalar_fn_t)(SScalarParam *pLeft, SScalarParam* pOutput);
//_unary_scalar_fn_t getUnaryScalarOperatorFn(int32_t binOperator);
#ifdef __cplusplus
}

9
source/libs/function/src/taggfunction.c
View File

@ -13,7 +13,6 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tscalarfunction.h"
#include "os.h"
#include "taosdef.h"
#include "tmsg.h"
@ -3221,6 +3220,7 @@ static void diff_function(SqlFunctionCtx *pCtx) {
}
}
#if 0
char *getArithColumnData(void *param, const char* name, int32_t colId) {
SScalarFunctionSupport *pSupport = (SScalarFunctionSupport *)param;
@ -3235,15 +3235,16 @@ char *getArithColumnData(void *param, const char* name, int32_t colId) {
assert(index >= 0);
return pSupport->data[index] + pSupport->offset * pSupport->colList[index].bytes;
}
#endif
static void arithmetic_function(SqlFunctionCtx *pCtx) {
GET_RES_INFO(pCtx)->numOfRes += pCtx->size;
SScalarFunctionSupport *pSup = (SScalarFunctionSupport *)pCtx->param[1].pz;
//SScalarFunctionSupport *pSup = (SScalarFunctionSupport *)pCtx->param[1].pz;
SScalarFuncParam output = {0};
SScalarParam output = {0};
output.data = pCtx->pOutput;
evaluateExprNodeTree(pSup->pExprInfo->pExpr, pCtx->size, &output, pSup, getArithColumnData);
//evaluateExprNodeTree(pSup->pExprInfo->pExpr, pCtx->size, &output, pSup, getArithColumnData);
}
#define LIST_MINMAX_N(ctx, minOutput, maxOutput, elemCnt, data, type, tsdbType, numOfNotNullElem) \

488
source/libs/function/src/tbinoperator.c
View File

@ -1,488 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "ttypes.h"
#include "tbinoperator.h"
#include "tcompare.h"
//GET_TYPED_DATA(v, double, pRight->type, (char *)&((right)[i]));
void calc_i32_i32_add(void *left, void *right, int32_t numLeft, int32_t numRight, void *output, int32_t order) {
int32_t *pLeft = (int32_t *)left;
int32_t *pRight = (int32_t *)right;
double * pOutput = (double *)output;
int32_t i = (order == TSDB_ORDER_ASC) ? 0 : TMAX(numLeft, numRight) - 1;
int32_t step = (order == TSDB_ORDER_ASC) ? 1 : -1;
if (numLeft == numRight) {
for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
SET_DOUBLE_NULL(pOutput);
continue;
}
*pOutput = (double)pLeft[i] + pRight[i];
}
} else if (numLeft == 1) {
for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
SET_DOUBLE_NULL(pOutput);
continue;
}
*pOutput = (double)pLeft[0] + pRight[i];
}
} else if (numRight == 1) {
for (; i >= 0 && i < numLeft; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) {
SET_DOUBLE_NULL(pOutput);
continue;
}
*pOutput = (double)pLeft[i] + pRight[0];
}
}
}
typedef double (*_getDoubleValue_fn_t)(void *src, int32_t index);
double getVectorDoubleValue_TINYINT(void *src, int32_t index) {
return (double)*((int8_t *)src + index);
}
double getVectorDoubleValue_UTINYINT(void *src, int32_t index) {
return (double)*((uint8_t *)src + index);
}
double getVectorDoubleValue_SMALLINT(void *src, int32_t index) {
return (double)*((int16_t *)src + index);
}
double getVectorDoubleValue_USMALLINT(void *src, int32_t index) {
return (double)*((uint16_t *)src + index);
}
double getVectorDoubleValue_INT(void *src, int32_t index) {
return (double)*((int32_t *)src + index);
}
double getVectorDoubleValue_UINT(void *src, int32_t index) {
return (double)*((uint32_t *)src + index);
}
double getVectorDoubleValue_BIGINT(void *src, int32_t index) {
return (double)*((int64_t *)src + index);
}
double getVectorDoubleValue_UBIGINT(void *src, int32_t index) {
return (double)*((uint64_t *)src + index);
}
double getVectorDoubleValue_FLOAT(void *src, int32_t index) {
return (double)*((float *)src + index);
}
double getVectorDoubleValue_DOUBLE(void *src, int32_t index) {
return (double)*((double *)src + index);
}
_getDoubleValue_fn_t getVectorDoubleValueFn(int32_t srcType) {
_getDoubleValue_fn_t p = NULL;
if(srcType==TSDB_DATA_TYPE_TINYINT) {
p = getVectorDoubleValue_TINYINT;
}else if(srcType==TSDB_DATA_TYPE_UTINYINT) {
p = getVectorDoubleValue_UTINYINT;
}else if(srcType==TSDB_DATA_TYPE_SMALLINT) {
p = getVectorDoubleValue_SMALLINT;
}else if(srcType==TSDB_DATA_TYPE_USMALLINT) {
p = getVectorDoubleValue_USMALLINT;
}else if(srcType==TSDB_DATA_TYPE_INT) {
p = getVectorDoubleValue_INT;
}else if(srcType==TSDB_DATA_TYPE_UINT) {
p = getVectorDoubleValue_UINT;
}else if(srcType==TSDB_DATA_TYPE_BIGINT) {
p = getVectorDoubleValue_BIGINT;
}else if(srcType==TSDB_DATA_TYPE_UBIGINT) {
p = getVectorDoubleValue_UBIGINT;
}else if(srcType==TSDB_DATA_TYPE_FLOAT) {
p = getVectorDoubleValue_FLOAT;
}else if(srcType==TSDB_DATA_TYPE_DOUBLE) {
p = getVectorDoubleValue_DOUBLE;
}else {
assert(0);
}
return p;
}
typedef void* (*_getValueAddr_fn_t)(void *src, int32_t index);
void* getVectorValueAddr_TINYINT(void *src, int32_t index) {
return (void*)((int8_t *)src + index);
}
void* getVectorValueAddr_UTINYINT(void *src, int32_t index) {
return (void*)((uint8_t *)src + index);
}
void* getVectorValueAddr_SMALLINT(void *src, int32_t index) {
return (void*)((int16_t *)src + index);
}
void* getVectorValueAddr_USMALLINT(void *src, int32_t index) {
return (void*)((uint16_t *)src + index);
}
void* getVectorValueAddr_INT(void *src, int32_t index) {
return (void*)((int32_t *)src + index);
}
void* getVectorValueAddr_UINT(void *src, int32_t index) {
return (void*)((uint32_t *)src + index);
}
void* getVectorValueAddr_BIGINT(void *src, int32_t index) {
return (void*)((int64_t *)src + index);
}
void* getVectorValueAddr_UBIGINT(void *src, int32_t index) {
return (void*)((uint64_t *)src + index);
}
void* getVectorValueAddr_FLOAT(void *src, int32_t index) {
return (void*)((float *)src + index);
}
void* getVectorValueAddr_DOUBLE(void *src, int32_t index) {
return (void*)((double *)src + index);
}
_getValueAddr_fn_t getVectorValueAddrFn(int32_t srcType) {
_getValueAddr_fn_t p = NULL;
if(srcType==TSDB_DATA_TYPE_TINYINT) {
p = getVectorValueAddr_TINYINT;
}else if(srcType==TSDB_DATA_TYPE_UTINYINT) {
p = getVectorValueAddr_UTINYINT;
}else if(srcType==TSDB_DATA_TYPE_SMALLINT) {
p = getVectorValueAddr_SMALLINT;
}else if(srcType==TSDB_DATA_TYPE_USMALLINT) {
p = getVectorValueAddr_USMALLINT;
}else if(srcType==TSDB_DATA_TYPE_INT) {
p = getVectorValueAddr_INT;
}else if(srcType==TSDB_DATA_TYPE_UINT) {
p = getVectorValueAddr_UINT;
}else if(srcType==TSDB_DATA_TYPE_BIGINT) {
p = getVectorValueAddr_BIGINT;
}else if(srcType==TSDB_DATA_TYPE_UBIGINT) {
p = getVectorValueAddr_UBIGINT;
}else if(srcType==TSDB_DATA_TYPE_FLOAT) {
p = getVectorValueAddr_FLOAT;
}else if(srcType==TSDB_DATA_TYPE_DOUBLE) {
p = getVectorValueAddr_DOUBLE;
}else {
assert(0);
}
return p;
}
void vectorAdd(SScalarFuncParam* pLeft, SScalarFuncParam* pRight, void *out, int32_t _ord) {
int32_t i = ((_ord) == TSDB_ORDER_ASC) ? 0 : TMAX(pLeft->num, pRight->num) - 1;
int32_t step = ((_ord) == TSDB_ORDER_ASC) ? 1 : -1;
double *output=(double*)out;
_getValueAddr_fn_t getVectorValueAddrFnLeft = getVectorValueAddrFn(pLeft->type);
_getValueAddr_fn_t getVectorValueAddrFnRight = getVectorValueAddrFn(pRight->type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeft->type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRight->type);
if (pLeft->num == pRight->num) {
for (; i < pRight->num && i >= 0; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, i), pLeft->type) ||
isNull(getVectorValueAddrFnRight(pRight->data, i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output, getVectorDoubleValueFnLeft(pLeft->data, i) + getVectorDoubleValueFnRight(pRight->data, i));
}
} else if (pLeft->num == 1) {
for (; i >= 0 && i < pRight->num; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, 0), pLeft->type) || isNull(getVectorValueAddrFnRight(pRight->data,i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output,getVectorDoubleValueFnLeft(pLeft->data, 0) + getVectorDoubleValueFnRight(pRight->data,i));
}
} else if (pRight->num == 1) {
for (; i >= 0 && i < pLeft->num; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data,i), pLeft->type) || isNull(getVectorValueAddrFnRight(pRight->data,0), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output,getVectorDoubleValueFnLeft(pLeft->data,i) + getVectorDoubleValueFnRight(pRight->data,0));
}
}
}
void vectorSub(SScalarFuncParam* pLeft, SScalarFuncParam* pRight, void *out, int32_t _ord) {
int32_t i = ((_ord) == TSDB_ORDER_ASC) ? 0 : TMAX(pLeft->num, pRight->num) - 1;
int32_t step = ((_ord) == TSDB_ORDER_ASC) ? 1 : -1;
double *output=(double*)out;
_getValueAddr_fn_t getVectorValueAddrFnLeft = getVectorValueAddrFn(pLeft->type);
_getValueAddr_fn_t getVectorValueAddrFnRight = getVectorValueAddrFn(pRight->type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeft->type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRight->type);
if (pLeft->num == pRight->num) {
for (; i < pRight->num && i >= 0; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, i), pLeft->type) ||
isNull(getVectorValueAddrFnRight(pRight->data, i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output, getVectorDoubleValueFnLeft(pLeft->data, i) - getVectorDoubleValueFnRight(pRight->data, i));
}
} else if (pLeft->num == 1) {
for (; i >= 0 && i < pRight->num; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, 0), pLeft->type) || isNull(getVectorValueAddrFnRight(pRight->data,i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output,getVectorDoubleValueFnLeft(pLeft->data, 0) - getVectorDoubleValueFnRight(pRight->data,i));
}
} else if (pRight->num == 1) {
for (; i >= 0 && i < pLeft->num; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data,i), pLeft->type) || isNull(getVectorValueAddrFnRight(pRight->data,0), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output,getVectorDoubleValueFnLeft(pLeft->data,i) - getVectorDoubleValueFnRight(pRight->data,0));
}
}
}
void vectorMultiply(SScalarFuncParam* pLeft, SScalarFuncParam* pRight, void *out, int32_t _ord) {
int32_t i = ((_ord) == TSDB_ORDER_ASC) ? 0 : TMAX(pLeft->num, pRight->num) - 1;
int32_t step = ((_ord) == TSDB_ORDER_ASC) ? 1 : -1;
double *output=(double*)out;
_getValueAddr_fn_t getVectorValueAddrFnLeft = getVectorValueAddrFn(pLeft->type);
_getValueAddr_fn_t getVectorValueAddrFnRight = getVectorValueAddrFn(pRight->type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeft->type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRight->type);
if (pLeft->num == pRight->num) {
for (; i < pRight->num && i >= 0; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, i), pLeft->type) ||
isNull(getVectorValueAddrFnRight(pRight->data, i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output, getVectorDoubleValueFnLeft(pLeft->data, i) * getVectorDoubleValueFnRight(pRight->data, i));
}
} else if (pLeft->num == 1) {
for (; i >= 0 && i < pRight->num; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, 0), pLeft->type) || isNull(getVectorValueAddrFnRight(pRight->data,i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output,getVectorDoubleValueFnLeft(pLeft->data, 0) * getVectorDoubleValueFnRight(pRight->data,i));
}
} else if (pRight->num == 1) {
for (; i >= 0 && i < pLeft->num; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data,i), pLeft->type) || isNull(getVectorValueAddrFnRight(pRight->data,0), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output,getVectorDoubleValueFnLeft(pLeft->data,i) * getVectorDoubleValueFnRight(pRight->data,0));
}
}
}
void vectorDivide(SScalarFuncParam* pLeft, SScalarFuncParam* pRight, void *out, int32_t _ord) {
int32_t i = ((_ord) == TSDB_ORDER_ASC) ? 0 : TMAX(pLeft->num, pRight->num) - 1;
int32_t step = ((_ord) == TSDB_ORDER_ASC) ? 1 : -1;
double *output=(double*)out;
_getValueAddr_fn_t getVectorValueAddrFnLeft = getVectorValueAddrFn(pLeft->type);
_getValueAddr_fn_t getVectorValueAddrFnRight = getVectorValueAddrFn(pRight->type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeft->type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRight->type);
if (pLeft->num == pRight->num) {
for (; i < pRight->num && i >= 0; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, i), pLeft->type) ||
isNull(getVectorValueAddrFnRight(pRight->data, i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output, getVectorDoubleValueFnLeft(pLeft->data, i) / getVectorDoubleValueFnRight(pRight->data, i));
}
} else if (pLeft->num == 1) {
double left = getVectorDoubleValueFnLeft(pLeft->data, 0);
for (; i >= 0 && i < pRight->num; i += step, output += 1) {
if (isNull(&left, pLeft->type) || isNull(getVectorValueAddrFnRight(pRight->data,i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output,left / getVectorDoubleValueFnRight(pRight->data,i));
}
} else if (pRight->num == 1) {
double right = getVectorDoubleValueFnRight(pRight->data, 0);
for (; i >= 0 && i < pLeft->num; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, i), pLeft->type) ||
isNull(&right, pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
SET_DOUBLE_VAL(output, getVectorDoubleValueFnLeft(pLeft->data, i) / right);
}
}
}
void vectorRemainder(SScalarFuncParam* pLeft, SScalarFuncParam* pRight, void *out, int32_t _ord) {
int32_t i = ((_ord) == TSDB_ORDER_ASC) ? 0 : TMAX(pLeft->num, pRight->num) - 1;
int32_t step = ((_ord) == TSDB_ORDER_ASC) ? 1 : -1;
double * output = (double *)out;
_getValueAddr_fn_t getVectorValueAddrFnLeft = getVectorValueAddrFn(pLeft->type);
_getValueAddr_fn_t getVectorValueAddrFnRight = getVectorValueAddrFn(pRight->type);
_getDoubleValue_fn_t getVectorDoubleValueFnLeft = getVectorDoubleValueFn(pLeft->type);
_getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRight->type);
if (pLeft->num == pRight->num) {
for (; i < pRight->num && i >= 0; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, i), pLeft->type) ||
isNull(getVectorValueAddrFnRight(pRight->data, i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
double v, u = 0.0;
GET_TYPED_DATA(v, double, pRight->type, getVectorValueAddrFnRight(pRight->data, i));
if (getComparFunc(TSDB_DATA_TYPE_DOUBLE, 0)(&v, &u) == 0) {
SET_DOUBLE_NULL(output);
continue;
}
double left = getVectorDoubleValueFnLeft(pLeft->data, i);
double right = getVectorDoubleValueFnRight(pRight->data, i);
SET_DOUBLE_VAL(output, left - ((int64_t)(left / right)) * right);
}
} else if (pLeft->num == 1) {
double left = getVectorDoubleValueFnLeft(pLeft->data, 0);
for (; i >= 0 && i < pRight->num; i += step, output += 1) {
if (isNull(&left, pLeft->type) ||
isNull(getVectorValueAddrFnRight(pRight->data, i), pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
double v, u = 0.0;
GET_TYPED_DATA(v, double, pRight->type, getVectorValueAddrFnRight(pRight->data, i));
if (getComparFunc(TSDB_DATA_TYPE_DOUBLE, 0)(&v, &u) == 0) {
SET_DOUBLE_NULL(output);
continue;
}
double right = getVectorDoubleValueFnRight(pRight->data, i);
SET_DOUBLE_VAL(output, left - ((int64_t)(left / right)) * right);
}
} else if (pRight->num == 1) {
double right = getVectorDoubleValueFnRight(pRight->data, 0);
for (; i >= 0 && i < pLeft->num; i += step, output += 1) {
if (isNull(getVectorValueAddrFnLeft(pLeft->data, i), pLeft->type) ||
isNull(&right, pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
double v, u = 0.0;
GET_TYPED_DATA(v, double, pRight->type, getVectorValueAddrFnRight(pRight->data, 0));
if (getComparFunc(TSDB_DATA_TYPE_DOUBLE, 0)(&v, &u) == 0) {
SET_DOUBLE_NULL(output);
continue;
}
double left = getVectorDoubleValueFnLeft(pLeft->data, i);
SET_DOUBLE_VAL(output, left - ((int64_t)(left / right)) * right);
}
}
}
void vectorConcat(SScalarFuncParam* pLeft, SScalarFuncParam* pRight, void *out, int32_t _ord) {
int32_t len = pLeft->bytes + pRight->bytes;
int32_t i = ((_ord) == TSDB_ORDER_ASC) ? 0 : TMAX(pLeft->num, pRight->num) - 1;
int32_t step = ((_ord) == TSDB_ORDER_ASC) ? 1 : -1;
char *output = (char *)out;
if (pLeft->num == pRight->num) {
for (; i < pRight->num && i >= 0; i += step, output += len) {
char* left = POINTER_SHIFT(pLeft->data, pLeft->bytes * i);
char* right = POINTER_SHIFT(pRight->data, pRight->bytes * i);
if (isNull(left, pLeft->type) || isNull(right, pRight->type)) {
setVardataNull(output, TSDB_DATA_TYPE_BINARY);
continue;
}
// todo define a macro
memcpy(varDataVal(output), varDataVal(left), varDataLen(left));
memcpy(varDataVal(output) + varDataLen(left), varDataVal(right), varDataLen(right));
varDataSetLen(output, varDataLen(left) + varDataLen(right));
}
} else if (pLeft->num == 1) {
for (; i >= 0 && i < pRight->num; i += step, output += len) {
char *right = POINTER_SHIFT(pRight->data, pRight->bytes * i);
if (isNull(pLeft->data, pLeft->type) || isNull(right, pRight->type)) {
setVardataNull(output, TSDB_DATA_TYPE_BINARY);
continue;
}
memcpy(varDataVal(output), varDataVal(pLeft->data), varDataLen(pLeft->data));
memcpy(varDataVal(output) + varDataLen(pLeft->data), varDataVal(right), varDataLen(right));
varDataSetLen(output, varDataLen(pLeft->data) + varDataLen(right));
}
} else if (pRight->num == 1) {
for (; i >= 0 && i < pLeft->num; i += step, output += len) {
char* left = POINTER_SHIFT(pLeft->data, pLeft->bytes * i);
if (isNull(left, pLeft->type) || isNull(pRight->data, pRight->type)) {
SET_DOUBLE_NULL(output);
continue;
}
memcpy(varDataVal(output), varDataVal(left), varDataLen(pRight->data));
memcpy(varDataVal(output) + varDataLen(left), varDataVal(pRight->data), varDataLen(pRight->data));
varDataSetLen(output, varDataLen(left) + varDataLen(pRight->data));
}
}
}
_bin_scalar_fn_t getBinScalarOperatorFn(int32_t binFunctionId) {
switch (binFunctionId) {
case TSDB_BINARY_OP_ADD:
return vectorAdd;
case TSDB_BINARY_OP_SUBTRACT:
return vectorSub;
case TSDB_BINARY_OP_MULTIPLY:
return vectorMultiply;
case TSDB_BINARY_OP_DIVIDE:
return vectorDivide;
case TSDB_BINARY_OP_REMAINDER:
return vectorRemainder;
case TSDB_BINARY_OP_CONCAT:
return vectorConcat;
default:
assert(0);
return NULL;
}
}
bool isBinaryStringOp(int32_t op) {
return op == TSDB_BINARY_OP_CONCAT;
}

3
source/libs/function/src/tfunction.c
View File

@ -3,7 +3,6 @@
#include "function.h"
#include "thash.h"
#include "taggfunction.h"
#include "tscalarfunction.h"
static SHashObj* functionHashTable = NULL;
static SHashObj* udfHashTable = NULL;
@ -18,12 +17,14 @@ static void doInitFunctionHashTable() {
taosHashPut(functionHashTable, aggFunc[i].name, len, (void*)&ptr, POINTER_BYTES);
}
/*
numOfEntries = tListLen(scalarFunc);
for(int32_t i = 0; i < numOfEntries; ++i) {
int32_t len = (int32_t) strlen(scalarFunc[i].name);
SScalarFunctionInfo* ptr = &scalarFunc[i];
taosHashPut(functionHashTable, scalarFunc[i].name, len, (void*)&ptr, POINTER_BYTES);
}
*/
udfHashTable = taosHashInit(numOfEntries, MurmurHash3_32, true, true);
}

12
source/libs/function/src/tunaryoperator.c
View File

@ -4,10 +4,10 @@
// TODO dynamic define these functions
_unary_scalar_fn_t getUnaryScalarOperatorFn(int32_t operator) {
assert(0);
}
//_unary_scalar_fn_t getUnaryScalarOperatorFn(int32_t operator) {
// assert(0);
//}
bool isStringOperatorFn(int32_t op) {
return op == FUNCTION_LENGTH;
}
//bool isStringOperatorFn(int32_t op) {
// return op == FUNCTION_LENGTH;
//}

31
source/libs/nodes/src/nodesUtilFuncs.c
View File

@ -17,6 +17,7 @@
#include "plannodes.h"
#include "taos.h"
#include "taoserror.h"
#include "taos.h"
#include "thash.h"
static SNode* makeNode(ENodeType type, size_t size) {
@ -172,6 +173,36 @@ void nodesDestroyList(SNodeList* pList) {
tfree(pList);
}
void *nodesGetValueFromNode(SValueNode *pNode) {
switch (pNode->node.resType.type) {
case TSDB_DATA_TYPE_BOOL:
return (void *)&pNode->datum.b;
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP:
return (void *)&pNode->datum.i;
case TSDB_DATA_TYPE_UTINYINT:
case TSDB_DATA_TYPE_USMALLINT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_UBIGINT:
return (void *)&pNode->datum.u;
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
return (void *)&pNode->datum.d;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY:
return (void *)pNode->datum.p;
default:
break;
}
return NULL;
}
bool nodesIsExprNode(const SNode* pNode) {
ENodeType type = nodeType(pNode);
return (QUERY_NODE_COLUMN == type || QUERY_NODE_VALUE == type || QUERY_NODE_OPERATOR == type || QUERY_NODE_FUNCTION == type);

16
source/libs/scalar/CMakeLists.txt Normal file
View File

@ -0,0 +1,16 @@
aux_source_directory(src SCALAR_SRC)
add_library(scalar STATIC ${SCALAR_SRC})
target_include_directories(
scalar
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/scalar"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(scalar
PRIVATE os util common nodes function qcom
)
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})

138
source/libs/executor/inc/tfilter.h → source/libs/scalar/inc/filterInt.h
View File

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_QFILTER_H
#define TDENGINE_QFILTER_H
#ifndef TDENGINE_FILTER_INT_H
#define TDENGINE_FILTER_INT_H
#ifdef __cplusplus
extern "C" {
@ -22,7 +22,10 @@ extern "C" {
#include "thash.h"
#include "tname.h"
#include "function.h"
#include "common.h"
#include "scalar.h"
#include "querynodes.h"
#include "query.h"
#define FILTER_DEFAULT_GROUP_SIZE 4
#define FILTER_DEFAULT_UNIT_SIZE 4
@ -83,6 +86,12 @@ enum {
RANGE_TYPE_MR_CTX = 3,
};
enum {
FI_ACTION_NO_NEED = 1,
FI_ACTION_CONTINUE,
FI_ACTION_STOP,
};
typedef struct OptrStr {
uint16_t optr;
char *str;
@ -95,17 +104,11 @@ typedef struct SFilterRange {
char eflag;
} SFilterRange;
typedef struct SFilterColRange {
uint16_t idx; //column field idx
bool isNull;
bool notNull;
bool isRange;
SFilterRange ra;
} SFilterColRange;
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 bool(*filter_exec_func)(void *, int32_t, int8_t**, SColumnDataAgg *, int16_t);
typedef int32_t (*filer_get_col_from_id)(void *, int32_t, void **);
typedef int32_t (*filer_get_col_from_name)(void *, int32_t, char*, void **);
typedef struct SFilterRangeCompare {
int64_t s;
@ -153,37 +156,39 @@ typedef struct SFilterField {
} SFilterField;
typedef struct SFilterFields {
uint16_t size;
uint16_t num;
uint32_t size;
uint32_t num;
SFilterField *fields;
} SFilterFields;
typedef struct SFilterFieldId {
uint16_t type;
uint16_t idx;
uint32_t idx;
} SFilterFieldId;
typedef struct SFilterGroup {
uint16_t unitSize;
uint16_t unitNum;
uint16_t *unitIdxs;
uint32_t unitSize;
uint32_t unitNum;
uint32_t *unitIdxs;
uint8_t *unitFlags; // !unit result
} SFilterGroup;
typedef struct SFilterColInfo {
uint8_t type;
int32_t dataType;
uint8_t optr; // for equal operation in the relation of RELATION_IN
int64_t value; // for equal operation in the relation of RELATION_IN
void *info;
} SFilterColInfo;
typedef struct SFilterGroupCtx {
uint16_t colNum;
uint16_t *colIdx;
uint32_t colNum;
uint32_t *colIdx;
SFilterColInfo *colInfo;
} SFilterGroupCtx;
typedef struct SFilterColCtx {
uint16_t colIdx;
uint32_t colIdx;
void* ctx;
} SFilterColCtx;
@ -217,16 +222,32 @@ typedef struct SFilterPCtx {
SHashObj *unitHash;
} SFilterPCtx;
typedef struct SFltTreeStat {
int32_t code;
bool scalarMode;
} SFltTreeStat;
typedef struct SFltScalarCtx {
SNode *node;
} SFltScalarCtx;
typedef struct SFltBuildGroupCtx {
SFilterInfo *info;
SArray *group;
int32_t code;
} SFltBuildGroupCtx;
typedef struct SFilterInfo {
bool scalarMode;
SFltScalarCtx sclCtx;
uint32_t options;
uint32_t status;
uint16_t unitSize;
uint16_t unitNum;
uint16_t groupNum;
uint16_t colRangeNum;
uint32_t unitSize;
uint32_t unitNum;
uint32_t groupNum;
uint32_t colRangeNum;
SFilterFields fields[FLD_TYPE_MAX];
SFilterGroup *groups;
uint16_t *cgroups;
SFilterUnit *units;
SFilterComUnit *cunits;
uint8_t *unitRes; // result
@ -234,16 +255,15 @@ typedef struct SFilterInfo {
SFilterRangeCtx **colRange;
filter_exec_func func;
uint8_t blkFlag;
uint16_t blkGroupNum;
uint16_t *blkUnits;
uint32_t blkGroupNum;
uint32_t *blkUnits;
int8_t *blkUnitRes;
void *pTable;
SFilterPCtx pctx;
} SFilterInfo;
#define COL_FIELD_SIZE (sizeof(SFilterField) + 2 * sizeof(int64_t))
#define FILTER_NO_MERGE_DATA_TYPE(t) ((t) == TSDB_DATA_TYPE_BINARY || (t) == TSDB_DATA_TYPE_NCHAR)
#define FILTER_NO_MERGE_DATA_TYPE(t) ((t) == TSDB_DATA_TYPE_BINARY || (t) == TSDB_DATA_TYPE_NCHAR || (t) == TSDB_DATA_TYPE_JSON)
#define FILTER_NO_MERGE_OPTR(o) ((o) == TSDB_RELATION_ISNULL || (o) == TSDB_RELATION_NOTNULL || (o) == FILTER_DUMMY_EMPTY_OPTR)
#define MR_EMPTY_RES(ctx) (ctx->rs == NULL)
@ -259,7 +279,7 @@ typedef struct SFilterInfo {
#define FILTER_CLR_FLAG(st, f) st &= (~f)
#define SIMPLE_COPY_VALUES(dst, src) *((int64_t *)dst) = *((int64_t *)src)
#define FILTER_PACKAGE_UNIT_HASH_KEY(v, optr, idx1, idx2) do { char *_t = (char *)v; _t[0] = optr; *(uint16_t *)(_t + 1) = idx1; *(uint16_t *)(_t + 3) = idx2; } while (0)
#define FILTER_PACKAGE_UNIT_HASH_KEY(v, optr, idx1, idx2) do { char *_t = (char *)v; _t[0] = optr; *(uint32_t *)(_t + 1) = idx1; *(uint32_t *)(_t + 3) = idx2; } while (0)
#define FILTER_GREATER(cr,sflag,eflag) ((cr > 0) || ((cr == 0) && (FILTER_GET_FLAG(sflag,RANGE_FLG_EXCLUDE) || FILTER_GET_FLAG(eflag,RANGE_FLG_EXCLUDE))))
#define FILTER_COPY_RA(dst, src) do { (dst)->sflag = (src)->sflag; (dst)->eflag = (src)->eflag; (dst)->s = (src)->s; (dst)->e = (src)->e; } while (0)
@ -269,25 +289,33 @@ typedef struct SFilterInfo {
#define INSERT_RANGE(ctx, r, ra) do { SFilterRangeNode *n = filterNewRange(ctx, ra); n->prev = (r)->prev; if ((r)->prev) { (r)->prev->next = n; } else { (ctx)->rs = n; } (r)->prev = n; n->next = r; } while (0)
#define APPEND_RANGE(ctx, r, ra) do { SFilterRangeNode *n = filterNewRange(ctx, ra); n->prev = (r); if (r) { (r)->next = n; } else { (ctx)->rs = n; } } while (0)
#define ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { return _code; } } while (0)
#define ERR_LRET(c,...) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { qError(__VA_ARGS__); return _code; } } while (0)
#define ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { goto _return; } } while (0)
#define FLT_IS_COMPARISON_OPERATOR(_op) ((_op) >= OP_TYPE_GREATER_THAN && (_op) < OP_TYPE_IS_NOT_NULL)
#define fltFatal(...) qFatal(__VA_ARGS__)
#define fltError(...) qError(__VA_ARGS__)
#define fltWarn(...) qWarn(__VA_ARGS__)
#define fltInfo(...) qInfo(__VA_ARGS__)
#define fltDebug(...) qDebug(__VA_ARGS__)
#define fltTrace(...) qTrace(__VA_ARGS__)
#define FLT_CHK_JMP(c) do { if (c) { goto _return; } } while (0)
#define FLT_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define FLT_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define FLT_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
#define CHK_RETV(c) do { if (c) { return; } } while (0)
#define CHK_RET(c, r) do { if (c) { return r; } } while (0)
#define CHK_JMP(c) do { if (c) { goto _return; } } while (0)
#define CHK_LRETV(c,...) do { if (c) { qError(__VA_ARGS__); return; } } while (0)
#define CHK_LRET(c, r,...) do { if (c) { if (r) {qError(__VA_ARGS__); } else { qDebug(__VA_ARGS__); } return r; } } while (0)
#define FILTER_GET_FIELD(i, id) (&((i)->fields[(id).type].fields[(id).idx]))
#define FILTER_GET_COL_FIELD(i, idx) (&((i)->fields[FLD_TYPE_COLUMN].fields[idx]))
#define FILTER_GET_COL_FIELD_TYPE(fi) (((SSchema *)((fi)->desc))->type)
#define FILTER_GET_COL_FIELD_SIZE(fi) (((SSchema *)((fi)->desc))->bytes)
#define FILTER_GET_COL_FIELD_ID(fi) (((SSchema *)((fi)->desc))->colId)
#define FILTER_GET_COL_FIELD_DESC(fi) ((SSchema *)((fi)->desc))
#define FILTER_GET_COL_FIELD_DATA(fi, ri) ((char *)(fi)->data + ((SSchema *)((fi)->desc))->bytes * (ri))
#define FILTER_GET_VAL_FIELD_TYPE(fi) (((tVariant *)((fi)->desc))->nType)
#define FILTER_GET_COL_FIELD_TYPE(fi) (((SColumnRefNode *)((fi)->desc))->dataType.type)
#define FILTER_GET_COL_FIELD_SIZE(fi) (((SColumnRefNode *)((fi)->desc))->dataType.bytes)
#define FILTER_GET_COL_FIELD_ID(fi) (((SColumnRefNode *)((fi)->desc))->columnId)
#define FILTER_GET_COL_FIELD_SLOT_ID(fi) (((SColumnRefNode *)((fi)->desc))->slotId)
#define FILTER_GET_COL_FIELD_DESC(fi) ((SColumnRefNode *)((fi)->desc))
#define FILTER_GET_COL_FIELD_DATA(fi, ri) ((char *)(fi)->data + ((SColumnRefNode *)((fi)->desc))->dataType.bytes * (ri))
#define FILTER_GET_VAL_FIELD_TYPE(fi) (((SValueNode *)((fi)->desc))->node.resType.type)
#define FILTER_GET_VAL_FIELD_DATA(fi) ((char *)(fi)->data)
#define FILTER_GET_JSON_VAL_FIELD_DATA(fi) ((char *)(fi)->desc)
#define FILTER_GET_TYPE(fl) ((fl) & FLD_TYPE_MAX)
#define FILTER_GROUP_UNIT(i, g, uid) ((i)->units + (g)->unitIdxs[uid])
@ -300,6 +328,7 @@ typedef struct SFilterInfo {
#define FILTER_UNIT_COL_SIZE(i, u) FILTER_GET_COL_FIELD_SIZE(FILTER_UNIT_LEFT_FIELD(i, u))
#define FILTER_UNIT_COL_ID(i, u) FILTER_GET_COL_FIELD_ID(FILTER_UNIT_LEFT_FIELD(i, u))
#define FILTER_UNIT_VAL_DATA(i, u) FILTER_GET_VAL_FIELD_DATA(FILTER_UNIT_RIGHT_FIELD(i, u))
#define FILTER_UNIT_JSON_VAL_DATA(i, u) FILTER_GET_JSON_VAL_FIELD_DATA(FILTER_UNIT_RIGHT_FIELD(i, u))
#define FILTER_UNIT_COL_IDX(u) ((u)->left.idx)
#define FILTER_UNIT_OPTR(u) ((u)->compare.optr)
#define FILTER_UNIT_COMP_FUNC(u) ((u)->compare.func)
@ -314,7 +343,7 @@ typedef struct SFilterInfo {
#define FILTER_PUSH_VAR_HASH(colInfo, ha) do { (colInfo).type = RANGE_TYPE_VAR_HASH; (colInfo).info = ha;} while (0)
#define FILTER_PUSH_CTX(colInfo, ctx) do { (colInfo).type = RANGE_TYPE_MR_CTX; (colInfo).info = ctx;} while (0)
#define FILTER_COPY_IDX(dst, src, n) do { *(dst) = malloc(sizeof(uint16_t) * n); memcpy(*(dst), src, sizeof(uint16_t) * n);} while (0)
#define FILTER_COPY_IDX(dst, src, n) do { *(dst) = malloc(sizeof(uint32_t) * n); memcpy(*(dst), src, sizeof(uint32_t) * n);} while (0)
#define FILTER_ADD_CTX_TO_GRES(gres, idx, ctx) do { if ((gres)->colCtxs == NULL) { (gres)->colCtxs = taosArrayInit(gres->colNum, sizeof(SFilterColCtx)); } SFilterColCtx cCtx = {idx, ctx}; taosArrayPush((gres)->colCtxs, &cCtx); } while (0)
@ -322,18 +351,25 @@ typedef struct SFilterInfo {
#define FILTER_ALL_RES(i) FILTER_GET_FLAG((i)->status, FI_STATUS_ALL)
#define FILTER_EMPTY_RES(i) FILTER_GET_FLAG((i)->status, FI_STATUS_EMPTY)
extern int32_t filterInitFromTree(tExprNode* tree, SFilterInfo **pinfo, uint32_t options);
#if 0
extern int32_t filterInitFromTree(tExprNode* tree, void **pinfo, uint32_t options);
extern bool filterExecute(SFilterInfo *info, int32_t numOfRows, int8_t** p, SColumnDataAgg *statis, int16_t numOfCols);
extern int32_t filterSetColFieldData(SFilterInfo *info, int32_t numOfCols, SArray* pDataBlock);
extern int32_t filterSetColFieldData(SFilterInfo *info, void *param, filer_get_col_from_id fp);
extern int32_t filterSetJsonColFieldData(SFilterInfo *info, void *param, filer_get_col_from_name fp);
extern int32_t filterGetTimeRange(SFilterInfo *info, STimeWindow *win);
extern int32_t filterConverNcharColumns(SFilterInfo* pFilterInfo, int32_t rows, bool *gotNchar);
extern int32_t filterFreeNcharColumns(SFilterInfo* pFilterInfo);
extern void filterFreeInfo(SFilterInfo *info);
extern bool filterRangeExecute(SFilterInfo *info, SColumnDataAgg *pDataStatis, int32_t numOfCols, int32_t numOfRows);
#else
//REMOVE THESE!!!!!!!!!!!!!!!!!!!!
#include "function.h"
#endif
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);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_QFILTER_H
#endif // TDENGINE_FILTER_INT_H

51
source/libs/scalar/inc/sclInt.h Normal file
View File

@ -0,0 +1,51 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_SCALARINT_H
#define TDENGINE_SCALARINT_H
#ifdef __cplusplus
extern "C" {
#endif
#include "common.h"
#include "thash.h"
#include "query.h"
typedef struct SScalarCtx {
int32_t code;
SSDataBlock *pSrc;
SHashObj *pRes; /* element is SScalarParam */
} SScalarCtx;
#define SCL_DEFAULT_OP_NUM 10
#define sclFatal(...) qFatal(__VA_ARGS__)
#define sclError(...) qError(__VA_ARGS__)
#define sclWarn(...) qWarn(__VA_ARGS__)
#define sclInfo(...) qInfo(__VA_ARGS__)
#define sclDebug(...) qDebug(__VA_ARGS__)
#define sclTrace(...) qTrace(__VA_ARGS__)
#define SCL_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define SCL_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define SCL_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_SCALARINT_H

10
source/libs/function/inc/tscalarfunction.h → source/libs/scalar/inc/sclfunc.h
View File

@ -20,13 +20,7 @@ extern "C" {
#endif
#include "function.h"
typedef struct SScalarFuncParam {
void* data;
int32_t num;
int32_t type;
int32_t bytes;
} SScalarFuncParam;
#include "scalar.h"
typedef struct SScalarFunctionSupport {
struct SExprInfo *pExprInfo;
@ -39,7 +33,7 @@ typedef struct SScalarFunctionSupport {
extern struct SScalarFunctionInfo scalarFunc[8];
int32_t evaluateExprNodeTree(tExprNode* pExprs, int32_t numOfRows, SScalarFuncParam* pOutput,
int32_t evaluateExprNodeTree(tExprNode* pExprs, int32_t numOfRows, SScalarParam* pOutput,
void* param, char* (*getSourceDataBlock)(void*, const char*, int32_t));

4
source/libs/function/inc/tbinoperator.h → source/libs/scalar/inc/sclvector.h
View File

@ -20,9 +20,9 @@
extern "C" {
#endif
#include "tscalarfunction.h"
#include "sclfunc.h"
typedef void (*_bin_scalar_fn_t)(SScalarFuncParam* pLeft, SScalarFuncParam* pRight, void *output, int32_t order);
typedef void (*_bin_scalar_fn_t)(SScalarParam* pLeft, SScalarParam* pRight, void *output, int32_t order);
_bin_scalar_fn_t getBinScalarOperatorFn(int32_t binOperator);
bool isBinaryStringOp(int32_t op);

1379
source/libs/executor/src/tfilter.c → source/libs/scalar/src/filter.c
View File

File diff suppressed because it is too large Load Diff

594
source/libs/scalar/src/scalar.c Normal file
View File

@ -0,0 +1,594 @@
#include "nodes.h"
#include "common.h"
#include "querynodes.h"
#include "function.h"
#include "functionMgt.h"
#include "sclvector.h"
#include "sclInt.h"
int32_t scalarGetOperatorParamNum(EOperatorType type) {
if (OP_TYPE_IS_NULL == type || OP_TYPE_IS_NOT_NULL == type) {
return 1;
}
return 2;
}
void sclFreeRes(SHashObj *res) {
SScalarParam *p = NULL;
void *pIter = taosHashIterate(res, NULL);
while (pIter) {
p = (SScalarParam *)pIter;
if (p) {
tfree(p->data);
}
pIter = taosHashIterate(res, pIter);
}
taosHashCleanup(res);
}
void sclFreeParam(SScalarParam *param) {
tfree(param->data);
}
int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t *rowNum) {
switch (nodeType(node)) {
case QUERY_NODE_VALUE: {
SValueNode *valueNode = (SValueNode *)node;
param->data = nodesGetValueFromNode(valueNode);
param->num = 1;
param->type = valueNode->node.resType.type;
param->bytes = valueNode->node.resType.bytes;
break;
}
case QUERY_NODE_NODE_LIST: {
SNodeListNode *nodeList = (SNodeListNode *)node;
//TODO BUILD HASH
break;
}
case QUERY_NODE_COLUMN_REF: {
if (NULL == ctx) {
sclError("invalid node type for constant calculating, type:%d, ctx:%p", nodeType(node), ctx);
SCL_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
}
SColumnRefNode *ref = (SColumnRefNode *)node;
if (ref->slotId >= taosArrayGetSize(ctx->pSrc->pDataBlock)) {
sclError("column ref slotId is too big, slodId:%d, dataBlockSize:%d", ref->slotId, (int32_t)taosArrayGetSize(ctx->pSrc->pDataBlock));
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
SColumnInfoData *columnData = (SColumnInfoData *)taosArrayGet(ctx->pSrc->pDataBlock, ref->slotId);
param->data = columnData->pData;
param->num = ctx->pSrc->info.rows;
param->type = columnData->info.type;
param->bytes = columnData->info.bytes;
break;
}
case QUERY_NODE_LOGIC_CONDITION:
case QUERY_NODE_OPERATOR: {
if (NULL == ctx) {
sclError("invalid node type for constant calculating, type:%d, ctx:%p", nodeType(node), ctx);
SCL_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
}
SScalarParam *res = (SScalarParam *)taosHashGet(ctx->pRes, &node, POINTER_BYTES);
if (NULL == res) {
sclError("no result for node, type:%d, node:%p", nodeType(node), node);
SCL_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
}
*param = *res;
break;
}
}
if (param->num > *rowNum) {
if ((1 != param->num) && (1 < *rowNum)) {
sclError("different row nums, rowNum:%d, newRowNum:%d", *rowNum, param->num);
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
*rowNum = param->num;
}
return TSDB_CODE_SUCCESS;
}
int32_t sclParamMoveNext(SScalarParam *params, int32_t num) {
SScalarParam *param = NULL;
for (int32_t i = 0; i < num; ++i) {
param = params + i;
if (1 == param->num) {
continue;
}
if (IS_VAR_DATA_TYPE(param->type)) {
param->data = (char *)(param->data) + varDataTLen(param->data);
} else {
param->data = (char *)(param->data) + tDataTypes[param->type].bytes;
}
}
return TSDB_CODE_SUCCESS;
}
int32_t sclInitParamList(SScalarParam **pParams, SNodeList* pParamList, SScalarCtx *ctx, int32_t *rowNum) {
int32_t code = 0;
SScalarParam *paramList = calloc(pParamList->length, sizeof(SScalarParam));
if (NULL == paramList) {
sclError("calloc %d failed", (int32_t)(pParamList->length * sizeof(SScalarParam)));
SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SListCell *cell = pParamList->pHead;
for (int32_t i = 0; i < pParamList->length; ++i) {
if (NULL == cell || NULL == cell->pNode) {
sclError("invalid cell, cell:%p, pNode:%p", cell, cell->pNode);
SCL_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
}
SCL_ERR_JRET(sclInitParam(cell->pNode, &paramList[i], ctx, rowNum));
cell = cell->pNext;
}
*pParams = paramList;
return TSDB_CODE_SUCCESS;
_return:
tfree(paramList);
SCL_RET(code);
}
int32_t sclInitOperatorParams(SScalarParam **pParams, SOperatorNode *node, SScalarCtx *ctx, int32_t *rowNum) {
int32_t code = 0;
int32_t paramNum = scalarGetOperatorParamNum(node->opType);
if (NULL == node->pLeft || (paramNum == 2 && NULL == node->pRight)) {
sclError("invalid operation node, left:%p, right:%p", node->pLeft, node->pRight);
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
SScalarParam *paramList = calloc(paramNum, sizeof(SScalarParam));
if (NULL == paramList) {
sclError("calloc %d failed", (int32_t)(paramNum * sizeof(SScalarParam)));
SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SCL_ERR_JRET(sclInitParam(node->pLeft, &paramList[0], ctx, rowNum));
if (paramNum > 1) {
SCL_ERR_JRET(sclInitParam(node->pRight, &paramList[1], ctx, rowNum));
}
*pParams = paramList;
return TSDB_CODE_SUCCESS;
_return:
tfree(paramList);
SCL_RET(code);
}
int32_t sclExecFuncion(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *output) {
if (NULL == node->pParameterList || node->pParameterList->length <= 0) {
sclError("invalid function parameter list, list:%p, paramNum:%d", node->pParameterList, node->pParameterList ? node->pParameterList->length : 0);
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
SScalarFuncExecFuncs ffpSet = {0};
int32_t code = fmGetScalarFuncExecFuncs(node->funcId, &ffpSet);
if (code) {
sclError( "fmGetFuncExecFuncs failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
SCL_ERR_RET(code);
}
SScalarParam *params = NULL;
int32_t rowNum = 0;
SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &rowNum));
output->type = node->node.resType.type;
output->data = calloc(rowNum, sizeof(tDataTypes[output->type].bytes));
if (NULL == output->data) {
sclError("calloc %d failed", (int32_t)(rowNum * sizeof(tDataTypes[output->type].bytes)));
SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
for (int32_t i = 0; i < rowNum; ++i) {
code = (*ffpSet.process)(params, node->pParameterList->length, output);
if (code) {
sclError( "scalar function exec failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
SCL_ERR_JRET(code);
}
sclParamMoveNext(output, 1);
sclParamMoveNext(params, node->pParameterList->length);
}
return TSDB_CODE_SUCCESS;
_return:
tfree(params);
SCL_RET(code);
}
int32_t sclExecLogic(SLogicConditionNode *node, SScalarCtx *ctx, SScalarParam *output) {
if (NULL == node->pParameterList || node->pParameterList->length <= 0) {
sclError("invalid logic parameter list, list:%p, paramNum:%d", node->pParameterList, node->pParameterList ? node->pParameterList->length : 0);
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
if (TSDB_DATA_TYPE_BOOL != node->node.resType.type) {
sclError("invalid logic resType, type:%d", node->node.resType.type);
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
if (LOGIC_COND_TYPE_NOT == node->condType && node->pParameterList->length > 1) {
sclError("invalid NOT operation parameter number, paramNum:%d", node->pParameterList->length);
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
SScalarParam *params = NULL;
int32_t rowNum = 0;
int32_t code = 0;
SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &rowNum));
output->type = node->node.resType.type;
output->data = calloc(rowNum, sizeof(bool));
if (NULL == output->data) {
sclError("calloc %d failed", (int32_t)(rowNum * sizeof(bool)));
SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
bool value = false;
for (int32_t i = 0; i < rowNum; ++i) {
for (int32_t m = 0; m < node->pParameterList->length; ++m) {
GET_TYPED_DATA(value, bool, params[m].type, params[m].data);
if (LOGIC_COND_TYPE_AND == node->condType && (false == value)) {
break;
} else if (LOGIC_COND_TYPE_OR == node->condType && value) {
break;
} else if (LOGIC_COND_TYPE_NOT == node->condType) {
value = !value;
}
}
*(bool *)output->data = value;
sclParamMoveNext(output, 1);
sclParamMoveNext(params, node->pParameterList->length);
}
return TSDB_CODE_SUCCESS;
_return:
tfree(params);
SCL_RET(code);
}
int32_t sclExecOperator(SOperatorNode *node, SScalarCtx *ctx, SScalarParam *output) {
SScalarParam *params = NULL;
int32_t rowNum = 0;
int32_t code = 0;
SCL_ERR_RET(sclInitOperatorParams(&params, node, ctx, &rowNum));
output->type = node->node.resType.type;
output->data = calloc(rowNum, tDataTypes[output->type].bytes);
if (NULL == output->data) {
sclError("calloc %d failed", (int32_t)rowNum * tDataTypes[output->type].bytes);
SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
_bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(node->opType);
int32_t paramNum = scalarGetOperatorParamNum(node->opType);
SScalarParam* pLeft = &params[0];
SScalarParam* pRight = paramNum > 1 ? &params[1] : NULL;
for (int32_t i = 0; i < rowNum; ++i) {
OperatorFn(pLeft, pRight, output->data, TSDB_ORDER_ASC);
sclParamMoveNext(output, 1);
sclParamMoveNext(pLeft, 1);
if (pRight) {
sclParamMoveNext(pRight, 1);
}
}
return TSDB_CODE_SUCCESS;
_return:
tfree(params);
SCL_RET(code);
}
EDealRes sclRewriteFunction(SNode** pNode, void* pContext) {
SFunctionNode *node = (SFunctionNode *)*pNode;
SScalarParam output = {0};
*(int32_t *)pContext = sclExecFuncion(node, NULL, &output);
if (*(int32_t *)pContext) {
return DEAL_RES_ERROR;
}
SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == res) {
sclError("make value node failed");
sclFreeParam(&output);
*(int32_t *)pContext = TSDB_CODE_QRY_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
res->node.resType = node->node.resType;
if (IS_VAR_DATA_TYPE(output.type)) {
res->datum.p = output.data;
output.data = NULL;
} else {
memcpy(nodesGetValueFromNode(res), output.data, tDataTypes[output.type].bytes);
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)res;
sclFreeParam(&output);
return DEAL_RES_CONTINUE;
}
EDealRes sclRewriteLogic(SNode** pNode, void* pContext) {
SLogicConditionNode *node = (SLogicConditionNode *)*pNode;
SScalarParam output = {0};
*(int32_t *)pContext = sclExecLogic(node, NULL, &output);
if (*(int32_t *)pContext) {
return DEAL_RES_ERROR;
}
SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == res) {
sclError("make value node failed");
sclFreeParam(&output);
*(int32_t *)pContext = TSDB_CODE_QRY_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
res->node.resType = node->node.resType;
if (IS_VAR_DATA_TYPE(output.type)) {
res->datum.p = output.data;
output.data = NULL;
} else {
memcpy(nodesGetValueFromNode(res), output.data, tDataTypes[output.type].bytes);
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)res;
sclFreeParam(&output);
return DEAL_RES_CONTINUE;
}
EDealRes sclRewriteOperator(SNode** pNode, void* pContext) {
SOperatorNode *node = (SOperatorNode *)*pNode;
SScalarParam output = {0};
*(int32_t *)pContext = sclExecOperator(node, NULL, &output);
if (*(int32_t *)pContext) {
return DEAL_RES_ERROR;
}
SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == res) {
sclError("make value node failed");
sclFreeParam(&output);
*(int32_t *)pContext = TSDB_CODE_QRY_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
res->node.resType = node->node.resType;
if (IS_VAR_DATA_TYPE(output.type)) {
res->datum.p = output.data;
output.data = NULL;
} else {
memcpy(nodesGetValueFromNode(res), output.data, tDataTypes[output.type].bytes);
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)res;
sclFreeParam(&output);
return DEAL_RES_CONTINUE;
}
EDealRes sclConstantsRewriter(SNode** pNode, void* pContext) {
if (QUERY_NODE_VALUE == nodeType(*pNode)) {
return DEAL_RES_CONTINUE;
}
if (QUERY_NODE_FUNCTION == nodeType(*pNode)) {
return sclRewriteFunction(pNode, pContext);
}
if (QUERY_NODE_LOGIC_CONDITION == nodeType(*pNode)) {
return sclRewriteLogic(pNode, pContext);
}
if (QUERY_NODE_OPERATOR == nodeType(*pNode)) {
return sclRewriteOperator(pNode, pContext);
}
sclError("invalid node type for calculating constants, type:%d", nodeType(*pNode));
*(int32_t *)pContext = TSDB_CODE_QRY_INVALID_INPUT;
return DEAL_RES_ERROR;
}
EDealRes sclWalkFunction(SNode* pNode, void* pContext) {
SScalarCtx *ctx = (SScalarCtx *)pContext;
SFunctionNode *node = (SFunctionNode *)pNode;
SScalarParam output = {0};
ctx->code = sclExecFuncion(node, ctx, &output);
if (ctx->code) {
return DEAL_RES_ERROR;
}
if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
return DEAL_RES_CONTINUE;
}
EDealRes sclWalkLogic(SNode* pNode, void* pContext) {
SScalarCtx *ctx = (SScalarCtx *)pContext;
SLogicConditionNode *node = (SLogicConditionNode *)pNode;
SScalarParam output = {0};
ctx->code = sclExecLogic(node, ctx, &output);
if (ctx->code) {
return DEAL_RES_ERROR;
}
if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
return DEAL_RES_CONTINUE;
}
EDealRes sclWalkOperator(SNode* pNode, void* pContext) {
SScalarCtx *ctx = (SScalarCtx *)pContext;
SOperatorNode *node = (SOperatorNode *)pNode;
SScalarParam output = {0};
ctx->code = sclExecOperator(node, ctx, &output);
if (ctx->code) {
return DEAL_RES_ERROR;
}
if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
return DEAL_RES_ERROR;
}
return DEAL_RES_CONTINUE;
}
EDealRes sclCalcWalker(SNode* pNode, void* pContext) {
if (QUERY_NODE_VALUE == nodeType(pNode)) {
return DEAL_RES_CONTINUE;
}
if (QUERY_NODE_FUNCTION == nodeType(pNode)) {
return sclWalkFunction(pNode, pContext);
}
if (QUERY_NODE_LOGIC_CONDITION == nodeType(pNode)) {
return sclWalkLogic(pNode, pContext);
}
if (QUERY_NODE_OPERATOR == nodeType(pNode)) {
return sclWalkOperator(pNode, pContext);
}
sclError("invalid node type for calculating constants, type:%d", nodeType(pNode));
SScalarCtx *ctx = (SScalarCtx *)pContext;
ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
return DEAL_RES_ERROR;
}
int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes) {
if (NULL == pNode) {
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
int32_t code = 0;
nodesRewriteNodePostOrder(&pNode, sclConstantsRewriter, (void *)&code);
if (code) {
nodesDestroyNode(pNode);
SCL_ERR_RET(code);
}
*pRes = pNode;
SCL_RET(code);
}
int32_t scalarCalculate(SNode *pNode, SSDataBlock *pSrc, SScalarParam *pDst) {
if (NULL == pNode || NULL == pSrc || NULL == pDst) {
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
int32_t code = 0;
SScalarCtx ctx = {.code = 0, .pSrc = pSrc};
ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if (NULL == ctx.pRes) {
sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
nodesWalkNodePostOrder(pNode, sclCalcWalker, (void *)&ctx);
if (ctx.code) {
nodesDestroyNode(pNode);
sclFreeRes(ctx.pRes);
SCL_ERR_RET(ctx.code);
}
SScalarParam *res = (SScalarParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES);
if (NULL == res) {
sclError("no res for calculating, node:%p, type:%d", pNode, nodeType(pNode));
SCL_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
}
*pDst = *res;
nodesDestroyNode(pNode);
return TSDB_CODE_SUCCESS;
}

28
source/libs/function/src/tscalarfunction.c → source/libs/scalar/src/sclfunc.c
View File

@ -1,14 +1,13 @@
#include "tscalarfunction.h"
#include "tbinoperator.h"
#include "tunaryoperator.h"
#include "sclfunc.h"
#include "sclvector.h"
static void assignBasicParaInfo(struct SScalarFuncParam* dst, const struct SScalarFuncParam* src) {
static void assignBasicParaInfo(struct SScalarParam* dst, const struct SScalarParam* src) {
dst->type = src->type;
dst->bytes = src->bytes;
dst->num = src->num;
}
static void tceil(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFuncParam *pLeft) {
static void tceil(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {
assignBasicParaInfo(pOutput, pLeft);
assert(numOfInput == 1);
@ -34,7 +33,7 @@ static void tceil(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFun
}
}
static void tfloor(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFuncParam *pLeft) {
static void tfloor(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {
assignBasicParaInfo(pOutput, pLeft);
assert(numOfInput == 1);
@ -62,7 +61,7 @@ static void tfloor(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFu
}
}
static void _tabs(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFuncParam *pLeft) {
static void _tabs(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {
assignBasicParaInfo(pOutput, pLeft);
assert(numOfInput == 1);
@ -120,7 +119,7 @@ static void _tabs(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFun
}
}
static void tround(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFuncParam *pLeft) {
static void tround(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {
assignBasicParaInfo(pOutput, pLeft);
assert(numOfInput == 1);
@ -146,7 +145,7 @@ static void tround(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFu
}
}
static void tlength(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFuncParam *pLeft) {
static void tlength(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {
assert(numOfInput == 1);
int64_t* out = (int64_t*) pOutput->data;
@ -157,7 +156,7 @@ static void tlength(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarF
}
}
static void tconcat(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFuncParam *pLeft) {
static void tconcat(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {
assert(numOfInput > 0);
int32_t rowLen = 0;
@ -189,11 +188,11 @@ static void tconcat(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarF
}
}
static void tltrim(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFuncParam *pLeft) {
static void tltrim(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {
}
static void trtrim(SScalarFuncParam* pOutput, size_t numOfInput, const SScalarFuncParam *pLeft) {
static void trtrim(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {
}
@ -262,7 +261,7 @@ static void reverseCopy(char* dest, const char* src, int16_t type, int32_t numOf
}
}
static void setScalarFuncParam(SScalarFuncParam* param, int32_t type, int32_t bytes, void* pInput, int32_t numOfRows) {
static void setScalarFuncParam(SScalarParam* param, int32_t type, int32_t bytes, void* pInput, int32_t numOfRows) {
param->bytes = bytes;
param->type = type;
param->num = numOfRows;
@ -273,6 +272,7 @@ bool isStringOp(int32_t op) {
return op == TSDB_BINARY_OP_CONCAT;
}
#if 0
int32_t evaluateExprNodeTree(tExprNode* pExprs, int32_t numOfRows, SScalarFuncParam* pOutput, void* param,
char* (*getSourceDataBlock)(void*, const char*, int32_t)) {
if (pExprs == NULL) {
@ -361,6 +361,8 @@ int32_t evaluateExprNodeTree(tExprNode* pExprs, int32_t numOfRows, SScalarFuncPa
return 0;
}
#endif
SScalarFunctionInfo scalarFunc[8] = {
{"ceil", FUNCTION_TYPE_SCALAR, FUNCTION_CEIL, tceil},

1177
source/libs/scalar/src/sclvector.c Normal file
View File

File diff suppressed because it is too large Load Diff

18
source/libs/scalar/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,18 @@
MESSAGE(STATUS "build scalar unit test")
# GoogleTest requires at least C++11
SET(CMAKE_CXX_STANDARD 11)
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
ADD_EXECUTABLE(scalarTest ${SOURCE_LIST})
TARGET_LINK_LIBRARIES(
scalarTest
PUBLIC os util common gtest qcom function nodes
)
TARGET_INCLUDE_DIRECTORIES(
scalarTest
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/scalar/"
PRIVATE "${CMAKE_SOURCE_DIR}/source/libs/scalar/inc"
)

55
source/libs/scalar/test/scalarTests.cpp Normal file
View File

@ -0,0 +1,55 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
#include <tglobal.h>
#include <iostream>
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wwrite-strings"
#pragma GCC diagnostic ignored "-Wunused-function"
#pragma GCC diagnostic ignored "-Wunused-variable"
#pragma GCC diagnostic ignored "-Wsign-compare"
#pragma GCC diagnostic ignored "-Wsign-compare"
#pragma GCC diagnostic ignored "-Wformat"
#pragma GCC diagnostic ignored "-Wint-to-pointer-cast"
#pragma GCC diagnostic ignored "-Wpointer-arith"
#include "os.h"
#include "taos.h"
#include "tdef.h"
#include "tvariant.h"
#include "tep.h"
#include "stub.h"
#include "addr_any.h"
#include "scalar.h"
namespace {
}
TEST(scalarTest, func) {
}
int main(int argc, char** argv) {
srand(time(NULL));
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
#pragma GCC diagnostic pop

190
source/util/src/compare.c
View File

@ -23,23 +23,51 @@
#include "thash.h"
#include "types.h"
#include "ulog.h"
#include "tdef.h"
#include "taos.h"
int32_t setCompareBytes1(const void *pLeft, const void *pRight) {
int32_t setChkInBytes1(const void *pLeft, const void *pRight) {
return NULL != taosHashGet((SHashObj *)pRight, pLeft, 1) ? 1 : 0;
}
int32_t setCompareBytes2(const void *pLeft, const void *pRight) {
int32_t setChkInBytes2(const void *pLeft, const void *pRight) {
return NULL != taosHashGet((SHashObj *)pRight, pLeft, 2) ? 1 : 0;
}
int32_t setCompareBytes4(const void *pLeft, const void *pRight) {
int32_t setChkInBytes4(const void *pLeft, const void *pRight) {
return NULL != taosHashGet((SHashObj *)pRight, pLeft, 4) ? 1 : 0;
}
int32_t setCompareBytes8(const void *pLeft, const void *pRight) {
int32_t setChkInBytes8(const void *pLeft, const void *pRight) {
return NULL != taosHashGet((SHashObj *)pRight, pLeft, 8) ? 1 : 0;
}
int32_t setChkNotInBytes1(const void *pLeft, const void *pRight) {
return NULL == taosHashGet((SHashObj *)pRight, pLeft, 1) ? 1 : 0;
}
int32_t setChkNotInBytes2(const void *pLeft, const void *pRight) {
return NULL == taosHashGet((SHashObj *)pRight, pLeft, 2) ? 1 : 0;
}
int32_t setChkNotInBytes4(const void *pLeft, const void *pRight) {
return NULL == taosHashGet((SHashObj *)pRight, pLeft, 4) ? 1 : 0;
}
int32_t setChkNotInBytes8(const void *pLeft, const void *pRight) {
return NULL == taosHashGet((SHashObj *)pRight, pLeft, 8) ? 1 : 0;
}
int32_t compareChkInString(const void *pLeft, const void* pRight) {
return NULL != taosHashGet((SHashObj *)pRight, varDataVal(pLeft), varDataLen(pLeft)) ? 1 : 0;
}
int32_t compareChkNotInString(const void *pLeft, const void* pRight) {
return NULL == taosHashGet((SHashObj *)pRight, varDataVal(pLeft), varDataLen(pLeft)) ? 1 : 0;
}
int32_t compareInt8Val(const void *pLeft, const void *pRight) {
int8_t left = GET_INT8_VAL(pLeft), right = GET_INT8_VAL(pRight);
if (left > right) return 1;
@ -393,6 +421,156 @@ int32_t taosArrayCompareString(const void* a, const void* b) {
return compareLenPrefixedStr(x, y);
}
int32_t compareFindItemInSet(const void *pLeft, const void* pRight) {
return NULL != taosHashGet((SHashObj *)pRight, varDataVal(pLeft), varDataLen(pLeft)) ? 1 : 0;
int32_t compareStrPatternMatch(const void* pLeft, const void* pRight) {
SPatternCompareInfo pInfo = {'%', '_'};
assert(varDataLen(pRight) <= TSDB_MAX_FIELD_LEN);
char *pattern = calloc(varDataLen(pRight) + 1, sizeof(char));
memcpy(pattern, varDataVal(pRight), varDataLen(pRight));
size_t sz = varDataLen(pLeft);
char *buf = malloc(sz + 1);
memcpy(buf, varDataVal(pLeft), sz);
buf[sz] = 0;
int32_t ret = patternMatch(pattern, buf, sz, &pInfo);
free(buf);
free(pattern);
return (ret == TSDB_PATTERN_MATCH) ? 0 : 1;
}
int32_t compareStrPatternNotMatch(const void* pLeft, const void* pRight) {
return compareStrPatternMatch(pLeft, pRight) ? 0 : 1;
}
int32_t compareWStrPatternMatch(const void* pLeft, const void* pRight) {
SPatternCompareInfo pInfo = {'%', '_'};
assert(varDataLen(pRight) <= TSDB_MAX_FIELD_LEN * TSDB_NCHAR_SIZE);
wchar_t *pattern = calloc(varDataLen(pRight) + 1, sizeof(wchar_t));
memcpy(pattern, varDataVal(pRight), varDataLen(pRight));
int32_t ret = WCSPatternMatch(pattern, varDataVal(pLeft), varDataLen(pLeft)/TSDB_NCHAR_SIZE, &pInfo);
free(pattern);
return (ret == TSDB_PATTERN_MATCH) ? 0 : 1;
}
int32_t compareWStrPatternNotMatch(const void* pLeft, const void* pRight) {
return compareWStrPatternMatch(pLeft, pRight) ? 0 : 1;
}
__compar_fn_t getComparFunc(int32_t type, int32_t optr) {
__compar_fn_t comparFn = NULL;
if (optr == TSDB_RELATION_IN && (type != TSDB_DATA_TYPE_BINARY && type != TSDB_DATA_TYPE_NCHAR)) {
switch (type) {
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_UTINYINT:
return setChkInBytes1;
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_USMALLINT:
return setChkInBytes2;
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_FLOAT:
return setChkInBytes4;
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_UBIGINT:
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_TIMESTAMP:
return setChkInBytes8;
default:
assert(0);
}
}
if (optr == TSDB_RELATION_NOT_IN && (type != TSDB_DATA_TYPE_BINARY && type != TSDB_DATA_TYPE_NCHAR)) {
switch (type) {
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_UTINYINT:
return setChkNotInBytes1;
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_USMALLINT:
return setChkNotInBytes2;
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_FLOAT:
return setChkNotInBytes4;
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_UBIGINT:
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_TIMESTAMP:
return setChkNotInBytes8;
default:
assert(0);
}
}
switch (type) {
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT: comparFn = compareInt8Val; break;
case TSDB_DATA_TYPE_SMALLINT: comparFn = compareInt16Val; break;
case TSDB_DATA_TYPE_INT: comparFn = compareInt32Val; break;
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP: comparFn = compareInt64Val; break;
case TSDB_DATA_TYPE_FLOAT: comparFn = compareFloatVal; break;
case TSDB_DATA_TYPE_DOUBLE: comparFn = compareDoubleVal; break;
case TSDB_DATA_TYPE_BINARY: {
if (optr == TSDB_RELATION_MATCH) {
comparFn = compareStrRegexCompMatch;
} else if (optr == TSDB_RELATION_NMATCH) {
comparFn = compareStrRegexCompNMatch;
} else if (optr == TSDB_RELATION_LIKE) { /* wildcard query using like operator */
comparFn = compareStrPatternMatch;
} else if (optr == TSDB_RELATION_NOT_LIKE) { /* wildcard query using like operator */
comparFn = compareStrPatternNotMatch;
} else if (optr == TSDB_RELATION_IN) {
comparFn = compareChkInString;
} else if (optr == TSDB_RELATION_NOT_IN) {
comparFn = compareChkNotInString;
} else { /* normal relational comparFn */
comparFn = compareLenPrefixedStr;
}
break;
}
case TSDB_DATA_TYPE_NCHAR: {
if (optr == TSDB_RELATION_MATCH) {
comparFn = compareStrRegexCompMatch;
} else if (optr == TSDB_RELATION_NMATCH) {
comparFn = compareStrRegexCompNMatch;
} else if (optr == TSDB_RELATION_LIKE) {
comparFn = compareWStrPatternMatch;
} else if (optr == TSDB_RELATION_NOT_LIKE) {
comparFn = compareWStrPatternNotMatch;
} else if (optr == TSDB_RELATION_IN) {
comparFn = compareChkInString;
} else if (optr == TSDB_RELATION_NOT_IN) {
comparFn = compareChkNotInString;
} else {
comparFn = compareLenPrefixedWStr;
}
break;
}
case TSDB_DATA_TYPE_UTINYINT: comparFn = compareUint8Val; break;
case TSDB_DATA_TYPE_USMALLINT: comparFn = compareUint16Val;break;
case TSDB_DATA_TYPE_UINT: comparFn = compareUint32Val;break;
case TSDB_DATA_TYPE_UBIGINT: comparFn = compareUint64Val;break;
default:
comparFn = compareInt32Val;
break;
}
return comparFn;
}

24
source/util/src/thashutil.c
View File

@ -164,20 +164,20 @@ _hash_fn_t taosGetDefaultHashFunction(int32_t type) {
return fn;
}
//int32_t taosFloatEqual(const void *a, const void *b, size_t UNUSED_PARAM(sz)) {
// return getComparFunc(TSDB_DATA_TYPE_FLOAT, -1)(a, b);
//}
//
//int32_t taosDoubleEqual(const void *a, const void *b, size_t UNUSED_PARAM(sz)) {
// return getComparFunc(TSDB_DATA_TYPE_DOUBLE, -1)(a, b);
//}
int32_t taosFloatEqual(const void *a, const void *b, size_t UNUSED_PARAM(sz)) {
return getComparFunc(TSDB_DATA_TYPE_FLOAT, -1)(a, b);
}
int32_t taosDoubleEqual(const void *a, const void *b, size_t UNUSED_PARAM(sz)) {
return getComparFunc(TSDB_DATA_TYPE_DOUBLE, -1)(a, b);
}
_equal_fn_t taosGetDefaultEqualFunction(int32_t type) {
_equal_fn_t fn = NULL;
// switch (type) {
// case TSDB_DATA_TYPE_FLOAT: fn = taosFloatEqual; break;
// case TSDB_DATA_TYPE_DOUBLE: fn = taosDoubleEqual; break;
// default: fn = memcmp; break;
// }
switch (type) {
case TSDB_DATA_TYPE_FLOAT: fn = taosFloatEqual; break;
case TSDB_DATA_TYPE_DOUBLE: fn = taosDoubleEqual; break;
default: fn = memcmp; break;
}
return fn;
}