提交 edd29b13 编写于 作者: H Hongze Cheng

Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/data_format

......@@ -93,6 +93,7 @@ int32_t histogramFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getStateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool stateFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t stateCountFunction(SqlFunctionCtx* pCtx);
int32_t stateDurationFunction(SqlFunctionCtx* pCtx);
bool getSelectivityFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
......
......@@ -263,7 +263,7 @@ static int32_t translateHistogram(SFunctionNode* pFunc, char* pErrBuf, int32_t l
return TSDB_CODE_SUCCESS;
}
static int32_t translateState(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
static int32_t translateStateCount(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
if (3 != LIST_LENGTH(pFunc->pParameterList)) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
}
......@@ -283,6 +283,31 @@ static int32_t translateState(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
return TSDB_CODE_SUCCESS;
}
static int32_t translateStateDuration(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
int32_t paraNum = LIST_LENGTH(pFunc->pParameterList);
if (3 != paraNum && 4 != paraNum) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
}
uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
if (!IS_NUMERIC_TYPE(colType)) {
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
}
if (((SExprNode*)nodesListGetNode(pFunc->pParameterList, 1))->resType.type != TSDB_DATA_TYPE_BINARY ||
(((SExprNode*)nodesListGetNode(pFunc->pParameterList, 2))->resType.type != TSDB_DATA_TYPE_BIGINT &&
((SExprNode*)nodesListGetNode(pFunc->pParameterList, 2))->resType.type != TSDB_DATA_TYPE_DOUBLE)) {
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
}
if (paraNum == 4 && ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 3))->resType.type != TSDB_DATA_TYPE_BIGINT) {
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
}
pFunc->node.resType = (SDataType) { .bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes, .type = TSDB_DATA_TYPE_BIGINT };
return TSDB_CODE_SUCCESS;
}
static int32_t translateLastRow(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
// todo
return TSDB_CODE_SUCCESS;
......@@ -701,12 +726,22 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.name = "state_count",
.type = FUNCTION_TYPE_STATE_COUNT,
.classification = FUNC_MGT_NONSTANDARD_SQL_FUNC,
.translateFunc = translateState,
.translateFunc = translateStateCount,
.getEnvFunc = getStateFuncEnv,
.initFunc = functionSetup,
.processFunc = stateCountFunction,
.finalizeFunc = NULL
},
{
.name = "state_duration",
.type = FUNCTION_TYPE_STATE_DURATION,
.classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
.translateFunc = translateStateDuration,
.getEnvFunc = getStateFuncEnv,
.initFunc = functionSetup,
.processFunc = stateDurationFunction,
.finalizeFunc = NULL
},
{
.name = "abs",
.type = FUNCTION_TYPE_ABS,
......
......@@ -125,7 +125,10 @@ typedef enum {
} EHistoBinType;
typedef struct SStateInfo {
union {
int64_t count;
int64_t durationStart;
};
} SStateInfo;
typedef enum {
......@@ -2844,3 +2847,52 @@ int32_t stateCountFunction(SqlFunctionCtx* pCtx) {
return numOfElems;
}
int32_t stateDurationFunction(SqlFunctionCtx* pCtx) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
SStateInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input;
TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
SColumnInfoData* pInputCol = pInput->pData[0];
SColumnInfoData* pTsOutput = pCtx->pTsOutput;
int32_t numOfElems = 0;
SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
//TODO: process timeUnit for different db precisions
int32_t timeUnit = 1000;
if (pCtx->numOfParams == 5) { //TODO: param number incorrect
timeUnit = pCtx->param[3].param.i;
}
int8_t op = getStateOpType(varDataVal(pCtx->param[1].param.pz));
if (STATE_OPER_INVALID == op) {
return 0;
}
for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
numOfElems++;
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
colDataAppendNULL(pOutput, i);
continue;
}
bool ret = checkStateOp(op, pInputCol, i, pCtx->param[2].param);
int64_t output = -1;
if (ret) {
if (pInfo->durationStart == 0) {
output = 0;
pInfo->durationStart = tsList[i];
} else {
output = (tsList[i] - pInfo->durationStart) / timeUnit;
}
} else {
pInfo->durationStart = 0;
}
colDataAppend(pOutput, i, (char *)&output, false);
}
return numOfElems;
}
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册