Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

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

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

[td-14393] support percentile.
上级 92fd7cc3 a4c7e78d
展开全部 隐藏空白更改
内联 并排
Showing with 1806 addition and 1512 deletion
include/libs/function/function.h
浏览文件 @ 9317b5f2
......@@ -36,7 +36,7 @@ typedef struct SFuncExecEnv {
typedef bool (*FExecGetEnv)(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
typedef bool (*FExecInit)(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
typedef void (*FExecProcess)(struct SqlFunctionCtx *pCtx);
typedef int32_t (*FExecProcess)(struct SqlFunctionCtx *pCtx);
typedef void (*FExecFinalize)(struct SqlFunctionCtx *pCtx);
typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
......@@ -154,7 +154,9 @@ typedef struct SResultDataInfo {
int32_t interBufSize;
} SResultDataInfo;
#define GET_RES_INFO(ctx) ((ctx)->resultInfo)
#define GET_RES_INFO(ctx) ((ctx)->resultInfo)
#define GET_ROWCELL_INTERBUF(_c) ((void*) ((char*)(_c) + sizeof(SResultRowEntryInfo)))
#define DATA_SET_FLAG ',' // to denote the output area has data, not null value
typedef struct SInputColumnInfoData {
int32_t totalRows; // total rows in current columnar data
......@@ -192,7 +194,8 @@ typedef struct SqlFunctionCtx {
int32_t numOfParams;
SVariant param[4]; // input parameter, e.g., top(k, 20), the number of results for top query is kept in param
int64_t *ptsList; // corresponding timestamp array list
void *ptsOutputBuf; // corresponding output buffer for timestamp of each result, e.g., top/bottom*/
SColumnInfoData *pTsOutput; // corresponding output buffer for timestamp of each result, e.g., top/bottom*/
int32_t offset;
SVariant tag;
struct SResultRowEntryInfo *resultInfo;
SSubsidiaryResInfo subsidiaryRes;
......
source/libs/executor/CMakeLists.txt
浏览文件 @ 9317b5f2
aux_source_directory(src EXECUTOR_SRC)
#add_library(executor ${EXECUTOR_SRC})
#target_link_libraries(
# executor
# PRIVATE os util common function parser planner qcom tsdb
#)
add_library(executor STATIC ${EXECUTOR_SRC})
#set_target_properties(executor PROPERTIES
# IMPORTED_LOCATION "${CMAKE_CURRENT_SOURCE_DIR}/libexecutor.a"
# INTERFACE_INCLUDE_DIRECTORIES "${CMAKE_SOURCE_DIR}/include/libs/executor"
# )
target_link_libraries(executor
PRIVATE os util common function parser planner qcom vnode scalar nodes
)
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 9317b5f2
......@@ -385,6 +385,12 @@ typedef struct SExchangeInfo {
SLoadRemoteDataInfo loadInfo;
} SExchangeInfo;
typedef struct SColMatchInfo {
int32_t colId;
int32_t targetSlotId;
bool output;
} SColMatchInfo;
typedef struct STableScanInfo {
void* dataReader;
int32_t numOfBlocks; // extract basic running information.
......@@ -497,8 +503,9 @@ typedef struct SAggOperatorInfo {
typedef struct SProjectOperatorInfo {
SOptrBasicInfo binfo;
SAggSupporter aggSup;
SSDataBlock *existDataBlock;
int32_t threshold;
SArray *pPseudoColInfo;
SLimit limit;
int64_t curOffset;
int64_t curOutput;
......@@ -623,13 +630,28 @@ typedef struct SDistinctOperatorInfo {
SHashObj* pSet;
SSDataBlock* pRes;
bool recordNullVal; // has already record the null value, no need to try again
int64_t threshold;
int64_t outputCapacity;
int32_t totalBytes;
int64_t threshold; // todo remove it
int64_t outputCapacity;// todo remove it
int32_t totalBytes; // todo remove it
char* buf;
SArray* pDistinctDataInfo;
} SDistinctOperatorInfo;
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator);
void operatorDummyCloseFn(void* param, int32_t numOfCols);
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
int32_t initAggInfo(SOptrBasicInfo* pBasicInfo, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols,
int32_t numOfRows, SSDataBlock* pResultBlock, const char* pkey);
void toSDatablock(SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf, SSDataBlock* pBlock, int32_t rowCapacity, int32_t* rowCellOffset);
void finalizeMultiTupleQueryResult(SqlFunctionCtx* pCtx, int32_t numOfOutput, SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
void doApplyFunctions(SqlFunctionCtx* pCtx, STimeWindow* pWin, SColumnInfoData* pTimeWindowData, int32_t offset, int32_t forwardStep, TSKEY* tsCol, int32_t numOfTotal, int32_t numOfOutput, int32_t order);
int32_t setGroupResultOutputBuf_rv(SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t type,
int16_t bytes, int32_t groupId, SDiskbasedBuf* pBuf, SExecTaskInfo* pTaskInfo, SAggSupporter* pAggSup);
void doDestroyBasicInfo(SOptrBasicInfo* pInfo, int32_t numOfOutput);
int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows,
char* pData, int32_t compLen, int32_t numOfOutput, int64_t startTs,
uint64_t* total, SArray* pColList);
SOperatorInfo* createExchangeOperatorInfo(const SNodeList* pSources, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfCols, int32_t repeatTime,
int32_t reverseTime, SArray* pColMatchInfo, SNode* pCondition, SExecTaskInfo* pTaskInfo);
......@@ -647,12 +669,12 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
SArray* pGroupColList, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList, SArray* pTableIdList, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SInterval* pInterval, SSDataBlock* pResBlock,
int32_t fillType, char* fillVal, bool multigroupResult, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createDistinctOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
int32_t numOfOutput);
SOperatorInfo* createDistinctOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
......
source/libs/executor/src/groupoperator.c 0 → 100644
浏览文件 @ 9317b5f2
/*
* 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 "function.h"
#include "tname.h"
#include "tdatablock.h"
#include "tmsg.h"
#include "executorimpl.h"
#include "tcompare.h"
#include "thash.h"
#include "ttypes.h"
static void destroyGroupbyOperatorInfo(void* param, int32_t numOfOutput) {
SGroupbyOperatorInfo* pInfo = (SGroupbyOperatorInfo*)param;
doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
taosMemoryFreeClear(pInfo->keyBuf);
taosArrayDestroy(pInfo->pGroupCols);
taosArrayDestroy(pInfo->pGroupColVals);
}
static int32_t initGroupOptrInfo(SGroupbyOperatorInfo* pInfo, SArray* pGroupColList) {
pInfo->pGroupColVals = taosArrayInit(4, sizeof(SGroupKeys));
if (pInfo->pGroupColVals == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t numOfGroupCols = taosArrayGetSize(pGroupColList);
for (int32_t i = 0; i < numOfGroupCols; ++i) {
SColumn* pCol = taosArrayGet(pGroupColList, i);
pInfo->groupKeyLen += pCol->bytes;
struct SGroupKeys key = {0};
key.bytes = pCol->bytes;
key.type = pCol->type;
key.isNull = false;
key.pData = taosMemoryCalloc(1, pCol->bytes);
if (key.pData == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
taosArrayPush(pInfo->pGroupColVals, &key);
}
int32_t nullFlagSize = sizeof(int8_t) * numOfGroupCols;
pInfo->keyBuf = taosMemoryCalloc(1, pInfo->groupKeyLen + nullFlagSize);
if (pInfo->keyBuf == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static bool groupKeyCompare(SGroupbyOperatorInfo* pInfo, SSDataBlock* pBlock, int32_t rowIndex,
int32_t numOfGroupCols) {
SColumnDataAgg* pColAgg = NULL;
for (int32_t i = 0; i < numOfGroupCols; ++i) {
SColumn* pCol = taosArrayGet(pInfo->pGroupCols, i);
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pCol->slotId);
if (pBlock->pBlockAgg != NULL) {
pColAgg = &pBlock->pBlockAgg[pCol->slotId]; // TODO is agg data matched?
}
bool isNull = colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg);
SGroupKeys* pkey = taosArrayGet(pInfo->pGroupColVals, i);
if (pkey->isNull && isNull) {
continue;
}
if (isNull || pkey->isNull) {
return false;
}
char* val = colDataGetData(pColInfoData, rowIndex);
if (IS_VAR_DATA_TYPE(pkey->type)) {
int32_t len = varDataLen(val);
if (len == varDataLen(pkey->pData) && memcmp(varDataVal(pkey->pData), varDataVal(val), len) == 0) {
continue;
} else {
return false;
}
} else {
if (memcmp(pkey->pData, val, pkey->bytes) != 0) {
return false;
}
}
}
return true;
}
static void keepGroupKeys(SGroupbyOperatorInfo* pInfo, SSDataBlock* pBlock, int32_t rowIndex, int32_t numOfGroupCols) {
SColumnDataAgg* pColAgg = NULL;
for (int32_t i = 0; i < numOfGroupCols; ++i) {
SColumn* pCol = taosArrayGet(pInfo->pGroupCols, i);
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pCol->slotId);
if (pBlock->pBlockAgg != NULL) {
pColAgg = &pBlock->pBlockAgg[pCol->slotId]; // TODO is agg data matched?
}
SGroupKeys* pkey = taosArrayGet(pInfo->pGroupColVals, i);
if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
pkey->isNull = true;
} else {
char* val = colDataGetData(pColInfoData, rowIndex);
if (IS_VAR_DATA_TYPE(pkey->type)) {
memcpy(pkey->pData, val, varDataTLen(val));
} else {
memcpy(pkey->pData, val, pkey->bytes);
}
}
}
}
static int32_t generatedHashKey(void* pKey, int32_t* length, SArray* pGroupColVals) {
ASSERT(pKey != NULL);
size_t numOfGroupCols = taosArrayGetSize(pGroupColVals);
char* isNull = (char*)pKey;
char* pStart = (char*)pKey + sizeof(int8_t) * numOfGroupCols;
for (int32_t i = 0; i < numOfGroupCols; ++i) {
SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);
if (pkey->isNull) {
isNull[i] = 1;
continue;
}
isNull[i] = 0;
if (IS_VAR_DATA_TYPE(pkey->type)) {
varDataCopy(pStart, pkey->pData);
pStart += varDataTLen(pkey->pData);
ASSERT(varDataTLen(pkey->pData) <= pkey->bytes);
} else {
memcpy(pStart, pkey->pData, pkey->bytes);
pStart += pkey->bytes;
}
}
*length = (pStart - (char*)pKey);
return 0;
}
// assign the group keys or user input constant values if required
static void doAssignGroupKeys(SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t totalRows, int32_t rowIndex) {
for (int32_t i = 0; i < numOfOutput; ++i) {
if (pCtx[i].functionId == -1) {
SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(&pCtx[i]);
SColumnInfoData* pColInfoData = pCtx[i].input.pData[0];
if (!colDataIsNull(pColInfoData, totalRows, rowIndex, NULL)) {
char* dest = GET_ROWCELL_INTERBUF(pEntryInfo);
char* data = colDataGetData(pColInfoData, rowIndex);
// set result exists, todo refactor
memcpy(dest, data, pColInfoData->info.bytes);
pEntryInfo->hasResult = DATA_SET_FLAG;
pEntryInfo->numOfRes = 1;
}
}
}
}
static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SGroupbyOperatorInfo* pInfo = pOperator->info;
SqlFunctionCtx* pCtx = pInfo->binfo.pCtx;
int32_t numOfGroupCols = taosArrayGetSize(pInfo->pGroupCols);
// if (type == TSDB_DATA_TYPE_FLOAT || type == TSDB_DATA_TYPE_DOUBLE) {
// qError("QInfo:0x%"PRIx64" group by not supported on double/float columns, abort", GET_TASKID(pRuntimeEnv));
// return;
// }
int32_t len = 0;
STimeWindow w = TSWINDOW_INITIALIZER;
int32_t num = 0;
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
// Compare with the previous row of this column, and do not set the output buffer again if they are identical.
if (!pInfo->isInit) {
keepGroupKeys(pInfo, pBlock, j, numOfGroupCols);
pInfo->isInit = true;
num++;
continue;
}
bool equal = groupKeyCompare(pInfo, pBlock, j, numOfGroupCols);
if (equal) {
num++;
continue;
}
/*int32_t ret = */ generatedHashKey(pInfo->keyBuf, &len, pInfo->pGroupColVals);
int32_t ret = setGroupResultOutputBuf_rv(&(pInfo->binfo), pOperator->numOfOutput, pInfo->keyBuf, TSDB_DATA_TYPE_VARCHAR, len, 0, pInfo->aggSup.pResultBuf, pTaskInfo, &pInfo->aggSup);
if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code
longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
}
int32_t rowIndex = j - num;
doApplyFunctions(pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->numOfOutput, TSDB_ORDER_ASC);
// assign the group keys or user input constant values if required
doAssignGroupKeys(pCtx, pOperator->numOfOutput, pBlock->info.rows, rowIndex);
keepGroupKeys(pInfo, pBlock, j, numOfGroupCols);
num = 1;
}
if (num > 0) {
/*int32_t ret = */ generatedHashKey(pInfo->keyBuf, &len, pInfo->pGroupColVals);
int32_t ret =
setGroupResultOutputBuf_rv(&(pInfo->binfo), pOperator->numOfOutput, pInfo->keyBuf, TSDB_DATA_TYPE_VARCHAR, len,
0, pInfo->aggSup.pResultBuf, pTaskInfo, &pInfo->aggSup);
if (ret != TSDB_CODE_SUCCESS) {
longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
}
int32_t rowIndex = pBlock->info.rows - num;
doApplyFunctions(pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->numOfOutput, TSDB_ORDER_ASC);
doAssignGroupKeys(pCtx, pOperator->numOfOutput, pBlock->info.rows, rowIndex);
}
}
static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator, bool* newgroup) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
}
SGroupbyOperatorInfo* pInfo = pOperator->info;
if (pOperator->status == OP_RES_TO_RETURN) {
toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pInfo->binfo.pRes, pInfo->binfo.capacity,
pInfo->binfo.rowCellInfoOffset);
if (pInfo->binfo.pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
pOperator->status = OP_EXEC_DONE;
}
return pInfo->binfo.pRes;
}
int32_t order = TSDB_ORDER_ASC;
SOperatorInfo* downstream = pOperator->pDownstream[0];
while (1) {
publishOperatorProfEvent(downstream, QUERY_PROF_BEFORE_OPERATOR_EXEC);
SSDataBlock* pBlock = downstream->getNextFn(downstream, newgroup);
publishOperatorProfEvent(downstream, QUERY_PROF_AFTER_OPERATOR_EXEC);
if (pBlock == NULL) {
break;
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order);
// setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->binfo.pCtx, pOperator->numOfOutput);
doHashGroupbyAgg(pOperator, pBlock);
}
pOperator->status = OP_RES_TO_RETURN;
closeAllResultRows(&pInfo->binfo.resultRowInfo);
finalizeMultiTupleQueryResult(pInfo->binfo.pCtx, pOperator->numOfOutput, pInfo->aggSup.pResultBuf,
&pInfo->binfo.resultRowInfo, pInfo->binfo.rowCellInfoOffset);
// if (!stableQuery) { // finalize include the update of result rows
// finalizeQueryResult(pInfo->binfo.pCtx, pOperator->numOfOutput);
// } else {
// updateNumOfRowsInResultRows(pInfo->binfo.pCtx, pOperator->numOfOutput, &pInfo->binfo.resultRowInfo,
// pInfo->binfo.rowCellInfoOffset);
// }
blockDataEnsureCapacity(pInfo->binfo.pRes, pInfo->binfo.capacity);
initGroupResInfo(&pInfo->groupResInfo, &pInfo->binfo.resultRowInfo);
toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pInfo->binfo.pRes, pInfo->binfo.capacity,
pInfo->binfo.rowCellInfoOffset);
if (pInfo->binfo.pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
pOperator->status = OP_EXEC_DONE;
}
return pInfo->binfo.pRes;
}
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SArray* pGroupColList, SExecTaskInfo* pTaskInfo,
const STableGroupInfo* pTableGroupInfo) {
SGroupbyOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupbyOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) {
goto _error;
}
pInfo->pGroupCols = pGroupColList;
initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, 4096, pResultBlock, pTaskInfo->id.str);
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);
int32_t code = initGroupOptrInfo(pInfo, pGroupColList);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
pOperator->name = "GroupbyAggOperator";
pOperator->blockingOptr = true;
pOperator->status = OP_NOT_OPENED;
// pOperator->operatorType = OP_Groupby;
pOperator->pExpr = pExprInfo;
pOperator->numOfOutput = numOfCols;
pOperator->info = pInfo;
pOperator->_openFn = operatorDummyOpenFn;
pOperator->getNextFn = hashGroupbyAggregate;
pOperator->closeFn = destroyGroupbyOperatorInfo;
code = appendDownstream(pOperator, &downstream, 1);
return pOperator;
_error:
taosMemoryFreeClear(pInfo);
taosMemoryFreeClear(pOperator);
return NULL;
}
\ No newline at end of file
source/libs/executor/src/scanoperator.c 0 → 100644
浏览文件 @ 9317b5f2
此差异已折叠。
source/libs/function/inc/builtinsimpl.h
浏览文件 @ 9317b5f2
......@@ -26,31 +26,34 @@ bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
void functionFinalize(SqlFunctionCtx *pCtx);
bool getCountFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void countFunction(SqlFunctionCtx *pCtx);
int32_t countFunction(SqlFunctionCtx *pCtx);
bool getSumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void sumFunction(SqlFunctionCtx *pCtx);
int32_t sumFunction(SqlFunctionCtx *pCtx);
bool minFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool maxFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getMinmaxFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void minFunction(SqlFunctionCtx* pCtx);
void maxFunction(SqlFunctionCtx *pCtx);
int32_t minFunction(SqlFunctionCtx* pCtx);
int32_t maxFunction(SqlFunctionCtx *pCtx);
bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool stddevFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
void stddevFunction(SqlFunctionCtx* pCtx);
int32_t stddevFunction(SqlFunctionCtx* pCtx);
void stddevFinalize(SqlFunctionCtx* pCtx);
bool getPercentileFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool percentileFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
void percentileFunction(SqlFunctionCtx *pCtx);
int32_t percentileFunction(SqlFunctionCtx *pCtx);
void percentileFinalize(SqlFunctionCtx* pCtx);
bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void firstFunction(SqlFunctionCtx *pCtx);
void lastFunction(SqlFunctionCtx *pCtx);
bool getDiffFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool diffFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo);
int32_t diffFunction(SqlFunctionCtx *pCtx);
void valFunction(SqlFunctionCtx *pCtx);
bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t firstFunction(SqlFunctionCtx *pCtx);
int32_t lastFunction(SqlFunctionCtx *pCtx);
#ifdef __cplusplus
}
......
source/libs/function/inc/taggfunction.h
浏览文件 @ 9317b5f2
......@@ -52,8 +52,6 @@ typedef struct SInterpInfoDetail {
int8_t primaryCol;
} SInterpInfoDetail;
#define GET_ROWCELL_INTERBUF(_c) ((void*) ((char*)(_c) + sizeof(SResultRowEntryInfo)))
typedef struct STwaInfo {
int8_t hasResult; // flag to denote has value
double dOutput;
......
source/libs/function/src/builtins.c
浏览文件 @ 9317b5f2
......@@ -63,74 +63,74 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.finalizeFunc = functionFinalize
},
{
.name = "stddev",
.type = FUNCTION_TYPE_STDDEV,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getStddevFuncEnv,
.initFunc = stddevFunctionSetup,
.processFunc = stddevFunction,
.finalizeFunc = stddevFinalize
},
{
.name = "percentile",
.type = FUNCTION_TYPE_PERCENTILE,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "apercentile",
.type = FUNCTION_TYPE_APERCENTILE,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "top",
.type = FUNCTION_TYPE_TOP,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "bottom",
.type = FUNCTION_TYPE_BOTTOM,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "spread",
.type = FUNCTION_TYPE_SPREAD,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "last_row",
.type = FUNCTION_TYPE_LAST_ROW,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
.name = "stddev",
.type = FUNCTION_TYPE_STDDEV,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getStddevFuncEnv,
.initFunc = stddevFunctionSetup,
.processFunc = stddevFunction,
.finalizeFunc = stddevFinalize
},
{
.name = "percentile",
.type = FUNCTION_TYPE_PERCENTILE,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getPercentileFuncEnv,
.initFunc = percentileFunctionSetup,
.processFunc = percentileFunction,
.finalizeFunc = percentileFinalize
},
{
.name = "apercentile",
.type = FUNCTION_TYPE_APERCENTILE,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "top",
.type = FUNCTION_TYPE_TOP,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "bottom",
.type = FUNCTION_TYPE_BOTTOM,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "spread",
.type = FUNCTION_TYPE_SPREAD,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "last_row",
.type = FUNCTION_TYPE_LAST_ROW,
.classification = FUNC_MGT_AGG_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = maxFunctionSetup,
.processFunc = maxFunction,
.finalizeFunc = functionFinalize
},
{
.name = "first",
......@@ -152,6 +152,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.processFunc = lastFunction,
.finalizeFunc = functionFinalize
},
{
.name = "diff",
.type = FUNCTION_TYPE_DIFF,
.classification = FUNC_MGT_NONSTANDARD_SQL_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = getDiffFuncEnv,
.initFunc = diffFunctionSetup,
.processFunc = diffFunction,
.finalizeFunc = functionFinalize
},
{
.name = "abs",
.type = FUNCTION_TYPE_ABS,
......@@ -377,7 +387,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.type = FUNCTION_TYPE_ROWTS,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = NULL,
.getEnvFunc = getTimePseudoFuncEnv,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
......@@ -459,9 +469,11 @@ const int32_t funcMgtBuiltinsNum = (sizeof(funcMgtBuiltins) / sizeof(SBuiltinFun
int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
switch(pFunc->funcType) {
case FUNCTION_TYPE_WDURATION:
case FUNCTION_TYPE_COUNT:
case FUNCTION_TYPE_COUNT: {
pFunc->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
break;
}
case FUNCTION_TYPE_SUM: {
SColumnNode* pParam = nodesListGetNode(pFunc->pParameterList, 0);
int32_t paraType = pParam->node.resType.type;
......@@ -480,6 +492,8 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
pFunc->node.resType = (SDataType) { .bytes = tDataTypes[resType].bytes, .type = resType };
break;
}
case FUNCTION_TYPE_DIFF:
case FUNCTION_TYPE_FIRST:
case FUNCTION_TYPE_LAST:
case FUNCTION_TYPE_MIN:
......@@ -490,10 +504,11 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
break;
}
case FUNCTION_TYPE_QENDTS:
case FUNCTION_TYPE_ROWTS:
case FUNCTION_TYPE_QSTARTTS:
case FUNCTION_TYPE_WENDTS:
case FUNCTION_TYPE_WSTARTTS: {
case FUNCTION_TYPE_QENDTS:
case FUNCTION_TYPE_WSTARTTS:
case FUNCTION_TYPE_WENDTS:{
pFunc->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_TIMESTAMP};
break;
}
......@@ -508,6 +523,7 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
break;
}
case FUNCTION_TYPE_PERCENTILE:
case FUNCTION_TYPE_STDDEV:
case FUNCTION_TYPE_SIN:
case FUNCTION_TYPE_COS:
......@@ -574,7 +590,6 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
break;
}
case FUNCTION_TYPE_ROWTS:
case FUNCTION_TYPE_TBNAME: {
// todo
break;
......
source/libs/function/src/builtinsimpl.c
浏览文件 @ 9317b5f2
......@@ -65,7 +65,7 @@ bool getCountFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
* count function does need the finalize, if data is missing, the default value, which is 0, is used
* count function does not use the pCtx->interResBuf to keep the intermediate buffer
*/
void countFunction(SqlFunctionCtx *pCtx) {
int32_t countFunction(SqlFunctionCtx *pCtx) {
int32_t numOfElem = 0;
/*
......@@ -111,7 +111,7 @@ void countFunction(SqlFunctionCtx *pCtx) {
} \
} while (0)
void sumFunction(SqlFunctionCtx *pCtx) {
int32_t sumFunction(SqlFunctionCtx *pCtx) {
int32_t numOfElem = 0;
// Only the pre-computing information loaded and actual data does not loaded
......@@ -432,12 +432,12 @@ int32_t doMinMaxHelper(SqlFunctionCtx *pCtx, int32_t isMinFunc) {
return numOfElems;
}
void minFunction(SqlFunctionCtx *pCtx) {
int32_t minFunction(SqlFunctionCtx *pCtx) {
int32_t numOfElems = doMinMaxHelper(pCtx, 1);
SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
}
void maxFunction(SqlFunctionCtx *pCtx) {
int32_t maxFunction(SqlFunctionCtx *pCtx) {
int32_t numOfElems = doMinMaxHelper(pCtx, 0);
SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
}
......@@ -475,12 +475,11 @@ bool stddevFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo)
return true;
}
void stddevFunction(SqlFunctionCtx* pCtx) {
int32_t stddevFunction(SqlFunctionCtx* pCtx) {
int32_t numOfElem = 0;
// Only the pre-computing information loaded and actual data does not loaded
SInputColumnInfoData* pInput = &pCtx->input;
SColumnDataAgg* pAgg = pInput->pColumnDataAgg[0];
int32_t type = pInput->pData[0]->info.type;
SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
......@@ -601,6 +600,7 @@ void stddevFinalize(SqlFunctionCtx* pCtx) {
}
typedef struct SPercentileInfo {
double result;
tMemBucket *pMemBucket;
int32_t stage;
double minval;
......@@ -627,19 +627,24 @@ bool percentileFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultI
return true;
}
void percentileFunction(SqlFunctionCtx *pCtx) {
int32_t percentileFunction(SqlFunctionCtx *pCtx) {
int32_t notNullElems = 0;
#if 0
SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
SPercentileInfo *pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
SInputColumnInfoData* pInput = &pCtx->input;
SColumnDataAgg *pAgg = pInput->pColumnDataAgg[0];
SColumnInfoData *pCol = pInput->pData[0];
int32_t type = pCol->info.type;
SPercentileInfo *pInfo = GET_ROWCELL_INTERBUF(pResInfo);
if (pCtx->currentStage == REPEAT_SCAN && pInfo->stage == 0) {
pInfo->stage += 1;
// all data are null, set it completed
if (pInfo->numOfElems == 0) {
pResInfo->complete = true;
return;
return 0;
} else {
pInfo->pMemBucket = tMemBucketCreate(pCtx->inputBytes, pCtx->inputType, pInfo->minval, pInfo->maxval);
}
......@@ -647,19 +652,17 @@ void percentileFunction(SqlFunctionCtx *pCtx) {
// the first stage, only acquire the min/max value
if (pInfo->stage == 0) {
if (pCtx->preAggVals.isSet) {
if (pCtx->input.colDataAggIsSet) {
double tmin = 0.0, tmax = 0.0;
if (IS_SIGNED_NUMERIC_TYPE(pCtx->inputType)) {
tmin = (double)GET_INT64_VAL(&pCtx->preAggVals.statis.min);
tmax = (double)GET_INT64_VAL(&pCtx->preAggVals.statis.max);
} else if (IS_FLOAT_TYPE(pCtx->inputType)) {
tmin = GET_DOUBLE_VAL(&pCtx->preAggVals.statis.min);
tmax = GET_DOUBLE_VAL(&pCtx->preAggVals.statis.max);
} else if (IS_UNSIGNED_NUMERIC_TYPE(pCtx->inputType)) {
tmin = (double)GET_UINT64_VAL(&pCtx->preAggVals.statis.min);
tmax = (double)GET_UINT64_VAL(&pCtx->preAggVals.statis.max);
} else {
assert(true);
if (IS_SIGNED_NUMERIC_TYPE(type)) {
tmin = (double)GET_INT64_VAL(&pAgg->min);
tmax = (double)GET_INT64_VAL(&pAgg->max);
} else if (IS_FLOAT_TYPE(type)) {
tmin = GET_DOUBLE_VAL(&pAgg->min);
tmax = GET_DOUBLE_VAL(&pAgg->max);
} else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
tmin = (double)GET_UINT64_VAL(&pAgg->min);
tmax = (double)GET_UINT64_VAL(&pAgg->max);
}
if (GET_DOUBLE_VAL(&pInfo->minval) > tmin) {
......@@ -670,17 +673,19 @@ void percentileFunction(SqlFunctionCtx *pCtx) {
SET_DOUBLE_VAL(&pInfo->maxval, tmax);
}
pInfo->numOfElems += (pCtx->size - pCtx->preAggVals.statis.numOfNull);
pInfo->numOfElems += (pInput->numOfRows - pAgg->numOfNull);
} else {
for (int32_t i = 0; i < pCtx->size; ++i) {
char *data = GET_INPUT_DATA(pCtx, i);
if (pCtx->hasNull && isNull(data, pCtx->inputType)) {
// check the valid data one by one
int32_t start = pInput->startRowIndex;
for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
if (colDataIsNull_f(pCol->nullbitmap, i)) {
continue;
}
char *data = colDataGetData(pCol, i);
double v = 0;
GET_TYPED_DATA(v, double, pCtx->inputType, data);
if (v < GET_DOUBLE_VAL(&pInfo->minval)) {
SET_DOUBLE_VAL(&pInfo->minval, v);
}
......@@ -693,24 +698,40 @@ void percentileFunction(SqlFunctionCtx *pCtx) {
}
}
return;
return 0;
}
// the second stage, calculate the true percentile value
for (int32_t i = 0; i < pCtx->size; ++i) {
char *data = GET_INPUT_DATA(pCtx, i);
if (pCtx->hasNull && isNull(data, pCtx->inputType)) {
int32_t start = pInput->startRowIndex;
for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
if (colDataIsNull_f(pCol->nullbitmap, i)) {
continue;
}
char *data = colDataGetData(pCol, i);
notNullElems += 1;
tMemBucketPut(pInfo->pMemBucket, data, 1);
}
SET_VAL(pCtx, notNullElems, 1);
SET_VAL(pResInfo, notNullElems, 1);
pResInfo->hasResult = DATA_SET_FLAG;
#endif
}
// TODO set the correct parameter.
void percentileFinalize(SqlFunctionCtx* pCtx) {
double v = 50;//pCtx->param[0].nType == TSDB_DATA_TYPE_INT ? pCtx->param[0].i64 : pCtx->param[0].dKey;
SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
SPercentileInfo* ppInfo = (SPercentileInfo *) GET_ROWCELL_INTERBUF(pResInfo);
tMemBucket * pMemBucket = ppInfo->pMemBucket;
if (pMemBucket != NULL && pMemBucket->total > 0) { // check for null
SET_DOUBLE_VAL(&ppInfo->result, getPercentile(pMemBucket, v));
}
tMemBucketDestroy(pMemBucket);
functionFinalize(pCtx);
}
bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
......@@ -721,9 +742,9 @@ bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
// TODO fix this
// This ordinary first function only handle the data block in ascending order
void firstFunction(SqlFunctionCtx *pCtx) {
int32_t firstFunction(SqlFunctionCtx *pCtx) {
if (pCtx->order == TSDB_ORDER_DESC) {
return;
return 0;
}
int32_t numOfElems = 0;
......@@ -737,7 +758,7 @@ void firstFunction(SqlFunctionCtx *pCtx) {
// All null data column, return directly.
if (pInput->colDataAggIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
ASSERT(pInputCol->hasNull == true);
return;
return 0;
}
// Check for the first not null data
......@@ -764,9 +785,9 @@ void firstFunction(SqlFunctionCtx *pCtx) {
SET_VAL(pResInfo, numOfElems, 1);
}
void lastFunction(SqlFunctionCtx *pCtx) {
int32_t lastFunction(SqlFunctionCtx *pCtx) {
if (pCtx->order != TSDB_ORDER_DESC) {
return;
return 0;
}
int32_t numOfElems = 0;
......@@ -775,13 +796,12 @@ void lastFunction(SqlFunctionCtx *pCtx) {
char* buf = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0];
// All null data column, return directly.
if (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) {
ASSERT(pInputCol->hasNull == true);
return;
return 0;
}
if (pCtx->order == TSDB_ORDER_DESC) {
......@@ -826,10 +846,242 @@ void lastFunction(SqlFunctionCtx *pCtx) {
SET_VAL(pResInfo, numOfElems, 1);
}
void valFunction(SqlFunctionCtx *pCtx) {
typedef struct SDiffInfo {
bool hasPrev;
bool includeNull;
bool ignoreNegative;
bool firstOutput;
union { int64_t i64; double d64;} prev;
} SDiffInfo;
bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(SDiffInfo);
return true;
}
bool diffFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo) {
if (!functionSetup(pCtx, pResInfo)) {
return false;
}
SDiffInfo* pDiffInfo = GET_ROWCELL_INTERBUF(pResInfo);
pDiffInfo->hasPrev = false;
pDiffInfo->prev.i64 = 0;
pDiffInfo->ignoreNegative = false; // TODO set correct param
pDiffInfo->includeNull = false;
pDiffInfo->firstOutput = false;
return true;
}
int32_t diffFunction(SqlFunctionCtx *pCtx) {
SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
char* buf = GET_ROWCELL_INTERBUF(pResInfo);
SDiffInfo *pDiffInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0];
bool isFirstBlock = (pDiffInfo->hasPrev == false);
int32_t numOfElems = 0;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pCtx->order);
// int32_t i = (pCtx->order == TSDB_ORDER_ASC) ? 0 : pCtx->size - 1;
SColumnInfoData* pTsOutput = pCtx->pTsOutput;
TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
int32_t startOffset = pCtx->offset;
switch (pInputCol->info.type) {
case TSDB_DATA_TYPE_INT: {
SColumnInfoData *pOutput = (SColumnInfoData *)pCtx->pOutput;
for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += step) {
int32_t pos = startOffset + (isFirstBlock? (numOfElems-1):numOfElems);
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
if (pDiffInfo->includeNull) {
colDataSetNull_f(pOutput->nullbitmap, pos);
if (tsList != NULL) {
colDataAppendInt64(pTsOutput, pos, &tsList[i]);
}
numOfElems += 1;
}
continue;
}
int32_t v = *(int32_t*) colDataGetData(pInputCol, i);
if (pDiffInfo->hasPrev) {
int32_t delta = (int32_t)(v - pDiffInfo->prev.i64); // direct previous may be null
if (delta < 0 && pDiffInfo->ignoreNegative) {
colDataSetNull_f(pOutput->nullbitmap, pos);
} else {
colDataAppendInt32(pOutput, pos, &delta);
}
if (tsList != NULL) {
colDataAppendInt64(pTsOutput, pos, &tsList[i]);
}
}
pDiffInfo->prev.i64 = v;
pDiffInfo->hasPrev = true;
numOfElems++;
}
break;
}
case TSDB_DATA_TYPE_BIGINT: {
SColumnInfoData *pOutput = (SColumnInfoData *)pCtx->pOutput;
for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += step) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
int32_t v = 0;
if (pDiffInfo->hasPrev) {
v = *(int64_t*) colDataGetData(pInputCol, i);
int64_t delta = (int64_t)(v - pDiffInfo->prev.i64); // direct previous may be null
if (pDiffInfo->ignoreNegative) {
continue;
}
// *(pOutput++) = delta;
// *pTimestamp = (tsList != NULL)? tsList[i]:0;
//
// pOutput += 1;
// pTimestamp += 1;
}
SColumnInfoData* pInputCol = pCtx->input.pData[0];
memcpy(buf, pInputCol->pData, pInputCol->info.bytes);
pDiffInfo->prev.i64 = v;
pDiffInfo->hasPrev = true;
numOfElems++;
}
break;
}
#if 0
case TSDB_DATA_TYPE_DOUBLE: {
double *pData = (double *)data;
double *pOutput = (double *)pCtx->pOutput;
for (; i < pCtx->size && i >= 0; i += step) {
if (pCtx->hasNull && isNull((const char*) &pData[i], pCtx->inputType)) {
continue;
}
if ((pDiffInfo->ignoreNegative) && (pData[i] < 0)) {
continue;
}
if (pDiffInfo->hasPrev) { // initial value is not set yet
SET_DOUBLE_VAL(pOutput, pData[i] - pDiffInfo->d64Prev); // direct previous may be null
*pTimestamp = (tsList != NULL)? tsList[i]:0;
pOutput += 1;
pTimestamp += 1;
}
pDiffInfo->d64Prev = pData[i];
pDiffInfo->hasPrev = true;
numOfElems++;
}
break;
}
case TSDB_DATA_TYPE_FLOAT: {
float *pData = (float *)data;
float *pOutput = (float *)pCtx->pOutput;
for (; i < pCtx->size && i >= 0; i += step) {
if (pCtx->hasNull && isNull((const char*) &pData[i], pCtx->inputType)) {
continue;
}
if ((pDiffInfo->ignoreNegative) && (pData[i] < 0)) {
continue;
}
if (pDiffInfo->hasPrev) { // initial value is not set yet
*pOutput = (float)(pData[i] - pDiffInfo->d64Prev); // direct previous may be null
*pTimestamp = (tsList != NULL)? tsList[i]:0;
pOutput += 1;
pTimestamp += 1;
}
pDiffInfo->d64Prev = pData[i];
pDiffInfo->hasPrev = true;
numOfElems++;
}
break;
}
case TSDB_DATA_TYPE_SMALLINT: {
int16_t *pData = (int16_t *)data;
int16_t *pOutput = (int16_t *)pCtx->pOutput;
for (; i < pCtx->size && i >= 0; i += step) {
if (pCtx->hasNull && isNull((const char*) &pData[i], pCtx->inputType)) {
continue;
}
if ((pDiffInfo->ignoreNegative) && (pData[i] < 0)) {
continue;
}
if (pDiffInfo->hasPrev) { // initial value is not set yet
*pOutput = (int16_t)(pData[i] - pDiffInfo->i64Prev); // direct previous may be null
*pTimestamp = (tsList != NULL)? tsList[i]:0;
pOutput += 1;
pTimestamp += 1;
}
pDiffInfo->i64Prev = pData[i];
pDiffInfo->hasPrev = true;
numOfElems++;
}
break;
}
case TSDB_DATA_TYPE_TINYINT: {
int8_t *pData = (int8_t *)data;
int8_t *pOutput = (int8_t *)pCtx->pOutput;
for (; i < pCtx->size && i >= 0; i += step) {
if (pCtx->hasNull && isNull((char *)&pData[i], pCtx->inputType)) {
continue;
}
if ((pDiffInfo->ignoreNegative) && (pData[i] < 0)) {
continue;
}
if (pDiffInfo->hasPrev) { // initial value is not set yet
*pOutput = (int8_t)(pData[i] - pDiffInfo->i64Prev); // direct previous may be null
*pTimestamp = (tsList != NULL)? tsList[i]:0;
pOutput += 1;
pTimestamp += 1;
}
pDiffInfo->i64Prev = pData[i];
pDiffInfo->hasPrev = true;
numOfElems++;
}
break;
}
#endif
default:
break;
// qError("error input type");
}
// initial value is not set yet
if (!pDiffInfo->hasPrev || numOfElems <= 0) {
/*
* 1. current block and blocks before are full of null
* 2. current block may be null value
*/
assert(pCtx->hasNull);
} else {
// for (int t = 0; t < pCtx->tagInfo.numOfTagCols; ++t) {
// SqlFunctionCtx* tagCtx = pCtx->tagInfo.pTagCtxList[t];
// if (tagCtx->functionId == TSDB_FUNC_TAG_DUMMY) {
// aAggs[TSDB_FUNC_TAGPRJ].xFunction(tagCtx);
// }
// }
int32_t forwardStep = (isFirstBlock) ? numOfElems - 1 : numOfElems;
return forwardStep;
}
}
source/libs/function/src/functionMgt.c
浏览文件 @ 9317b5f2
......@@ -116,6 +116,11 @@ bool fmIsWindowClauseFunc(int32_t funcId) {
return fmIsAggFunc(funcId) || fmIsWindowPseudoColumnFunc(funcId);
}
bool fmIsNonstandardSQLFunc(int32_t funcId) {
return isSpecificClassifyFunc(funcId, FUNC_MGT_NONSTANDARD_SQL_FUNC);
}
void fmFuncMgtDestroy() {
void* m = gFunMgtService.pFuncNameHashTable;
if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {
......
source/libs/function/src/taggfunction.c
浏览文件 @ 9317b5f2
......@@ -1902,10 +1902,10 @@ static void copyTopBotRes(SqlFunctionCtx *pCtx, int32_t type) {
}
// set the output timestamp of each record.
TSKEY *output = pCtx->ptsOutputBuf;
for (int32_t i = 0; i < len; ++i, output += step) {
*output = tvp[i]->timestamp;
}
// TSKEY *output = pCtx->pTsOutput;
// for (int32_t i = 0; i < len; ++i, output += step) {
// *output = tvp[i]->timestamp;
// }
// set the corresponding tag data for each record
// todo check malloc failure
......@@ -2687,7 +2687,7 @@ static void deriv_function(SqlFunctionCtx *pCtx) {
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pCtx->order);
int32_t i = (pCtx->order == TSDB_ORDER_ASC) ? 0 : pCtx->size - 1;
TSKEY *pTimestamp = pCtx->ptsOutputBuf;
TSKEY *pTimestamp = NULL;//pCtx->pTsOutput;
TSKEY *tsList = GET_TS_LIST(pCtx);
double *pOutput = (double *)pCtx->pOutput;
......@@ -2867,7 +2867,7 @@ static void deriv_function(SqlFunctionCtx *pCtx) {
} else { \
*(type *)(ctx)->pOutput = *(type *)(d) - (*(type *)(&(ctx)->param[1].i)); \
*(type *)(&(ctx)->param[1].i) = *(type *)(d); \
*(int64_t *)(ctx)->ptsOutputBuf = GET_TS_DATA(ctx, index); \
*(int64_t *)(ctx)->pTsOutput = GET_TS_DATA(ctx, index); \
} \
} while (0);
......@@ -2881,7 +2881,7 @@ static void diff_function(SqlFunctionCtx *pCtx) {
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pCtx->order);
int32_t i = (pCtx->order == TSDB_ORDER_ASC) ? 0 : pCtx->size - 1;
TSKEY* pTimestamp = pCtx->ptsOutputBuf;
TSKEY* pTimestamp = NULL;//pCtx->pTsOutput;
TSKEY* tsList = GET_TS_LIST(pCtx);
switch (pCtx->inputType) {
......
source/libs/scalar/src/sclfunc.c
浏览文件 @ 9317b5f2
......@@ -687,6 +687,7 @@ int32_t charLengthFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam
return doLengthFunction(pInput, inputNum, pOutput, tcharlength);
}
#if 0
static void reverseCopy(char* dest, const char* src, int16_t type, int32_t numOfRows) {
switch(type) {
case TSDB_DATA_TYPE_TINYINT:
......@@ -751,6 +752,7 @@ static void reverseCopy(char* dest, const char* src, int16_t type, int32_t numOf
default: assert(0);
}
}
#endif
bool getTimePseudoFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(int64_t);
......
tests/script/tsim/query/session.sim
浏览文件 @ 9317b5f2
......@@ -284,14 +284,15 @@ print ====> select count(*),first(tagtype),last(tagtype),avg(tagtype),sum(tagtyp
# return -1
#endi
sql_error select * from dev_001 session(ts,1w)
sql_error select count(*) from st session(ts,1w)
sql_error select count(*) from dev_001 group by tagtype session(ts,1w)
sql_error select count(*) from dev_001 session(ts,1n)
sql_error select count(*) from dev_001 session(ts,1y)
sql_error select count(*) from dev_001 session(ts,0s)
sql_error select count(*) from dev_001 session(i,1y)
sql_error select count(*) from dev_001 session(ts,1d) where ts <'2020-05-20 0:0:0'
print ================> syntax error check not active ================> reactive
#sql_error select * from dev_001 session(ts,1w)
#sql_error select count(*) from st session(ts,1w)
#sql_error select count(*) from dev_001 group by tagtype session(ts,1w)
#sql_error select count(*) from dev_001 session(ts,1n)
#sql_error select count(*) from dev_001 session(ts,1y)
#sql_error select count(*) from dev_001 session(ts,0s)
#sql_error select count(*) from dev_001 session(i,1y)
#sql_error select count(*) from dev_001 session(ts,1d) where ts <'2020-05-20 0:0:0'
print ====> create database d1 precision 'us'
sql create database d1 precision 'us'
......@@ -299,17 +300,19 @@ sql use d1
sql create table dev_001 (ts timestamp ,i timestamp ,j int)
sql insert into dev_001 values(1623046993681000,now,1)(1623046993681001,now+1s,2)(1623046993681002,now+2s,3)(1623046993681004,now+5s,4)
print ====> select count(*) from dev_001 session(ts,1u)
sql select count(*) from dev_001 session(ts,1u)
sql select _wstartts, count(*) from dev_001 session(ts,1u)
if $rows != 2 then
print expect 2, actual: $rows
return -1
endi
if $data01 != 3 then
return -1
endi
sql_error select count(*) from dev_001 session(i,1s)
sql create table secondts(ts timestamp,t2 timestamp,i int)
sql_error select count(*) from secondts session(t2,2s)
#sql_error select count(*) from dev_001 session(i,1s)
#sql create table secondts(ts timestamp,t2 timestamp,i int)
#sql_error select count(*) from secondts session(t2,2s)
if $loop_test == 0 then
print =============== stop and restart taosd
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或