From e06dce4892560d3b271f6fe4ce6c56fcfebc8bb8 Mon Sep 17 00:00:00 2001 From: dapan1121 <89396746@qq.com> Date: Wed, 7 Apr 2021 18:40:20 +0800 Subject: [PATCH] support stable udf --- src/client/inc/tscUtil.h | 8 ++--- src/client/src/tscLocalMerge.c | 65 ++++++++++++++++++++++++++++------ src/client/src/tscSQLParser.c | 10 +++--- src/client/src/tscServer.c | 2 +- src/client/src/tscSql.c | 6 ++-- src/client/src/tscSubquery.c | 36 ++++++++++++++----- src/client/src/tscUtil.c | 60 +++++++++++++++++++++++-------- src/query/inc/qUdf.h | 11 +++--- src/query/inc/qUtil.h | 2 ++ src/query/src/qExecutor.c | 8 ++++- tests/script/sh/add_one.c | 8 ++--- tests/script/sh/sum_double.c | 16 ++++----- 12 files changed, 165 insertions(+), 67 deletions(-) diff --git a/src/client/inc/tscUtil.h b/src/client/inc/tscUtil.h index 1e151584a1..536fcc7b6b 100644 --- a/src/client/inc/tscUtil.h +++ b/src/client/inc/tscUtil.h @@ -133,13 +133,13 @@ bool tscGroupbyColumn(SQueryInfo* pQueryInfo); bool tscIsTopbotQuery(SQueryInfo* pQueryInfo); int32_t tscGetTopbotQueryParam(SQueryInfo* pQueryInfo); -bool tscNonOrderedProjectionQueryOnSTable(SQueryInfo *pQueryInfo, int32_t tableIndex); -bool tscOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex); -bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex); +bool tscNonOrderedProjectionQueryOnSTable(SSqlCmd *pCmd, SQueryInfo *pQueryInfo, int32_t tableIndex); +bool tscOrderedProjectionQueryOnSTable(SSqlCmd *pCmd, SQueryInfo* pQueryInfo, int32_t tableIndex); +bool tscIsProjectionQueryOnSTable(SSqlCmd *pCmd, SQueryInfo* pQueryInfo, int32_t tableIndex); bool tscIsProjectionQuery(SQueryInfo* pQueryInfo); -bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex); +bool tscIsTwoStageSTableQuery(SSqlCmd *pCmd, SQueryInfo* pQueryInfo, int32_t tableIndex); bool tscQueryTags(SQueryInfo* pQueryInfo); bool tscMultiRoundQuery(SQueryInfo* pQueryInfo, int32_t tableIndex); bool tscQueryBlockInfo(SQueryInfo* pQueryInfo); diff --git a/src/client/src/tscLocalMerge.c b/src/client/src/tscLocalMerge.c index fd8d842871..9511a10f88 100644 --- a/src/client/src/tscLocalMerge.c +++ b/src/client/src/tscLocalMerge.c @@ -300,7 +300,7 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); param->groupOrderType = pQueryInfo->groupbyExpr.orderType; - pReducer->orderPrjOnSTable = tscOrderedProjectionQueryOnSTable(pQueryInfo, 0); + pReducer->orderPrjOnSTable = tscOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0); pRes->code = tLoserTreeCreate(&pReducer->pLoserTree, pReducer->numOfBuffer, param, treeComparator); if (pReducer->pLoserTree == NULL || pRes->code != 0) { @@ -377,7 +377,7 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde pReducer->pDesc->pColumnModel->capacity = 1; // restore the limitation value at the last stage - if (tscOrderedProjectionQueryOnSTable(pQueryInfo, 0)) { + if (tscOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0)) { pQueryInfo->limit.limit = pQueryInfo->clauseLimit; pQueryInfo->limit.offset = pQueryInfo->prjOffset; } @@ -561,7 +561,7 @@ static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SSqlCmd *pCm } // primary timestamp column is involved in final result - if (pQueryInfo->interval.interval != 0 || tscOrderedProjectionQueryOnSTable(pQueryInfo, 0)) { + if (pQueryInfo->interval.interval != 0 || tscOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0)) { numOfGroupByCols++; } @@ -751,6 +751,10 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr if (functionId >= TSDB_FUNC_TS && functionId <= TSDB_FUNC_DIFF) { type = pModel->pFields[i].field.type; bytes = pModel->pFields[i].field.bytes; + } else if (functionId < 0) { + SUdfInfo* pUdfInfo = taosArrayGet(pCmd->pUdfInfo, -1 * functionId - 1); + type = pUdfInfo->resType; + bytes = pUdfInfo->resBytes; } else { if (functionId == TSDB_FUNC_FIRST_DST) { functionId = TSDB_FUNC_FIRST; @@ -1082,6 +1086,10 @@ static void doExecuteFinalMerge(SSqlCmd *pCmd, SLocalMerger *pLocalMerge, bool n } pCtx->currentStage = MERGE_STAGE; + if (functionId < 0) { + continue; + } + if (needInit) { aAggs[pCtx->functionId].init(pCtx, pCtx->resultInfo); } @@ -1093,7 +1101,24 @@ static void doExecuteFinalMerge(SSqlCmd *pCmd, SLocalMerger *pLocalMerge, bool n continue; } - aAggs[functionId].mergeFunc(&pLocalMerge->pCtx[j]); + if (functionId < 0) { + int32_t output = 0; + + SUdfInfo* pUdfInfo = taosArrayGet(pCmd->pUdfInfo, -1 * functionId - 1); + assert (pUdfInfo->funcType == TSDB_UDF_TYPE_AGGREGATE); + + if (pUdfInfo->funcs[TSDB_UDF_FUNC_MERGE]) { + (*(udfMergeFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_MERGE])(pLocalMerge->pCtx[j].pInput, pLocalMerge->pCtx[j].size, pLocalMerge->pCtx[j].pOutput, &output, &pUdfInfo->init); + + // set the output value exist + pLocalMerge->pCtx[j].resultInfo->numOfRes = output; + if (output > 0) { + pLocalMerge->pCtx[j].resultInfo->hasResult = DATA_SET_FLAG; + } + } + } else { + aAggs[functionId].mergeFunc(&pLocalMerge->pCtx[j]); + } } } @@ -1168,12 +1193,30 @@ static void fillMultiRowsOfTagsVal(SQueryInfo *pQueryInfo, int32_t numOfRes, SLo free(buf); } -int32_t finalizeRes(SQueryInfo *pQueryInfo, SLocalMerger *pLocalMerge) { +int32_t finalizeRes(SSqlCmd *pCmd, SQueryInfo *pQueryInfo, SLocalMerger *pLocalMerge) { size_t size = tscSqlExprNumOfExprs(pQueryInfo); for (int32_t k = 0; k < size; ++k) { SQLFunctionCtx* pCtx = &pLocalMerge->pCtx[k]; - aAggs[pCtx->functionId].xFinalize(pCtx); + + if (pCtx->functionId < 0) { + int32_t output = 0; + + SUdfInfo* pUdfInfo = taosArrayGet(pCmd->pUdfInfo, -1 * pCtx->functionId - 1); + assert (pUdfInfo->funcType == TSDB_UDF_TYPE_AGGREGATE); + + if (pUdfInfo && pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE]) { + (*(udfFinalizeFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE])(pCtx->pOutput, &output, &pUdfInfo->init); + + // set the output value exist + pCtx->resultInfo->numOfRes = output; + if (output > 0) { + pCtx->resultInfo->hasResult = DATA_SET_FLAG; + } + } + } else { + aAggs[pCtx->functionId].xFinalize(pCtx); + } } pLocalMerge->hasPrevRow = false; @@ -1192,13 +1235,13 @@ int32_t finalizeRes(SQueryInfo *pQueryInfo, SLocalMerger *pLocalMerge) { * results generated by simple aggregation function, we merge them all into one points * *Exception*: column projection query, required no merge procedure */ -bool needToMerge(SQueryInfo *pQueryInfo, SLocalMerger *pLocalMerge, tFilePage *tmpBuffer) { +bool needToMerge(SSqlCmd *pCmd, SQueryInfo *pQueryInfo, SLocalMerger *pLocalMerge, tFilePage *tmpBuffer) { int32_t ret = 0; // merge all result by default int16_t functionId = pLocalMerge->pCtx[0].functionId; // todo opt performance - if ((/*functionId == TSDB_FUNC_PRJ || */functionId == TSDB_FUNC_ARITHM) || (tscIsProjectionQueryOnSTable(pQueryInfo, 0) && pQueryInfo->distinctTag == false)) { // column projection query + if ((/*functionId == TSDB_FUNC_PRJ || */functionId == TSDB_FUNC_ARITHM) || (tscIsProjectionQueryOnSTable(pCmd, pQueryInfo, 0) && pQueryInfo->distinctTag == false)) { // column projection query ret = 1; // disable merge procedure } else { tOrderDescriptor *pDesc = pLocalMerge->pDesc; @@ -1522,7 +1565,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) { } if (pLocalMerge->hasPrevRow) { - if (needToMerge(pQueryInfo, pLocalMerge, tmpBuffer)) { + if (needToMerge(pCmd, pQueryInfo, pLocalMerge, tmpBuffer)) { // belong to the group of the previous row, continue process it doExecuteFinalMerge(pCmd, pLocalMerge, false); @@ -1533,7 +1576,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) { * current row does not belong to the group of previous row. * so the processing of previous group is completed. */ - int32_t numOfRes = finalizeRes(pQueryInfo, pLocalMerge); + int32_t numOfRes = finalizeRes(pCmd, pQueryInfo, pLocalMerge); bool sameGroup = isSameGroup(pCmd, pLocalMerge, pLocalMerge->prevRowOfInput, tmpBuffer); tFilePage *pResBuf = pLocalMerge->pResultBuf; @@ -1605,7 +1648,7 @@ int32_t tscDoLocalMerge(SSqlObj *pSql) { } if (pLocalMerge->hasPrevRow) { - finalizeRes(pQueryInfo, pLocalMerge); + finalizeRes(pCmd, pQueryInfo, pLocalMerge); } if (pLocalMerge->pResultBuf->num) { diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index 6cdd186c28..7ff2e83251 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -5193,7 +5193,7 @@ int32_t parseOrderbyClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySqlNode* pQueryInfo->order.orderColId = PRIMARYKEY_TIMESTAMP_COL_INDEX; // orderby ts query on super table - if (tscOrderedProjectionQueryOnSTable(pQueryInfo, 0)) { + if (tscOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0)) { addPrimaryTsColIntoResult(pQueryInfo); } } @@ -5817,13 +5817,13 @@ int32_t parseLimitClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t clauseIn // todo refactor if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { if (!tscQueryTags(pQueryInfo)) { // local handle the super table tag query - if (tscIsProjectionQueryOnSTable(pQueryInfo, 0)) { + if (tscIsProjectionQueryOnSTable(pCmd, pQueryInfo, 0)) { if (pQueryInfo->slimit.limit > 0 || pQueryInfo->slimit.offset > 0) { return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2); } // for projection query on super table, all queries are subqueries - if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && + if (tscNonOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0) && !TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_QUERY)) { pQueryInfo->type |= TSDB_QUERY_TYPE_SUBQUERY; } @@ -5859,7 +5859,7 @@ int32_t parseLimitClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t clauseIn pQueryInfo->prjOffset = pQueryInfo->limit.offset; pQueryInfo->vgroupLimit = -1; - if (tscOrderedProjectionQueryOnSTable(pQueryInfo, 0)) { + if (tscOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0)) { /* * the offset value should be removed during retrieve data from virtual node, since the * global order are done in client side, so the offset is applied at the client side @@ -6462,7 +6462,7 @@ int32_t doFunctionsCompatibleCheck(SSqlCmd* pCmd, SQueryInfo* pQueryInfo) { } // projection query on super table does not compatible with "group by" syntax - if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0)) { + if (tscNonOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0)) { return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3); } diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index b94690f7fc..9ebd22d842 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -2485,7 +2485,7 @@ int tscProcessRetrieveRspFromNode(SSqlObj *pSql) { tscSetResRawPtr(pRes, pQueryInfo); } else if ((UTIL_TABLE_IS_CHILD_TABLE(pTableMetaInfo) || UTIL_TABLE_IS_NORMAL_TABLE(pTableMetaInfo)) && !TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_SUBQUERY)) { tscSetResRawPtr(pRes, pQueryInfo); - } else if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && !TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_QUERY) && !TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_SEC_STAGE)) { + } else if (tscNonOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0) && !TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_QUERY) && !TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_SEC_STAGE)) { tscSetResRawPtr(pRes, pQueryInfo); } diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 93d0e9fd09..22ae192db1 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -560,7 +560,7 @@ static bool tscKillQueryInDnode(SSqlObj* pSql) { SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); - if ((pQueryInfo == NULL) || tscIsTwoStageSTableQuery(pQueryInfo, 0)) { + if ((pQueryInfo == NULL) || tscIsTwoStageSTableQuery(pCmd, pQueryInfo, 0)) { return true; } @@ -673,7 +673,7 @@ static void tscKillSTableQuery(SSqlObj *pSql) { SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); - if (!tscIsTwoStageSTableQuery(pQueryInfo, 0)) { + if (!tscIsTwoStageSTableQuery(pCmd, pQueryInfo, 0)) { return; } @@ -724,7 +724,7 @@ void taos_stop_query(TAOS_RES *res) { SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); - if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { + if (tscIsTwoStageSTableQuery(pCmd, pQueryInfo, 0)) { assert(pSql->rpcRid <= 0); tscKillSTableQuery(pSql); } else { diff --git a/src/client/src/tscSubquery.c b/src/client/src/tscSubquery.c index 299cf03805..2fd0b5e8bd 100644 --- a/src/client/src/tscSubquery.c +++ b/src/client/src/tscSubquery.c @@ -23,6 +23,7 @@ #include "tscSubquery.h" #include "tschemautil.h" #include "tsclient.h" +#include "qUdf.h" #include "qUtil.h" typedef struct SInsertSupporter { @@ -653,7 +654,7 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) { if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { assert(pTableMetaInfo->pVgroupTables != NULL); - if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0)) { + if (tscNonOrderedProjectionQueryOnSTable(&pNew->cmd, pQueryInfo, 0)) { SArray* p = buildVgroupTableByResult(pQueryInfo, pTableMetaInfo->pVgroupTables); tscFreeVgroupTableInfo(pTableMetaInfo->pVgroupTables); pTableMetaInfo->pVgroupTables = p; @@ -1442,7 +1443,7 @@ static void joinRetrieveFinalResCallback(void* param, TAOS_RES* tres, int numOfR } SSubqueryState* pState = &pParentSql->subState; - if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && numOfRows == 0) { + if (tscNonOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0) && numOfRows == 0) { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); assert(pQueryInfo->numOfTables == 1); @@ -1479,7 +1480,7 @@ static void joinRetrieveFinalResCallback(void* param, TAOS_RES* tres, int numOfR } // update the records for each subquery in parent sql object. - bool stableQuery = tscIsTwoStageSTableQuery(pQueryInfo, 0); + bool stableQuery = tscIsTwoStageSTableQuery(pCmd, pQueryInfo, 0); for (int32_t i = 0; i < pState->numOfSub; ++i) { if (pParentSql->pSubs[i] == NULL) { tscDebug("%p %p sub:%d not retrieve data", pParentSql, NULL, i); @@ -1576,7 +1577,7 @@ void tscFetchDatablockForSubquery(SSqlObj* pSql) { } SQueryInfo* p = tscGetQueryInfoDetail(&pSub->cmd, 0); - orderedPrjQuery = tscNonOrderedProjectionQueryOnSTable(p, 0); + orderedPrjQuery = tscNonOrderedProjectionQueryOnSTable(&pSub->cmd, p, 0); if (orderedPrjQuery) { break; } @@ -1601,7 +1602,7 @@ void tscFetchDatablockForSubquery(SSqlObj* pSql) { SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSub->cmd, 0); - if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && pSub->res.row >= pSub->res.numOfRows && + if (tscNonOrderedProjectionQueryOnSTable(&pSub->cmd, pQueryInfo, 0) && pSub->res.row >= pSub->res.numOfRows && pSub->res.completed) { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); assert(pQueryInfo->numOfTables == 1); @@ -1804,7 +1805,7 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) { } // In case of consequence query from other vnode, do not wait for other query response here. - if (!(pTableMetaInfo->vgroupIndex > 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0))) { + if (!(pTableMetaInfo->vgroupIndex > 0 && tscNonOrderedProjectionQueryOnSTable(&pSql->cmd, pQueryInfo, 0))) { if (!subAndCheckDone(pSql, pParentSql, pSupporter->subqueryIndex)) { return; } @@ -1816,7 +1817,7 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) { * if the query is a continue query (vgroupIndex > 0 for projection query) for next vnode, do the retrieval of * data instead of returning to its invoker */ - if (pTableMetaInfo->vgroupIndex > 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0)) { + if (pTableMetaInfo->vgroupIndex > 0 && tscNonOrderedProjectionQueryOnSTable(&pSql->cmd, pQueryInfo, 0)) { pSql->fp = joinRetrieveFinalResCallback; // continue retrieve data pSql->cmd.command = TSDB_SQL_FETCH; @@ -2791,6 +2792,22 @@ static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* p tscFreeRetrieveSup(pSql); + size_t size = tscSqlExprNumOfExprs(pQueryInfo); + for (int32_t j = 0; j < size; ++j) { + SQLFunctionCtx *pCtx = &pParentSql->res.pLocalMerger->pCtx[j]; + + int32_t functionId = pCtx->functionId; + + if (functionId < 0) { + SUdfInfo* pUdfInfo = taosArrayGet(pParentSql->cmd.pUdfInfo, -1 * functionId - 1); + int32_t code = initUdfInfo(pUdfInfo); + if (code != TSDB_CODE_SUCCESS) { + pParentSql->res.code = code; + tscAsyncResultOnError(pParentSql); + } + } + } + // set the command flag must be after the semaphore been correctly set. pParentSql->cmd.command = TSDB_SQL_RETRIEVE_LOCALMERGE; if (pParentSql->res.code == TSDB_CODE_SUCCESS) { @@ -2864,7 +2881,8 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR tscDebug("%p sub:%p retrieve numOfRows:%d totalNumOfRows:%" PRIu64 " from ep:%s, orderOfSub:%d", pParentSql, pSql, pRes->numOfRows, pState->numOfRetrievedRows, pSql->epSet.fqdn[pSql->epSet.inUse], idx); - if (num > tsMaxNumOfOrderedResults && tscIsProjectionQueryOnSTable(pQueryInfo, 0)) { + SSqlCmd* pCmd = &pSql->cmd; + if (num > tsMaxNumOfOrderedResults && tscIsProjectionQueryOnSTable(pCmd, pQueryInfo, 0)) { tscError("%p sub:%p num of OrderedRes is too many, max allowed:%" PRId32 " , current:%" PRId64, pParentSql, pSql, tsMaxNumOfOrderedResults, num); tscAbortFurtherRetryRetrieval(trsupport, tres, TSDB_CODE_TSC_SORTED_RES_TOO_MANY); @@ -3430,7 +3448,7 @@ static UNUSED_FUNC bool tscHasRemainDataInSubqueryResultSet(SSqlObj *pSql) { SSqlCmd *pCmd = &pSql->cmd; SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); - if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0)) { + if (tscNonOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0)) { bool allSubqueryExhausted = true; for (int32_t i = 0; i < pSql->subState.numOfSub; ++i) { diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index f5d616fef1..58f4cb3fef 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -113,7 +113,7 @@ bool tscQueryBlockInfo(SQueryInfo* pQueryInfo) { return false; } -bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) { +bool tscIsTwoStageSTableQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t tableIndex) { if (pQueryInfo == NULL) { return false; } @@ -128,7 +128,7 @@ bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) { } // for ordered projection query, iterate all qualified vnodes sequentially - if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, tableIndex)) { + if (tscNonOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, tableIndex)) { return false; } @@ -139,7 +139,7 @@ bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) { return false; } -bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) { +bool tscIsProjectionQueryOnSTable(SSqlCmd *pCmd, SQueryInfo* pQueryInfo, int32_t tableIndex) { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, tableIndex); /* @@ -156,6 +156,15 @@ bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) { for (int32_t i = 0; i < numOfExprs; ++i) { int32_t functionId = tscSqlExprGet(pQueryInfo, i)->functionId; + if (functionId < 0) { + SUdfInfo* pUdfInfo = taosArrayGet(pCmd->pUdfInfo, -1 * functionId - 1); + if (pUdfInfo->funcType == TSDB_UDF_TYPE_AGGREGATE) { + return false; + } + + continue; + } + if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TAGPRJ && functionId != TSDB_FUNC_TAG && @@ -171,8 +180,8 @@ bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) { } // not order by timestamp projection query on super table -bool tscNonOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) { - if (!tscIsProjectionQueryOnSTable(pQueryInfo, tableIndex)) { +bool tscNonOrderedProjectionQueryOnSTable(SSqlCmd *pCmd, SQueryInfo* pQueryInfo, int32_t tableIndex) { + if (!tscIsProjectionQueryOnSTable(pCmd, pQueryInfo, tableIndex)) { return false; } @@ -180,8 +189,8 @@ bool tscNonOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableI return pQueryInfo->order.orderColId < 0; } -bool tscOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) { - if (!tscIsProjectionQueryOnSTable(pQueryInfo, tableIndex)) { +bool tscOrderedProjectionQueryOnSTable(SSqlCmd *pCmd, SQueryInfo* pQueryInfo, int32_t tableIndex) { + if (!tscIsProjectionQueryOnSTable(pCmd, pQueryInfo, tableIndex)) { return false; } @@ -659,6 +668,30 @@ void* tscDestroyBlockArrayList(SArray* pDataBlockList) { return NULL; } + +void freeUdfInfo(SUdfInfo* pUdfInfo) { + if (pUdfInfo == NULL) { + return; + } + + if (pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY]) { + (*(udfDestroyFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY])(&pUdfInfo->init); + } + + tfree(pUdfInfo->name); + + if (pUdfInfo->path) { + unlink(pUdfInfo->path); + } + + tfree(pUdfInfo->path); + + tfree(pUdfInfo->content); + + taosCloseDll(pUdfInfo->handle); +} + + void* tscDestroyUdfArrayList(SArray* pUdfList) { if (pUdfList == NULL) { return NULL; @@ -667,10 +700,7 @@ void* tscDestroyUdfArrayList(SArray* pUdfList) { size_t size = taosArrayGetSize(pUdfList); for (int32_t i = 0; i < size; i++) { SUdfInfo* udf = taosArrayGet(pUdfList, i); - if (udf) { - tfree(udf->content); - tfree(udf->name); - } + freeUdfInfo(udf); } taosArrayDestroy(pUdfList); @@ -2440,7 +2470,7 @@ void tscDoQuery(SSqlObj* pSql) { if (!TSDB_QUERY_HAS_TYPE(type, TSDB_QUERY_TYPE_JOIN_SEC_STAGE)) { tscProcessSql(pSql); } else { // secondary stage join query. - if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { // super table query + if (tscIsTwoStageSTableQuery(pCmd, pQueryInfo, 0)) { // super table query tscLockByThread(&pSql->squeryLock); tscHandleMasterSTableQuery(pSql); tscUnlockByThread(&pSql->squeryLock); @@ -2454,7 +2484,7 @@ void tscDoQuery(SSqlObj* pSql) { } else if (tscMultiRoundQuery(pQueryInfo, 0) && pQueryInfo->round == 0) { tscHandleFirstRoundStableQuery(pSql); // todo lock? return; - } else if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { // super table query + } else if (tscIsTwoStageSTableQuery(pCmd, pQueryInfo, 0)) { // super table query tscLockByThread(&pSql->squeryLock); tscHandleMasterSTableQuery(pSql); tscUnlockByThread(&pSql->squeryLock); @@ -2619,7 +2649,7 @@ bool hasMoreVnodesToTry(SSqlObj* pSql) { numOfVgroups = (int32_t)taosArrayGetSize(pTableMetaInfo->pVgroupTables); } - return tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && + return tscNonOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0) && (!tscHasReachLimitation(pQueryInfo, pRes)) && (pTableMetaInfo->vgroupIndex < numOfVgroups - 1); } @@ -2637,7 +2667,7 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) { * no result returned from the current virtual node anymore, try the next vnode if exists * if case of: multi-vnode super table projection query */ - assert(pRes->numOfRows == 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && !tscHasReachLimitation(pQueryInfo, pRes)); + assert(pRes->numOfRows == 0 && tscNonOrderedProjectionQueryOnSTable(pCmd, pQueryInfo, 0) && !tscHasReachLimitation(pQueryInfo, pRes)); STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); int32_t totalVgroups = pTableMetaInfo->vgroupList->numOfVgroups; diff --git a/src/query/inc/qUdf.h b/src/query/inc/qUdf.h index 7a1facebb1..d141e5f8f3 100644 --- a/src/query/inc/qUdf.h +++ b/src/query/inc/qUdf.h @@ -16,7 +16,7 @@ #ifndef TDENGINE_QUDF_H #define TDENGINE_QUDF_H -enum { TSDB_UDF_FUNC_NORMAL = 0, TSDB_UDF_FUNC_INIT, TSDB_UDF_FUNC_FINALIZE, TSDB_UDF_FUNC_DESTROY, TSDB_UDF_FUNC_MAX_NUM }; +enum { TSDB_UDF_FUNC_NORMAL = 0, TSDB_UDF_FUNC_INIT, TSDB_UDF_FUNC_FINALIZE, TSDB_UDF_FUNC_MERGE, TSDB_UDF_FUNC_DESTROY, TSDB_UDF_FUNC_MAX_NUM }; @@ -43,16 +43,15 @@ typedef struct SUdfInfo { void *handle; // handle loaded in mem void *funcs[TSDB_UDF_FUNC_MAX_NUM]; // function ptr SUdfInit init; - union { // file path or [in memory] binary content - char *content; - char *path; - }; + char *content; + char *path; } SUdfInfo; typedef void (*udfNormalFunc)(char* data, int16_t itype, int16_t iBytes, int32_t numOfRows, int64_t* ts, char* dataOutput, char* tsOutput, - int32_t* numOfOutput, int16_t oType, int16_t oByte, SUdfInit* buf); + int32_t* numOfOutput, int16_t oType, int16_t oBytes, SUdfInit* buf); typedef int32_t (*udfInitFunc)(SUdfInit* data); typedef void (*udfFinalizeFunc)(char* dataOutput, int32_t* numOfOutput, SUdfInit* buf); +typedef void (*udfMergeFunc)(char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput, SUdfInit* buf); typedef void (*udfDestroyFunc)(SUdfInit* buf); diff --git a/src/query/inc/qUtil.h b/src/query/inc/qUtil.h index cdd8b0707a..87f2458658 100644 --- a/src/query/inc/qUtil.h +++ b/src/query/inc/qUtil.h @@ -90,4 +90,6 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo); int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, SQueryRuntimeEnv *pRuntimeEnv, int32_t* offset); +int32_t initUdfInfo(SUdfInfo* pUdfInfo); + #endif // TDENGINE_QUERYUTIL_H diff --git a/src/query/src/qExecutor.c b/src/query/src/qExecutor.c index 005cf5ac4b..f11a0772c8 100644 --- a/src/query/src/qExecutor.c +++ b/src/query/src/qExecutor.c @@ -6008,6 +6008,8 @@ void destroyUdfInfo(SUdfInfo* pUdfInfo) { } tfree(pUdfInfo->path); + + tfree(pUdfInfo->content); taosCloseDll(pUdfInfo->handle); @@ -6027,6 +6029,9 @@ static char* getUdfFuncName(char* name, int type) { case TSDB_UDF_FUNC_FINALIZE: sprintf(funcname, "%s_finalize", name); break; + case TSDB_UDF_FUNC_MERGE: + sprintf(funcname, "%s_merge", name); + break; case TSDB_UDF_FUNC_DESTROY: sprintf(funcname, "%s_destroy", name); break; @@ -6038,7 +6043,7 @@ static char* getUdfFuncName(char* name, int type) { return funcname; } -static int32_t initUdfInfo(SUdfInfo* pUdfInfo) { +int32_t initUdfInfo(SUdfInfo* pUdfInfo) { if (pUdfInfo == NULL) { return TSDB_CODE_SUCCESS; } @@ -6088,6 +6093,7 @@ static int32_t initUdfInfo(SUdfInfo* pUdfInfo) { if (pUdfInfo->funcType == TSDB_UDF_TYPE_AGGREGATE) { pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(pUdfInfo->name, TSDB_UDF_FUNC_FINALIZE)); + pUdfInfo->funcs[TSDB_UDF_FUNC_MERGE] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(pUdfInfo->name, TSDB_UDF_FUNC_MERGE)); } pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(pUdfInfo->name, TSDB_UDF_FUNC_DESTROY)); diff --git a/tests/script/sh/add_one.c b/tests/script/sh/add_one.c index 4c1cb28a90..ec26f1dbd7 100644 --- a/tests/script/sh/add_one.c +++ b/tests/script/sh/add_one.c @@ -10,12 +10,12 @@ typedef struct SUdfInit{ int const_item; /* 0 if result is independent of arguments */ } SUdfInit; -void add_one(char* data, char type, int numOfRows, long long* ts, char* dataOutput, char* tsOutput, - int* numOfOutput, SUdfInit* buf) { +void add_one(char* data, short itype, short ibytes, int numOfRows, long long* ts, char* dataOutput, char* tsOutput, + int* numOfOutput, short otype, short obytes, SUdfInit* buf) { int i; int r = 0; - printf("add_one input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, type, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf); - if (type == 4) { + printf("add_one input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf); + if (itype == 4) { for(i=0;iptr)=*(int*)dataOutput*2; *(int*)dataOutput=*(int*)(buf->ptr); printf("sum_double finalize, dataoutput:%d, numOfOutput:%d\n", *(int *)dataOutput, *numOfOutput); } -void sum_double_merge(char* dataOutput, int* numOfOutput, SUdfInit* buf) { +void sum_double_merge(char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput, SUdfInit* buf) { int r = 0; int sum = 0; - printf("sum_double_merge dataoutput:%p, numOfOutput:%p, buf:%p\n", dataOutput, numOfOutput, buf); - for (int i = 0; i < *numOfOutput; ++i) { - printf("sum_double_merge %d - %d\n", i, *((int*)dataOutput + i)); - sum +=*((int*)dataOutput + i); + printf("sum_double_merge dataoutput:%p, numOfOutput:%d, buf:%p\n", dataOutput, *numOfOutput, buf); + for (int i = 0; i < numOfRows; ++i) { + printf("sum_double_merge %d - %d\n", i, *((int*)data + i)); + sum +=*((int*)data + i); } - *(int*)dataOutput=sum; + *(int*)dataOutput+=sum; *numOfOutput=1; - printf("sum_double sum_double_merge, dataoutput:%d, numOfOutput:%d\n", *(int *)dataOutput, *numOfOutput); + printf("sum_double_merge, dataoutput:%d, numOfOutput:%d\n", *(int *)dataOutput, *numOfOutput); } -- GitLab