Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
未验证 提交 ddfcac19 编写于 作者: H Haojun Liao 提交者： GitHub
浏览文件
操作

Merge pull request #11043 from taosdata/feature/3.0_liaohj 

Feature/3.0 liaohj
上级 9d7db266 9d5e46b6
展开全部 隐藏空白更改
内联 并排
Showing with 1427 addition and 2293 deletion
include/common/tcommon.h
浏览文件 @ ddfcac19
......@@ -54,16 +54,13 @@ typedef struct SColumnDataAgg {
} SColumnDataAgg;
typedef struct SDataBlockInfo {
STimeWindow window;
int32_t rows;
int32_t rowSize;
int16_t numOfCols;
int16_t hasVarCol;
union {
int64_t uid;
int64_t blockId;
};
int64_t groupId; // no need to serialize
STimeWindow window;
int32_t rows;
int32_t rowSize;
int16_t numOfCols;
int16_t hasVarCol;
union {int64_t uid; int64_t blockId;};
int64_t groupId; // no need to serialize
} SDataBlockInfo;
typedef struct SSDataBlock {
......@@ -96,6 +93,7 @@ void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock);
int32_t tEncodeDataBlocks(void** buf, const SArray* blocks);
void* tDecodeDataBlocks(const void* buf, SArray** blocks);
void colDataDestroy(SColumnInfoData* pColData) ;
static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) {
// WARNING: do not use info.numOfCols,
......@@ -103,13 +101,7 @@ static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) {
int32_t numOfOutput = taosArrayGetSize(pBlock->pDataBlock);
for (int32_t i = 0; i < numOfOutput; ++i) {
SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
taosMemoryFreeClear(pColInfoData->varmeta.offset);
} else {
taosMemoryFreeClear(pColInfoData->nullbitmap);
}
taosMemoryFreeClear(pColInfoData->pData);
colDataDestroy(pColInfoData);
}
taosArrayDestroy(pBlock->pDataBlock);
......
include/common/tdatablock.h
浏览文件 @ ddfcac19
......@@ -101,6 +101,54 @@ static FORCE_INLINE bool colDataIsNull(const SColumnInfoData* pColumnInfoData, u
((IS_VAR_DATA_TYPE((p1_)->info.type)) ? ((p1_)->pData + (p1_)->varmeta.offset[(r_)]) \
: ((p1_)->pData + ((r_) * (p1_)->info.bytes)))
static FORCE_INLINE void colDataAppendNULL(SColumnInfoData* pColumnInfoData, uint32_t currentRow) {
// There is a placehold for each NULL value of binary or nchar type.
if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
pColumnInfoData->varmeta.offset[currentRow] = -1; // it is a null value of VAR type.
} else {
colDataSetNull_f(pColumnInfoData->nullbitmap, currentRow);
}
pColumnInfoData->hasNull = true;
}
static FORCE_INLINE int32_t colDataAppendInt8(SColumnInfoData* pColumnInfoData, uint32_t currentRow, int8_t* v) {
ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_TINYINT ||
pColumnInfoData->info.type == TSDB_DATA_TYPE_UTINYINT || pColumnInfoData->info.type == TSDB_DATA_TYPE_BOOL);
char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
*(int8_t*)p = *(int8_t*)v;
}
static FORCE_INLINE int32_t colDataAppendInt16(SColumnInfoData* pColumnInfoData, uint32_t currentRow, int16_t* v) {
ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_SMALLINT || pColumnInfoData->info.type == TSDB_DATA_TYPE_USMALLINT);
char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
*(int16_t*)p = *(int16_t*)v;
}
static FORCE_INLINE int32_t colDataAppendInt32(SColumnInfoData* pColumnInfoData, uint32_t currentRow, int32_t* v) {
ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_INT || pColumnInfoData->info.type == TSDB_DATA_TYPE_UINT);
char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
*(int32_t*)p = *(int32_t*)v;
}
static FORCE_INLINE int32_t colDataAppendInt64(SColumnInfoData* pColumnInfoData, uint32_t currentRow, int64_t* v) {
ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_BIGINT || pColumnInfoData->info.type == TSDB_DATA_TYPE_UBIGINT);
char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
*(int64_t*)p = *(int64_t*)v;
}
static FORCE_INLINE int32_t colDataAppendFloat(SColumnInfoData* pColumnInfoData, uint32_t currentRow, float* v) {
ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_FLOAT);
char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
*(float*)p = *(float*)v;
}
static FORCE_INLINE int32_t colDataAppendDouble(SColumnInfoData* pColumnInfoData, uint32_t currentRow, double* v) {
ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_DOUBLE);
char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
*(double*)p = *(double*)v;
}
int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull);
int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, const SColumnInfoData* pSource,
uint32_t numOfRow2);
......
include/libs/function/function.h
浏览文件 @ ddfcac19
......@@ -27,16 +27,22 @@ extern "C" {
struct SqlFunctionCtx;
struct SResultRowEntryInfo;
typedef struct SFunctionNode SFunctionNode;
struct SFunctionNode;
typedef struct SScalarParam SScalarParam;
typedef struct SFuncExecEnv {
int32_t calcMemSize;
} SFuncExecEnv;
typedef bool (*FExecGetEnv)(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
typedef bool (*FExecGetEnv)(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
typedef bool (*FExecInit)(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
typedef void (*FExecProcess)(struct SqlFunctionCtx *pCtx);
typedef void (*FExecFinalize)(struct SqlFunctionCtx *pCtx);
typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
typedef struct SScalarFuncExecFuncs {
FScalarExecProcess process;
} SScalarFuncExecFuncs;
typedef struct SFuncExecFuncs {
FExecGetEnv getEnv;
......@@ -191,6 +197,7 @@ typedef struct SqlFunctionCtx {
SPoint1 start;
SPoint1 end;
SFuncExecFuncs fpSet;
SScalarFuncExecFuncs sfp;
} SqlFunctionCtx;
enum {
......@@ -203,7 +210,7 @@ enum {
};
typedef struct tExprNode {
uint8_t nodeType;
int32_t nodeType;
union {
struct {
int32_t optr; // binary operator
......@@ -219,7 +226,7 @@ typedef struct tExprNode {
char functionName[FUNCTIONS_NAME_MAX_LENGTH]; // todo refactor
int32_t functionId;
int32_t num;
SFunctionNode *pFunctNode;
struct SFunctionNode *pFunctNode;
// Note that the attribute of pChild is not the parameter of function, it is the columns that involved in the
// calculation instead.
// E.g., Cov(col1, col2), the column information, w.r.t. the col1 and col2, is kept in pChild nodes.
......@@ -227,6 +234,10 @@ typedef struct tExprNode {
// operator and is kept in the attribute of _node.
struct tExprNode **pChild;
} _function;
struct {
struct SNode* pRootNode;
} _optrRoot;
};
} tExprNode;
......@@ -250,25 +261,11 @@ typedef struct SAggFunctionInfo {
int32_t (*dataReqFunc)(SqlFunctionCtx *pCtx, STimeWindow* w, int32_t colId);
} SAggFunctionInfo;
typedef struct SScalarParam {
void *data;
union {
SColumnInfoData *columnData;
void *data;
} orig;
char *bitmap;
bool dataInBlock;
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 SScalarParam* pOutput, size_t numOfInput, const struct SScalarParam *pInput);
} SScalarFunctionInfo;
struct SScalarParam {
SColumnInfoData *columnData;
SHashObj *pHashFilter;
int32_t numOfRows;
};
typedef struct SMultiFunctionsDesc {
bool stableQuery;
......
include/libs/function/functionMgt.h
浏览文件 @ ddfcac19
......@@ -103,13 +103,6 @@ struct SqlFunctionCtx;
struct SResultRowEntryInfo;
struct STimeWindow;
typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
typedef struct SScalarFuncExecFuncs {
FScalarExecProcess process;
} SScalarFuncExecFuncs;
int32_t fmFuncMgtInit();
void fmFuncMgtDestroy();
......
include/libs/scalar/scalar.h
浏览文件 @ ddfcac19
......@@ -40,7 +40,7 @@ int32_t scalarGetOperatorParamNum(EOperatorType type);
int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type);
int32_t vectorGetConvertType(int32_t type1, int32_t type2);
int32_t vectorConvertImpl(SScalarParam* pIn, SScalarParam* pOut);
int32_t vectorConvertImpl(const SScalarParam* pIn, SScalarParam* pOut);
int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t logFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
......
source/client/inc/clientInt.h
浏览文件 @ ddfcac19
......@@ -235,7 +235,7 @@ void initMsgHandleFp();
TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
uint16_t port);
void* doFetchRow(SRequestObj* pRequest);
void* doFetchRow(SRequestObj* pRequest, bool setupOneRowPtr);
int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows);
......
source/client/src/clientImpl.c
浏览文件 @ ddfcac19
......@@ -545,7 +545,42 @@ TAOS* taos_connect_l(const char* ip, int ipLen, const char* user, int userLen, c
return taos_connect(ipStr, userStr, passStr, dbStr, port);
}
void* doFetchRow(SRequestObj* pRequest) {
static void doSetOneRowPtr(SReqResultInfo* pResultInfo) {
for (int32_t i = 0; i < pResultInfo->numOfCols; ++i) {
SResultColumn* pCol = &pResultInfo->pCol[i];
int32_t type = pResultInfo->fields[i].type;
int32_t bytes = pResultInfo->fields[i].bytes;
if (IS_VAR_DATA_TYPE(type)) {
if (pCol->offset[pResultInfo->current] != -1) {
char* pStart = pResultInfo->pCol[i].offset[pResultInfo->current] + pResultInfo->pCol[i].pData;
pResultInfo->length[i] = varDataLen(pStart);
pResultInfo->row[i] = varDataVal(pStart);
if (type == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4*)varDataVal(pStart), varDataLen(pStart), varDataVal(pResultInfo->convertBuf[i]));
ASSERT(len <= bytes);
pResultInfo->row[i] = varDataVal(pResultInfo->convertBuf[i]);
varDataSetLen(pResultInfo->convertBuf[i], len);
pResultInfo->length[i] = len;
}
} else {
pResultInfo->row[i] = NULL;
}
} else {
if (!colDataIsNull_f(pCol->nullbitmap, pResultInfo->current)) {
pResultInfo->row[i] = pResultInfo->pCol[i].pData + bytes * pResultInfo->current;
} else {
pResultInfo->row[i] = NULL;
}
}
}
}
void* doFetchRow(SRequestObj* pRequest, bool setupOneRowPtr) {
assert(pRequest != NULL);
SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
......@@ -555,17 +590,20 @@ void* doFetchRow(SRequestObj* pRequest) {
if (pRequest->type == TDMT_VND_QUERY) {
// All data has returned to App already, no need to try again
if (pResultInfo->completed) {
pResultInfo->numOfRows = 0;
return NULL;
}
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
pRequest->code = schedulerFetchRows(pRequest->body.queryJob, (void**)&pResInfo->pData);
if (pRequest->code != TSDB_CODE_SUCCESS) {
pResultInfo->numOfRows = 0;
return NULL;
}
pRequest->code = setQueryResultFromRsp(&pRequest->body.resInfo, (SRetrieveTableRsp*)pResInfo->pData);
if (pRequest->code != TSDB_CODE_SUCCESS) {
pResultInfo->numOfRows = 0;
return NULL;
}
......@@ -633,41 +671,11 @@ void* doFetchRow(SRequestObj* pRequest) {
}
_return:
for (int32_t i = 0; i < pResultInfo->numOfCols; ++i) {
SResultColumn* pCol = &pResultInfo->pCol[i];
int32_t type = pResultInfo->fields[i].type;
int32_t bytes = pResultInfo->fields[i].bytes;
if (IS_VAR_DATA_TYPE(type)) {
if (pCol->offset[pResultInfo->current] != -1) {
char* pStart = pResultInfo->pCol[i].offset[pResultInfo->current] + pResultInfo->pCol[i].pData;
pResultInfo->length[i] = varDataLen(pStart);
pResultInfo->row[i] = varDataVal(pStart);
if (type == TSDB_DATA_TYPE_NCHAR) {
int32_t len = taosUcs4ToMbs((TdUcs4*)varDataVal(pStart), varDataLen(pStart), varDataVal(pResultInfo->convertBuf[i]));
ASSERT(len <= bytes);
pResultInfo->row[i] = varDataVal(pResultInfo->convertBuf[i]);
varDataSetLen(pResultInfo->convertBuf[i], len);
pResultInfo->length[i] = len;
}
} else {
pResultInfo->row[i] = NULL;
}
} else {
if (!colDataIsNull_f(pCol->nullbitmap, pResultInfo->current)) {
pResultInfo->row[i] = pResultInfo->pCol[i].pData + bytes * pResultInfo->current;
} else {
pResultInfo->row[i] = NULL;
}
}
if (setupOneRowPtr) {
doSetOneRowPtr(pResultInfo);
pResultInfo->current += 1;
}
pResultInfo->current += 1;
return pResultInfo->row;
}
......
source/client/src/clientMain.c
浏览文件 @ ddfcac19
......@@ -138,20 +138,20 @@ TAOS_RES *taos_query(TAOS *taos, const char *sql) {
return taos_query_l(taos, sql, (int32_t) strlen(sql));
}
TAOS_ROW taos_fetch_row(TAOS_RES *pRes) {
if (pRes == NULL) {
TAOS_ROW taos_fetch_row(TAOS_RES *res) {
if (res == NULL) {
return NULL;
}
SRequestObj *pRequest = (SRequestObj *) pRes;
SRequestObj *pRequest = (SRequestObj *) res;
if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
pRequest->type == TSDB_SQL_INSERT ||
pRequest->code != TSDB_CODE_SUCCESS ||
taos_num_fields(pRes) == 0) {
taos_num_fields(res) == 0) {
return NULL;
}
return doFetchRow(pRequest);
return doFetchRow(pRequest, true);
}
int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) {
......@@ -246,6 +246,7 @@ int* taos_fetch_lengths(TAOS_RES *res) {
return ((SRequestObj*) res)->body.resInfo.length;
}
// todo intergrate with tDataTypes
const char *taos_data_type(int type) {
switch (type) {
case TSDB_DATA_TYPE_NULL: return "TSDB_DATA_TYPE_NULL";
......@@ -256,9 +257,11 @@ const char *taos_data_type(int type) {
case TSDB_DATA_TYPE_BIGINT: return "TSDB_DATA_TYPE_BIGINT";
case TSDB_DATA_TYPE_FLOAT: return "TSDB_DATA_TYPE_FLOAT";
case TSDB_DATA_TYPE_DOUBLE: return "TSDB_DATA_TYPE_DOUBLE";
case TSDB_DATA_TYPE_BINARY: return "TSDB_DATA_TYPE_BINARY";
case TSDB_DATA_TYPE_VARCHAR: return "TSDB_DATA_TYPE_VARCHAR";
// case TSDB_DATA_TYPE_BINARY: return "TSDB_DATA_TYPE_VARCHAR";
case TSDB_DATA_TYPE_TIMESTAMP: return "TSDB_DATA_TYPE_TIMESTAMP";
case TSDB_DATA_TYPE_NCHAR: return "TSDB_DATA_TYPE_NCHAR";
case TSDB_DATA_TYPE_JSON: return "TSDB_DATA_TYPE_JSON";
default: return "UNKNOWN";
}
}
......@@ -316,11 +319,37 @@ void taos_stop_query(TAOS_RES *res) {
}
bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
return false;
SRequestObj* pRequestObj = res;
SReqResultInfo* pResultInfo = &pRequestObj->body.resInfo;
if (col >= pResultInfo->numOfCols || col < 0 || row >= pResultInfo->numOfRows || row < 0) {
return true;
}
SResultColumn* pCol = &pRequestObj->body.resInfo.pCol[col];
return colDataIsNull_f(pCol->nullbitmap, row);
}
int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
return 0;
if (res == NULL) {
return 0;
}
SRequestObj *pRequest = (SRequestObj *) res;
if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
pRequest->type == TSDB_SQL_INSERT ||
pRequest->code != TSDB_CODE_SUCCESS ||
taos_num_fields(res) == 0) {
return 0;
}
doFetchRow(pRequest, false);
// TODO refactor
SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
pResultInfo->current = pResultInfo->numOfRows;
*rows = pResultInfo->row;
return pResultInfo->numOfRows;
}
int taos_validate_sql(TAOS *taos, const char *sql) {
......
source/common/src/tdatablock.c
浏览文件 @ ddfcac19
......@@ -1241,6 +1241,16 @@ size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize) {
return pageSize / (blockDataGetSerialRowSize(pBlock) + blockDataGetSerialMetaSize(pBlock));
}
void colDataDestroy(SColumnInfoData* pColData) {
if (IS_VAR_DATA_TYPE(pColData->info.type)) {
taosMemoryFree(pColData->varmeta.offset);
} else {
taosMemoryFree(pColData->nullbitmap);
}
taosMemoryFree(pColData->pData);
}
int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
int64_t tbUid = pBlock->info.uid;
int16_t numOfCols = pBlock->info.numOfCols;
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ ddfcac19
......@@ -469,7 +469,7 @@ typedef struct SOptrBasicInfo {
int32_t* rowCellInfoOffset; // offset value for each row result cell info
SqlFunctionCtx* pCtx;
SSDataBlock* pRes;
int32_t capacity;
int32_t capacity; // TODO remove it
} SOptrBasicInfo;
//TODO move the resultrowsiz together with SOptrBasicInfo:rowCellInfoOffset
......
source/libs/executor/src/executorimpl.c
浏览文件 @ ddfcac19
......@@ -1246,17 +1246,40 @@ static void doAggregateImpl(SOperatorInfo* pOperator, TSKEY startTs, SqlFunction
}
}
static void projectApplyFunctions(SSDataBlock* pResult, SqlFunctionCtx *pCtx, int32_t numOfOutput) {
static void projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx *pCtx, int32_t numOfOutput) {
for (int32_t k = 0; k < numOfOutput; ++k) {
if (pCtx[k].fpSet.init == NULL) { // it is a project query
if (pExpr[k].pExpr->nodeType == QUERY_NODE_COLUMN) { // it is a project query
SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, k);
colDataAssign(pColInfoData, pCtx[k].input.pData[0], pCtx[k].input.numOfRows);
} else { // TODO: arithmetic and other process.
pResult->info.rows = pCtx[0].input.numOfRows;
} else if (pExpr[k].pExpr->nodeType == QUERY_NODE_OPERATOR) {
SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
taosArrayPush(pBlockList, &pSrcBlock);
SScalarParam dest = {0};
dest.columnData = taosArrayGet(pResult->pDataBlock, k);
scalarCalculate(pExpr[k].pExpr->_optrRoot.pRootNode, pBlockList, &dest);
pResult->info.rows = dest.numOfRows;
taosArrayDestroy(pBlockList);
} else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
ASSERT(!fmIsAggFunc(pCtx->functionId));
SScalarParam p = {.numOfRows = pSrcBlock->info.rows};
int32_t slotId = pExpr[k].base.pParam[0].pCol->slotId;
p.columnData = taosArrayGet(pSrcBlock->pDataBlock, slotId);
SScalarParam dest = {0};
dest.columnData = taosArrayGet(pResult->pDataBlock, k);
pCtx[k].sfp.process(&p, 1, &dest);
pResult->info.rows = dest.numOfRows;
} else {
ASSERT(0);
}
}
pResult->info.rows = pCtx[0].input.numOfRows;
}
void doTimeWindowInterpolation(SOperatorInfo* pOperator, SOptrBasicInfo* pInfo, SArray* pDataBlock, TSKEY prevTs,
......@@ -2013,107 +2036,6 @@ static int32_t setCtxTagColumnInfo(SqlFunctionCtx *pCtx, int32_t numOfOutput) {
return TSDB_CODE_SUCCESS;
}
static SqlFunctionCtx* createSqlFunctionCtx(STaskRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput,
int32_t** rowCellInfoOffset) {
STaskAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
SqlFunctionCtx * pFuncCtx = (SqlFunctionCtx *)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx));
if (pFuncCtx == NULL) {
return NULL;
}
*rowCellInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t));
if (*rowCellInfoOffset == 0) {
taosMemoryFreeClear(pFuncCtx);
return NULL;
}
for (int32_t i = 0; i < numOfOutput; ++i) {
SExprBasicInfo *pFunct = &pExpr[i].base;
SqlFunctionCtx* pCtx = &pFuncCtx[i];
#if 0
SColIndex *pIndex = &pFunct->colInfo;
if (TSDB_COL_REQ_NULL(pIndex->flag)) {
pCtx->requireNull = true;
pIndex->flag &= ~(TSDB_COL_NULL);
} else {
pCtx->requireNull = false;
}
#endif
// pCtx->inputBytes = pFunct->colBytes;
// pCtx->inputType = pFunct->colType;
pCtx->ptsOutputBuf = NULL;
pCtx->resDataInfo.bytes = pFunct->resSchema.bytes;
pCtx->resDataInfo.type = pFunct->resSchema.type;
pCtx->order = pQueryAttr->order.order;
// pCtx->functionId = pFunct->functionId;
pCtx->stableQuery = pQueryAttr->stableQuery;
// pCtx->resDataInfo.interBufSize = pFunct->interBytes;
pCtx->start.key = INT64_MIN;
pCtx->end.key = INT64_MIN;
pCtx->numOfParams = pFunct->numOfParams;
for (int32_t j = 0; j < pCtx->numOfParams; ++j) {
int16_t type = pFunct->pParam[j].param.nType;
int16_t bytes = pFunct->pParam[j].param.nType;
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
// taosVariantCreateFromBinary(&pCtx->param[j], pFunct->param[j].pz, bytes, type);
} else {
// taosVariantCreateFromBinary(&pCtx->param[j], (char *)&pFunct->param[j].i, bytes, type);
}
}
// set the order information for top/bottom query
int32_t functionId = pCtx->functionId;
if (functionId == FUNCTION_TOP || functionId == FUNCTION_BOTTOM || functionId == FUNCTION_DIFF) {
int32_t f = getExprFunctionId(&pExpr[0]);
assert(f == FUNCTION_TS || f == FUNCTION_TS_DUMMY);
pCtx->param[2].i = pQueryAttr->order.order;
pCtx->param[2].nType = TSDB_DATA_TYPE_BIGINT;
pCtx->param[3].i = functionId;
pCtx->param[3].nType = TSDB_DATA_TYPE_BIGINT;
pCtx->param[1].i = pQueryAttr->order.col.colId;
} else if (functionId == FUNCTION_INTERP) {
pCtx->param[2].i = (int8_t)pQueryAttr->fillType;
if (pQueryAttr->fillVal != NULL) {
if (isNull((const char *)&pQueryAttr->fillVal[i], pCtx->inputType)) {
pCtx->param[1].nType = TSDB_DATA_TYPE_NULL;
} else { // todo refactor, taosVariantCreateFromBinary should handle the NULL value
if (pCtx->inputType != TSDB_DATA_TYPE_BINARY && pCtx->inputType != TSDB_DATA_TYPE_NCHAR) {
taosVariantCreateFromBinary(&pCtx->param[1], (char *)&pQueryAttr->fillVal[i], pCtx->inputBytes, pCtx->inputType);
}
}
}
} else if (functionId == FUNCTION_TS_COMP) {
pCtx->param[0].i = pQueryAttr->vgId; //TODO this should be the parameter from client
pCtx->param[0].nType = TSDB_DATA_TYPE_BIGINT;
} else if (functionId == FUNCTION_TWA) {
pCtx->param[1].i = pQueryAttr->window.skey;
pCtx->param[1].nType = TSDB_DATA_TYPE_BIGINT;
pCtx->param[2].i = pQueryAttr->window.ekey;
pCtx->param[2].nType = TSDB_DATA_TYPE_BIGINT;
} else if (functionId == FUNCTION_ARITHM) {
// pCtx->param[1].pz = (char*) getScalarFuncSupport(pRuntimeEnv->scalarSup, i);
}
}
// for(int32_t i = 1; i < numOfOutput; ++i) {
// (*rowCellInfoOffset)[i] = (int32_t)((*rowCellInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) + pExpr[i - 1].base.interBytes);
// }
setCtxTagColumnInfo(pFuncCtx, numOfOutput);
return pFuncCtx;
}
static SqlFunctionCtx* createSqlFunctionCtx_rv(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowCellInfoOffset) {
SqlFunctionCtx * pFuncCtx = (SqlFunctionCtx *)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx));
if (pFuncCtx == NULL) {
......@@ -2132,15 +2054,22 @@ static SqlFunctionCtx* createSqlFunctionCtx_rv(SExprInfo* pExprInfo, int32_t num
SExprBasicInfo *pFunct = &pExpr->base;
SqlFunctionCtx* pCtx = &pFuncCtx[i];
if (pExpr->pExpr->_function.pFunctNode != NULL) {
pCtx->functionId = -1;
if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) {
SFuncExecEnv env = {0};
pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId;
fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);
pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
if (fmIsAggFunc(pCtx->functionId)) {
fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);
pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
} else {
fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp);
}
pCtx->resDataInfo.interBufSize = env.calcMemSize;
} else {
pCtx->functionId = -1;
} else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN) {
} else if (pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR) {
}
pCtx->input.numOfInputCols = pFunct->numOfParams;
......@@ -5578,11 +5507,11 @@ EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) {
SOperatorNode *node = (SOperatorNode *)pNode;
if (OP_TYPE_EQUAL == node->opType) {
*(int32_t *)pContext = 1;
*(int32_t *)pContext = 1;
return DEAL_RES_CONTINUE;
}
*(int32_t *)pContext = 0;
*(int32_t *)pContext = 0;
return DEAL_RES_IGNORE_CHILD;
}
......@@ -5590,14 +5519,14 @@ EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) {
if (1 != *(int32_t *)pContext) {
return DEAL_RES_CONTINUE;
}
SColumnNode *node = (SColumnNode *)pNode;
if (TSDB_INS_USER_STABLES_DBNAME_COLID == node->colId) {
*(int32_t *)pContext = 2;
*(int32_t *)pContext = 2;
return DEAL_RES_CONTINUE;
}
*(int32_t *)pContext = 0;
*(int32_t *)pContext = 0;
return DEAL_RES_CONTINUE;
}
case QUERY_NODE_VALUE: {
......@@ -5607,7 +5536,7 @@ EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) {
SValueNode *node = (SValueNode *)pNode;
char *dbName = nodesGetValueFromNode(node);
strncpy(pContext, varDataVal(dbName), varDataLen(dbName));
strncpy(pContext, varDataVal(dbName), varDataLen(dbName));
*((char *)pContext + varDataLen(dbName)) = 0;
return DEAL_RES_ERROR; // stop walk
}
......@@ -6654,7 +6583,6 @@ static SSDataBlock* doProjectOperation(SOperatorInfo *pOperator, bool* newgroup)
blockDataCleanup(pRes);
if (pProjectInfo->existDataBlock) { // TODO refactor
// STableQueryInfo* pTableQueryInfo = pRuntimeEnv->current;
SSDataBlock* pBlock = pProjectInfo->existDataBlock;
pProjectInfo->existDataBlock = NULL;
*newgroup = true;
......@@ -6668,9 +6596,7 @@ static SSDataBlock* doProjectOperation(SOperatorInfo *pOperator, bool* newgroup)
setInputDataBlock(pOperator, pInfo->pCtx, pBlock, TSDB_ORDER_ASC);
blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);
projectApplyFunctions(pInfo->pRes, pInfo->pCtx, pOperator->numOfOutput);
pRes->info.rows = getNumOfResult(pInfo->pCtx, pOperator->numOfOutput, NULL);
projectApplyFunctions(pOperator->pExpr, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->numOfOutput);
if (pRes->info.rows >= pProjectInfo->binfo.capacity*0.8) {
copyTsColoum(pRes, pInfo->pCtx, pOperator->numOfOutput);
resetResultRowEntryResult(pInfo->pCtx, pOperator->numOfOutput);
......@@ -6713,15 +6639,15 @@ static SSDataBlock* doProjectOperation(SOperatorInfo *pOperator, bool* newgroup)
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pInfo->pCtx, pBlock, TSDB_ORDER_ASC);
updateOutputBuf(pInfo, &pInfo->capacity, pBlock->info.rows);
blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
projectApplyFunctions(pInfo->pRes, pInfo->pCtx, pOperator->numOfOutput);
if (pRes->info.rows >= pProjectInfo->threshold) {
projectApplyFunctions(pOperator->pExpr, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->numOfOutput);
if (pRes->info.rows >= pOperator->resultInfo.threshold) {
break;
}
}
copyTsColoum(pRes, pInfo->pCtx, pOperator->numOfOutput);
// copyTsColoum(pRes, pInfo->pCtx, pOperator->numOfOutput);
return (pInfo->pRes->info.rows > 0)? pInfo->pRes:NULL;
}
......@@ -7811,7 +7737,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput) {
STableIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableIntervalOperatorInfo));
pInfo->binfo.pCtx = createSqlFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// pInfo->binfo.pCtx = createSqlFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, pResultInfo->capacity);
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);
......@@ -7836,7 +7762,7 @@ SOperatorInfo* createStatewindowOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOper
SStateWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStateWindowOperatorInfo));
pInfo->colIndex = -1;
pInfo->reptScan = false;
pInfo->binfo.pCtx = createSqlFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// pInfo->binfo.pCtx = createSqlFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, pResultInfo->capacity);
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);
......@@ -7901,7 +7827,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo
SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput) {
STableIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableIntervalOperatorInfo));
pInfo->binfo.pCtx = createSqlFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// pInfo->binfo.pCtx = createSqlFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, pResultInfo->capacity);
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);
......@@ -7925,7 +7851,7 @@ SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntim
SOperatorInfo* createAllMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput) {
STableIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableIntervalOperatorInfo));
pInfo->binfo.pCtx = createSqlFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// pInfo->binfo.pCtx = createSqlFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
// pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, pResultInfo->capacity);
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);
......@@ -8533,16 +8459,18 @@ SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t*
// it is a project query, or group by column
if (nodeType(pTargetNode->pExpr) == QUERY_NODE_COLUMN) {
pExp->pExpr->nodeType = QUERY_NODE_COLUMN;
SColumnNode* pColNode = (SColumnNode*) pTargetNode->pExpr;
SDataType* pType = &pColNode->node.resType;
pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, pType->precision, pColNode->colName);
pCol->slotId = pColNode->slotId;
pCol->slotId = pColNode->slotId; // TODO refactor
pCol->bytes = pType->bytes;
pCol->type = pType->type;
pCol->scale = pType->scale;
pCol->precision = pType->precision;
} else {
} else if (nodeType(pTargetNode->pExpr) == QUERY_NODE_FUNCTION) {
pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;
SFunctionNode* pFuncNode = (SFunctionNode*)pTargetNode->pExpr;
SDataType* pType = &pFuncNode->node.resType;
......@@ -8556,7 +8484,7 @@ SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t*
int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);
for (int32_t j = 0; j < numOfParam; ++j) {
SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j);
SColumnNode* pcn = (SColumnNode*)p1;
SColumnNode* pcn = (SColumnNode*)p1; // TODO refactor
pCol->slotId = pcn->slotId;
pCol->bytes = pcn->node.resType.bytes;
......@@ -8565,6 +8493,22 @@ SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t*
pCol->precision = pcn->node.resType.precision;
pCol->dataBlockId = pcn->dataBlockId;
}
} else if (nodeType(pTargetNode->pExpr) == QUERY_NODE_OPERATOR) {
pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
SOperatorNode* pNode = (SOperatorNode*) pTargetNode->pExpr;
SDataType* pType = &pNode->node.resType;
pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, pType->precision, pNode->node.aliasName);
pExp->pExpr->_optrRoot.pRootNode = pTargetNode->pExpr;
pCol->slotId = pTargetNode->slotId; // TODO refactor
pCol->bytes = pType->bytes;
pCol->type = pType->type;
pCol->scale = pType->scale;
pCol->precision = pType->precision;
} else {
ASSERT(0);
}
}
......@@ -8627,7 +8571,7 @@ SOperatorInfo* doCreateOperatorTreeNode(SPhysiNode* pPhyNode, SExecTaskInfo* pTa
SArray* colList = extractScanColumnId(pScanNode->pScanCols);
SOperatorInfo* pOperator = createSysTableScanOperatorInfo(pHandle->meta, pResBlock, &pScanNode->tableName,
pScanNode->node.pConditions, pSysScanPhyNode->mgmtEpSet,
pScanNode->node.pConditions, pSysScanPhyNode->mgmtEpSet,
colList, pTaskInfo, pSysScanPhyNode->showRewrite, pSysScanPhyNode->accountId);
return pOperator;
} else {
......
source/libs/function/inc/builtinsimpl.h
浏览文件 @ ddfcac19
......@@ -25,23 +25,23 @@ extern "C" {
bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
void functionFinalize(SqlFunctionCtx *pCtx);
bool getCountFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getCountFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void countFunction(SqlFunctionCtx *pCtx);
bool getSumFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getSumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void sumFunction(SqlFunctionCtx *pCtx);
bool minFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool maxFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getMinmaxFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getMinmaxFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void minFunction(SqlFunctionCtx* pCtx);
void maxFunction(SqlFunctionCtx *pCtx);
bool getStddevFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void stddevFunction(SqlFunctionCtx* pCtx);
void stddevFinalize(SqlFunctionCtx* pCtx);
bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void firstFunction(SqlFunctionCtx *pCtx);
void lastFunction(SqlFunctionCtx *pCtx);
......
source/libs/function/src/builtins.c
浏览文件 @ ddfcac19
......@@ -331,6 +331,14 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
case FUNCTION_TYPE_CONCAT:
// todo
break;
case FUNCTION_TYPE_ABS: {
SColumnNode* pParam = nodesListGetNode(pFunc->pParameterList, 0);
int32_t paraType = pParam->node.resType.type;
pFunc->node.resType = (SDataType) { .bytes = tDataTypes[paraType].bytes, .type = paraType };
break;
}
default:
ASSERT(0); // to found the fault ASAP.
}
......
source/libs/function/src/taggfunction.c
浏览文件 @ ddfcac19
......@@ -3078,8 +3078,8 @@ static void arithmetic_function(SqlFunctionCtx *pCtx) {
GET_RES_INFO(pCtx)->numOfRes += pCtx->size;
//SScalarFunctionSupport *pSup = (SScalarFunctionSupport *)pCtx->param[1].pz;
SScalarParam output = {0};
output.data = pCtx->pOutput;
// SScalarParam output = {0};
// output.data = pCtx->pOutput;
//evaluateExprNodeTree(pSup->pExprInfo->pExpr, pCtx->size, &output, pSup, getArithColumnData);
}
......
source/libs/scalar/inc/sclInt.h
浏览文件 @ ddfcac19
......@@ -43,10 +43,12 @@ typedef struct SScalarCtx {
#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)
int32_t doConvertDataType(SValueNode* pValueNode, SScalarParam* out);
SColumnInfoData* createColumnInfoData(SDataType* pType, int32_t numOfRows);
#define GET_PARAM_TYPE(_c) ((_c)->columnData->info.type)
#define GET_PARAM_BYTES(_c) ((_c)->pColumnInfoData->info.bytes)
int32_t sclMoveParamListData(SScalarParam *params, int32_t listNum, int32_t idx);
bool sclIsNull(SScalarParam* param, int32_t idx);
void sclSetNull(SScalarParam* param, int32_t idx);
void sclFreeParam(SScalarParam *param);
#ifdef __cplusplus
......
source/libs/scalar/inc/sclfunc.h
浏览文件 @ ddfcac19
......@@ -22,19 +22,7 @@ extern "C" {
#include "function.h"
#include "scalar.h"
typedef struct SScalarFunctionSupport {
struct SExprInfo *pExprInfo;
int32_t numOfCols;
SColumnInfo *colList;
void *exprList; // client side used
int32_t offset;
char** data;
} SScalarFunctionSupport;
extern struct SScalarFunctionInfo scalarFunc[8];
int32_t evaluateExprNodeTree(tExprNode* pExprs, int32_t numOfRows, SScalarParam* pOutput,
void* param, char* (*getSourceDataBlock)(void*, const char*, int32_t));
#ifdef __cplusplus
......
source/libs/scalar/inc/sclvector.h
浏览文件 @ ddfcac19
......@@ -20,10 +20,66 @@
extern "C" {
#endif
#include "sclfunc.h"
typedef double (*_getDoubleValue_fn_t)(void *src, int32_t index);
typedef double (*_mathFunc)(double, double, bool *);
static FORCE_INLINE double getVectorDoubleValue_TINYINT(void *src, int32_t index) {
return (double)*((int8_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_UTINYINT(void *src, int32_t index) {
return (double)*((uint8_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_SMALLINT(void *src, int32_t index) {
return (double)*((int16_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_USMALLINT(void *src, int32_t index) {
return (double)*((uint16_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_INT(void *src, int32_t index) {
return (double)*((int32_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_UINT(void *src, int32_t index) {
return (double)*((uint32_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_BIGINT(void *src, int32_t index) {
return (double)*((int64_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_UBIGINT(void *src, int32_t index) {
return (double)*((uint64_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_FLOAT(void *src, int32_t index) {
return (double)*((float *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_DOUBLE(void *src, int32_t index) {
return (double)*((double *)src + index);
}
static FORCE_INLINE _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 (*_bufConverteFunc)(char *buf, SScalarParam* pOut, int32_t outType);
typedef void (*_bin_scalar_fn_t)(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *output, int32_t order);
......
source/libs/scalar/src/filter.c
浏览文件 @ ddfcac19
......@@ -318,7 +318,7 @@ static FORCE_INLINE SFilterRangeNode* filterNewRange(SFilterRangeCtx *ctx, SFilt
r->prev = NULL;
r->next = NULL;
} else {
r = taosMemoryCalloc(1, sizeof(SFilterRangeNode));
r = taosMemoryCalloc(1, sizeof(SFilterRangeNode));
}
FILTER_COPY_RA(&r->ra, ra);
......@@ -1021,26 +1021,21 @@ int32_t fltAddGroupUnitFromNode(SFilterInfo *info, SNode* tree, SArray *group) {
if (node->opType == OP_TYPE_IN && (!IS_VAR_DATA_TYPE(type))) {
SNodeListNode *listNode = (SNodeListNode *)node->pRight;
SListCell *cell = listNode->pNodeList->pHead;
SScalarParam in = {.num = 1}, out = {.num = 1, .type = type};
SScalarParam out = {.columnData = taosMemoryCalloc(1, sizeof(SColumnInfoData))};
out.columnData->info.type = type;
for (int32_t i = 0; i < listNode->pNodeList->length; ++i) {
SValueNode *valueNode = (SValueNode *)cell->pNode;
in.type = valueNode->node.resType.type;
in.bytes = valueNode->node.resType.bytes;
in.data = nodesGetValueFromNode(valueNode);
out.data = taosMemoryMalloc(sizeof(int64_t));
code = vectorConvertImpl(&in, &out);
code = doConvertDataType(valueNode, &out);
if (code) {
fltError("convert from %d to %d failed", in.type, out.type);
taosMemoryFreeClear(out.data);
// fltError("convert from %d to %d failed", in.type, out.type);
FLT_ERR_RET(code);
}
len = tDataTypes[type].bytes;
filterAddField(info, NULL, &out.data, FLD_TYPE_VALUE, &right, len, true);
filterAddField(info, NULL, (void**) &out.columnData->pData, FLD_TYPE_VALUE, &right, len, true);
filterAddUnit(info, OP_TYPE_EQUAL, &left, &right, &uidx);
SFilterGroup fgroup = {0};
......@@ -1054,7 +1049,6 @@ int32_t fltAddGroupUnitFromNode(SFilterInfo *info, SNode* tree, SArray *group) {
filterAddFieldFromNode(info, node->pRight, &right);
FLT_ERR_RET(filterAddUnit(info, node->opType, &left, &right, &uidx));
SFilterGroup fgroup = {0};
filterAddUnitToGroup(&fgroup, uidx);
......@@ -1080,7 +1074,6 @@ int32_t filterAddUnitFromUnit(SFilterInfo *dst, SFilterInfo *src, SFilterUnit* u
filterAddField(dst, NULL, &data, FLD_TYPE_VALUE, &right, POINTER_BYTES, false); // POINTER_BYTES should be sizeof(SHashObj), but POINTER_BYTES is also right.
t = FILTER_GET_FIELD(dst, right);
FILTER_SET_FLAG(t->flag, FLD_DATA_IS_HASH);
} else {
filterAddField(dst, NULL, &data, FLD_TYPE_VALUE, &right, varDataTLen(data), false);
......@@ -1101,14 +1094,12 @@ int32_t filterAddUnitFromUnit(SFilterInfo *dst, SFilterInfo *src, SFilterUnit* u
int32_t filterAddUnitRight(SFilterInfo *info, uint8_t optr, SFilterFieldId *right, uint32_t uidx) {
SFilterUnit *u = &info->units[uidx];
u->compare.optr2 = optr;
u->right2 = *right;
return TSDB_CODE_SUCCESS;
}
int32_t filterAddGroupUnitFromCtx(SFilterInfo *dst, SFilterInfo *src, SFilterRangeCtx *ctx, uint32_t cidx, SFilterGroup *g, int32_t optr, SArray *res) {
SFilterFieldId left, right, right2;
uint32_t uidx = 0;
......@@ -1800,9 +1791,12 @@ int32_t fltInitValFieldData(SFilterInfo *info) {
if (dType->type == type) {
assignVal(fi->data, nodesGetValueFromNode(var), dType->bytes, type);
} else {
SScalarParam in = {.data = nodesGetValueFromNode(var), .num = 1, .type = dType->type, .bytes = dType->bytes};
SScalarParam out = {.data = fi->data, .num = 1, .type = type};
if (vectorConvertImpl(&in, &out)) {
SScalarParam out = {.columnData = taosMemoryCalloc(1, sizeof(SColumnInfoData))};
out.columnData->info.type = type;
// todo refactor the convert
int32_t code = doConvertDataType(var, &out);
if (code != TSDB_CODE_SUCCESS) {
qError("convert value to type[%d] failed", type);
return TSDB_CODE_TSC_INVALID_OPERATION;
}
......@@ -3636,7 +3630,7 @@ int32_t filterInitFromNode(SNode* pNode, SFilterInfo **pInfo, uint32_t options)
if (*pInfo == NULL) {
*pInfo = taosMemoryCalloc(1, sizeof(SFilterInfo));
if (NULL == *pInfo) {
fltError("calloc %d failed", (int32_t)sizeof(SFilterInfo));
fltError("taosMemoryCalloc %d failed", (int32_t)sizeof(SFilterInfo));
FLT_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
}
......@@ -3676,18 +3670,18 @@ bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t** p, SColumnData
FLT_ERR_RET(scalarCalculate(info->sclCtx.node, pList, &output));
taosArrayDestroy(pList);
*p = output.orig.data;
output.orig.data = NULL;
sclFreeParam(&output);
int8_t *r = output.data;
for (int32_t i = 0; i < output.num; ++i) {
if (0 == *(r+i)) {
return false;
}
}
// TODO Fix it
// *p = output.orig.data;
// output.orig.data = NULL;
//
// sclFreeParam(&output);
//
// int8_t *r = output.data;
// for (int32_t i = 0; i < output.num; ++i) {
// if (0 == *(r+i)) {
// return false;
// }
// }
return true;
}
......
source/libs/scalar/src/scalar.c
浏览文件 @ ddfcac19
此差异已折叠。
source/libs/scalar/src/sclfunc.c
浏览文件 @ ddfcac19
此差异已折叠。
source/libs/scalar/src/sclvector.c
浏览文件 @ ddfcac19
此差异已折叠。
source/libs/transport/test/transportTests.cpp
浏览文件 @ ddfcac19
......@@ -156,14 +156,14 @@ TEST_F(TransCtxEnv, mergeTest) {
STransCtx *src = (STransCtx *)taosMemoryCalloc(1, sizeof(STransCtx));
transCtxInit(src);
{
STransCtxVal val1 = {.val = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
val1.val = taosMemoryMalloc(12);
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
key++;
}
{
STransCtxVal val1 = {.val = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
val1.val = taosMemoryMalloc(12);
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
key++;
......@@ -176,14 +176,14 @@ TEST_F(TransCtxEnv, mergeTest) {
STransCtx *src = (STransCtx *)taosMemoryCalloc(1, sizeof(STransCtx));
transCtxInit(src);
{
STransCtxVal val1 = {.val = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
val1.val = taosMemoryMalloc(12);
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
key++;
}
{
STransCtxVal val1 = {.val = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
val1.val = taosMemoryMalloc(12);
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
key++;
......@@ -198,7 +198,7 @@ TEST_F(TransCtxEnv, mergeTest) {
STransCtx *src = (STransCtx *)taosMemoryCalloc(1, sizeof(STransCtx));
transCtxInit(src);
{
STransCtxVal val1 = {.val = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
val1.val = taosMemoryCalloc(1, 11);
memcpy(val1.val, val.c_str(), val.size());
......@@ -206,7 +206,7 @@ TEST_F(TransCtxEnv, mergeTest) {
key++;
}
{
STransCtxVal val1 = {.val = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
val1.val = taosMemoryCalloc(1, 11);
memcpy(val1.val, val.c_str(), val.size());
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
......
source/os/src/osMemory.c
浏览文件 @ ddfcac19
......@@ -21,11 +21,10 @@
#define TD_MEMORY_STACK_TRACE_DEPTH 10
typedef struct TdMemoryInfo
{
typedef struct TdMemoryInfo {
int32_t symbol;
int32_t memorySize;
void *stackTrace[TD_MEMORY_STACK_TRACE_DEPTH]; // gdb: disassemble /m 0xXXX
int32_t memorySize;
} *TdMemoryInfoPtr , TdMemoryInfo;
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或