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

Merge pull request #23542 from taosdata/fix/TD-27041 

adj builtin function
This commit is contained in:
Haojun Liao 2023-11-07 13:43:23 +08:00 committed by GitHub
parent 3d1afe7adf e6fffe58a5
commit acd20506f6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 47 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -580,6 +580,7 @@ typedef struct SStreamSessionAggOperatorInfo {
bool reCkBlock; bool reCkBlock;
SSDataBlock* pCheckpointRes; SSDataBlock* pCheckpointRes;
bool clearState; bool clearState;
bool recvGetAll;
} SStreamSessionAggOperatorInfo; } SStreamSessionAggOperatorInfo;
typedef struct SStreamStateAggOperatorInfo { typedef struct SStreamStateAggOperatorInfo {
@ -603,6 +604,7 @@ typedef struct SStreamStateAggOperatorInfo {
SArray* historyWins; SArray* historyWins;
bool reCkBlock; bool reCkBlock;
SSDataBlock* pCheckpointRes; SSDataBlock* pCheckpointRes;
bool recvGetAll;
} SStreamStateAggOperatorInfo; } SStreamStateAggOperatorInfo;
typedef struct SStreamPartitionOperatorInfo { typedef struct SStreamPartitionOperatorInfo {

25
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -2535,6 +2535,15 @@ void doStreamSessionSaveCheckpoint(SOperatorInfo* pOperator) {
taosMemoryFree(buf); taosMemoryFree(buf);
} }
static void resetUnCloseSessionWinInfo(SSHashObj* winMap) {
void* pIte = NULL;
int32_t iter = 0;
while ((pIte = tSimpleHashIterate(winMap, pIte, &iter)) != NULL) {
SResultWindowInfo* pResInfo = pIte;
pResInfo->pStatePos->beUsed = true;
}
}
static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) { static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
SExprSupp* pSup = &pOperator->exprSupp; SExprSupp* pSup = &pOperator->exprSupp;
SStreamSessionAggOperatorInfo* pInfo = pOperator->info; SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
@ -2549,6 +2558,12 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
if (opRes) { if (opRes) {
return opRes; return opRes;
} }
if (pInfo->recvGetAll) {
pInfo->recvGetAll = false;
resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
}
setOperatorCompleted(pOperator); setOperatorCompleted(pOperator);
return NULL; return NULL;
} }
@ -2586,6 +2601,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
taosArrayDestroy(pWins); taosArrayDestroy(pWins);
continue; continue;
} else if (pBlock->info.type == STREAM_GET_ALL) { } else if (pBlock->info.type == STREAM_GET_ALL) {
pInfo->recvGetAll = true;
getAllSessionWindow(pAggSup->pResultRows, pInfo->pStUpdated); getAllSessionWindow(pAggSup->pResultRows, pInfo->pStUpdated);
continue; continue;
} else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) { } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
@ -2841,6 +2857,8 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT); pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT);
pInfo->clearState = false; pInfo->clearState = false;
pInfo->recvGetAll = false;
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
// for stream // for stream
void* buff = NULL; void* buff = NULL;
@ -3457,6 +3475,11 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
return resBlock; return resBlock;
} }
if (pInfo->recvGetAll) {
pInfo->recvGetAll = false;
resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
}
setOperatorCompleted(pOperator); setOperatorCompleted(pOperator);
return NULL; return NULL;
} }
@ -3485,6 +3508,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
taosArrayDestroy(pWins); taosArrayDestroy(pWins);
continue; continue;
} else if (pBlock->info.type == STREAM_GET_ALL) { } else if (pBlock->info.type == STREAM_GET_ALL) {
pInfo->recvGetAll = true;
getAllSessionWindow(pInfo->streamAggSup.pResultRows, pInfo->pSeUpdated); getAllSessionWindow(pInfo->streamAggSup.pResultRows, pInfo->pSeUpdated);
continue; continue;
} else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) { } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
@ -3715,6 +3739,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
} }
pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT); pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT);
pInfo->recvGetAll = false;
// for stream // for stream
void* buff = NULL; void* buff = NULL;

40
source/libs/function/src/builtins.c
View File

@ -3312,26 +3312,6 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.sprocessFunc = castFunction, .sprocessFunc = castFunction,
.finalizeFunc = NULL .finalizeFunc = NULL
}, },
{
.name = "to_timestamp",
.type = FUNCTION_TYPE_TO_TIMESTAMP,
.classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_DATETIME_FUNC,
.translateFunc = translateToTimestamp,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = toTimestampFunction,
.finalizeFunc = NULL
},
{
.name = "to_char",
.type = FUNCTION_TYPE_TO_CHAR,
.classification = FUNC_MGT_SCALAR_FUNC,
.translateFunc = translateToChar,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = toCharFunction,
.finalizeFunc = NULL
},
{ {
.name = "to_iso8601", .name = "to_iso8601",
.type = FUNCTION_TYPE_TO_ISO8601, .type = FUNCTION_TYPE_TO_ISO8601,
@ -3709,6 +3689,26 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.sprocessFunc = qVgIdFunction, .sprocessFunc = qVgIdFunction,
.finalizeFunc = NULL .finalizeFunc = NULL
}, },
{
.name = "to_timestamp",
.type = FUNCTION_TYPE_TO_TIMESTAMP,
.classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_DATETIME_FUNC,
.translateFunc = translateToTimestamp,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = toTimestampFunction,
.finalizeFunc = NULL
},
{
.name = "to_char",
.type = FUNCTION_TYPE_TO_CHAR,
.classification = FUNC_MGT_SCALAR_FUNC,
.translateFunc = translateToChar,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = toCharFunction,
.finalizeFunc = NULL
},
}; };
// clang-format on // clang-format on