From 55e3352d7be4c3739bc6355f41c143bef0c37d33 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Sun, 19 Jul 2020 23:29:17 +0800 Subject: [PATCH] [td-225] refactor codes. opt resbuf perf. rename sources file --- src/client/inc/tsclient.h | 2 +- src/client/src/tscFunctionImpl.c | 8 +- src/client/src/tscServer.c | 69 ++---- src/query/inc/{qExecutor.h => qexecutor.h} | 6 +- src/query/inc/{qresultBuf.h => qresultbuf.h} | 24 +- src/query/inc/{qUtil.h => qutil.h} | 4 +- src/query/inc/tsqlfunction.h | 10 +- src/query/src/{qExecutor.c => qexecutor.c} | 132 +++++++---- .../src/{qFilterFunc.c => qfilterfunc.c} | 4 +- src/query/src/qresultBuf.c | 184 --------------- src/query/src/qresultbuf.c | 218 ++++++++++++++++++ src/query/src/{qUtil.c => qutil.c} | 8 +- src/query/tests/resultBufferTest.cpp | 2 +- src/tsdb/src/tsdbRead.c | 2 +- src/util/src/tutil.c | 2 +- 15 files changed, 368 insertions(+), 307 deletions(-) rename src/query/inc/{qExecutor.h => qexecutor.h} (99%) rename src/query/inc/{qresultBuf.h => qresultbuf.h} (75%) rename src/query/inc/{qUtil.h => qutil.h} (95%) rename src/query/src/{qExecutor.c => qexecutor.c} (98%) rename src/query/src/{qFilterFunc.c => qfilterfunc.c} (99%) delete mode 100644 src/query/src/qresultBuf.c create mode 100644 src/query/src/qresultbuf.c rename src/query/src/{qUtil.c => qutil.c} (98%) diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index ef8184990d..3f548c18ca 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -30,7 +30,7 @@ extern "C" { #include "tsqlfunction.h" #include "tutil.h" -#include "qExecutor.h" +#include "qexecutor.h" #include "qsqlparser.h" #include "qtsbuf.h" #include "tcmdtype.h" diff --git a/src/client/src/tscFunctionImpl.c b/src/client/src/tscFunctionImpl.c index 262b7ab3f6..fb15e4efd5 100644 --- a/src/client/src/tscFunctionImpl.c +++ b/src/client/src/tscFunctionImpl.c @@ -74,7 +74,7 @@ for (int32_t i = 0; i < (ctx)->tagInfo.numOfTagCols; ++i) { \ void noop1(SQLFunctionCtx *UNUSED_PARAM(pCtx)) {} void noop2(SQLFunctionCtx *UNUSED_PARAM(pCtx), int32_t UNUSED_PARAM(index)) {} -void doFinalizer(SQLFunctionCtx *pCtx) { resetResultInfo(GET_RES_INFO(pCtx)); } +void doFinalizer(SQLFunctionCtx *pCtx) { RESET_RESULT_INFO(GET_RES_INFO(pCtx)); } typedef struct tValuePair { tVariant v; @@ -330,12 +330,6 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI return TSDB_CODE_SUCCESS; } -/** - * the numOfRes should be kept, since it may be used later - * and allow the ResultInfo to be re initialized - */ -void resetResultInfo(SResultInfo *pResInfo) { pResInfo->initialized = false; } - void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf) { assert(pResInfo->interResultBuf == NULL); diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 398dea9d09..2bc41a834d 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -242,52 +242,31 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { if (pEpSet) tscMgmtEpSet = *pEpSet; } - if (rpcMsg->pCont == NULL) { - rpcMsg->code = TSDB_CODE_RPC_NETWORK_UNAVAIL; - } else { - STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0); - // if (rpcMsg->code != TSDB_CODE_RPC_NETWORK_UNAVAIL) { - // if (pCmd->command == TSDB_SQL_CONNECT) { - // rpcMsg->code = TSDB_CODE_RPC_NETWORK_UNAVAIL; - // rpcFreeCont(rpcMsg->pCont); - // return; - // } - - // if (pCmd->command == TSDB_SQL_HB) { - // rpcMsg->code = TSDB_CODE_RPC_NOT_READY; - // rpcFreeCont(rpcMsg->pCont); - // return; - // } - - // if (pCmd->command == TSDB_SQL_META || pCmd->command == TSDB_SQL_DESCRIBE_TABLE || - // pCmd->command == TSDB_SQL_STABLEVGROUP || pCmd->command == TSDB_SQL_SHOW || - // pCmd->command == TSDB_SQL_RETRIEVE) { - // // get table meta/vgroup query will not retry, do nothing - // } - // } - - if ((pCmd->command == TSDB_SQL_SELECT || pCmd->command == TSDB_SQL_FETCH || pCmd->command == TSDB_SQL_INSERT || - pCmd->command == TSDB_SQL_UPDATE_TAGS_VAL) && - (rpcMsg->code == TSDB_CODE_TDB_INVALID_TABLE_ID || rpcMsg->code == TSDB_CODE_VND_INVALID_VGROUP_ID || - rpcMsg->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE)) { - tscWarn("%p it shall renew table meta, code:%s, retry:%d", pSql, tstrerror(rpcMsg->code), ++pSql->retry); - // set the flag to denote that sql string needs to be re-parsed and build submit block with table schema - if (rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE) { - pSql->cmd.submitSchema = 1; - } + STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0); - pSql->res.code = rpcMsg->code; // keep the previous error code - if (pSql->retry > pSql->maxRetry) { - tscError("%p max retry %d reached, give up", pSql, pSql->maxRetry); - } else { - rpcMsg->code = tscRenewTableMeta(pSql, pTableMetaInfo->name); - - // if there is an error occurring, proceed to the following error handling procedure. - // todo add test cases - if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { - rpcFreeCont(rpcMsg->pCont); - return; - } + int32_t cmd = pCmd->command; + if ((cmd == TSDB_SQL_SELECT || cmd == TSDB_SQL_FETCH || cmd == TSDB_SQL_INSERT || cmd == TSDB_SQL_UPDATE_TAGS_VAL) && + (rpcMsg->code == TSDB_CODE_TDB_INVALID_TABLE_ID || + rpcMsg->code == TSDB_CODE_VND_INVALID_VGROUP_ID || + rpcMsg->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || + rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE)) { + tscWarn("%p it shall renew table meta, code:%s, retry:%d", pSql, tstrerror(rpcMsg->code), ++pSql->retry); + + // set the flag to denote that sql string needs to be re-parsed and build submit block with table schema + if (rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE) { + pSql->cmd.submitSchema = 1; + } + + pSql->res.code = rpcMsg->code; // keep the previous error code + if (pSql->retry > pSql->maxRetry) { + tscError("%p max retry %d reached, give up", pSql, pSql->maxRetry); + } else { + rpcMsg->code = tscRenewTableMeta(pSql, pTableMetaInfo->name); + + // if there is an error occurring, proceed to the following error handling procedure. + if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { + rpcFreeCont(rpcMsg->pCont); + return; } } } diff --git a/src/query/inc/qExecutor.h b/src/query/inc/qexecutor.h similarity index 99% rename from src/query/inc/qExecutor.h rename to src/query/inc/qexecutor.h index fb8750323f..97a7565de8 100644 --- a/src/query/inc/qExecutor.h +++ b/src/query/inc/qexecutor.h @@ -19,15 +19,15 @@ #include "hash.h" #include "qfill.h" -#include "qresultBuf.h" +#include "qresultbuf.h" #include "qsqlparser.h" #include "qtsbuf.h" +#include "query.h" #include "taosdef.h" #include "tarray.h" #include "tlockfree.h" #include "tsdb.h" #include "tsqlfunction.h" -#include "query.h" struct SColumnFilterElem; typedef bool (*__filter_func_t)(struct SColumnFilterElem* pFilter, char* val1, char* val2); @@ -158,7 +158,7 @@ typedef struct SQueryRuntimeEnv { SResultInfo* resultInfo; // todo refactor to merge with SWindowResInfo SQuery* pQuery; SQLFunctionCtx* pCtx; - int16_t numOfRowsPerPage; + int32_t numOfRowsPerPage; int16_t offset[TSDB_MAX_COLUMNS]; uint16_t scanFlag; // denotes reversed scan of data or not SFillInfo* pFillInfo; diff --git a/src/query/inc/qresultBuf.h b/src/query/inc/qresultbuf.h similarity index 75% rename from src/query/inc/qresultBuf.h rename to src/query/inc/qresultbuf.h index a323d530a2..f901474cc1 100644 --- a/src/query/inc/qresultBuf.h +++ b/src/query/inc/qresultbuf.h @@ -33,14 +33,19 @@ typedef struct SDiskbasedResultBuf { int32_t fd; // data file fd int32_t allocateId; // allocated page id int32_t incStep; // minimum allocated pages - char* pBuf; // mmap buffer pointer + void* pBuf; // mmap buffer pointer char* path; // file path - + int32_t pageSize; // current used page size + int32_t inMemPages; // numOfPages that are allocated in memory SHashObj* idsTable; // id hash table SIDList list; // for each id, there is a page id list + + void* iBuf; // inmemory buf + void* handle; // for debug purpose } SDiskbasedResultBuf; -#define DEFAULT_INTERN_BUF_PAGE_SIZE (8192L*5) +#define DEFAULT_INTERN_BUF_PAGE_SIZE (1024L) +#define DEFAULT_INMEM_BUF_PAGES 10 /** * create disk-based result buffer @@ -49,7 +54,8 @@ typedef struct SDiskbasedResultBuf { * @param rowSize * @return */ -int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t size, int32_t rowSize, void* handle); +int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t numOfPages, int32_t rowSize, int32_t pagesize, + int32_t inMemPages, void* handle); /** * @@ -81,8 +87,14 @@ SIDList getDataBufPagesIdList(SDiskbasedResultBuf* pResultBuf, int32_t groupId); * @param id * @return */ -#define GET_RES_BUF_PAGE_BY_ID(buf, id) ((tFilePage*)((buf)->pBuf + DEFAULT_INTERN_BUF_PAGE_SIZE*(id))) - +//#define getResBufPage(buf, id) ((tFilePage*)((buf)->pBuf + (buf)->pageSize * (id))) +static FORCE_INLINE tFilePage* getResBufPage(SDiskbasedResultBuf* pResultBuf, int32_t id) { + if (id < pResultBuf->inMemPages) { + return pResultBuf->iBuf + id * pResultBuf->pageSize; + } else { + return pResultBuf->pBuf + (id - pResultBuf->inMemPages) * pResultBuf->pageSize; + } +} /** * get the total buffer size in the format of disk file * @param pResultBuf diff --git a/src/query/inc/qUtil.h b/src/query/inc/qutil.h similarity index 95% rename from src/query/inc/qUtil.h rename to src/query/inc/qutil.h index 78ae7be030..7119cb75fe 100644 --- a/src/query/inc/qUtil.h +++ b/src/query/inc/qutil.h @@ -49,7 +49,7 @@ static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int3 assert(pResult != NULL && pRuntimeEnv != NULL); SQuery *pQuery = pRuntimeEnv->pQuery; - tFilePage *page = GET_RES_BUF_PAGE_BY_ID(pRuntimeEnv->pResultBuf, pResult->pos.pageId); + tFilePage *page = getResBufPage(pRuntimeEnv->pResultBuf, pResult->pos.pageId); int32_t realRowId = pResult->pos.rowId * GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, pRuntimeEnv->topBotQuery, pRuntimeEnv->stableQuery); return ((char *)page->data) + pRuntimeEnv->offset[columnIndex] * pRuntimeEnv->numOfRowsPerPage + @@ -59,6 +59,4 @@ static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int3 __filter_func_t *getRangeFilterFuncArray(int32_t type); __filter_func_t *getValueFilterFuncArray(int32_t type); -bool supportPrefilter(int32_t type); - #endif // TDENGINE_QUERYUTIL_H diff --git a/src/query/inc/tsqlfunction.h b/src/query/inc/tsqlfunction.h index c687f01cbc..5ce9121cf1 100644 --- a/src/query/inc/tsqlfunction.h +++ b/src/query/inc/tsqlfunction.h @@ -255,7 +255,15 @@ extern int32_t functionCompatList[]; // compatible check array list bool topbot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, const char *minval, const char *maxval); -void resetResultInfo(SResultInfo *pResInfo); +/** + * the numOfRes should be kept, since it may be used later + * and allow the ResultInfo to be re initialized + */ +#define RESET_RESULT_INFO(_r) \ + do { \ + (_r)->initialized = false; \ + } while (0) + void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf); static FORCE_INLINE void initResultInfo(SResultInfo *pResInfo) { diff --git a/src/query/src/qExecutor.c b/src/query/src/qexecutor.c similarity index 98% rename from src/query/src/qExecutor.c rename to src/query/src/qexecutor.c index 42fc90f478..9348f702d0 100644 --- a/src/query/src/qExecutor.c +++ b/src/query/src/qexecutor.c @@ -20,12 +20,12 @@ #include "exception.h" #include "hash.h" -#include "qExecutor.h" -#include "qUtil.h" -#include "qresultBuf.h" +#include "qast.h" +#include "qexecutor.h" +#include "qresultbuf.h" #include "query.h" #include "queryLog.h" -#include "qast.h" +#include "qutil.h" #include "tlosertree.h" #include "tscompression.h" #include "ttime.h" @@ -233,9 +233,7 @@ bool isGroupbyNormalCol(SSqlGroupbyExpr *pGroupbyExpr) { for (int32_t i = 0; i < pGroupbyExpr->numOfGroupCols; ++i) { SColIndex *pColIndex = taosArrayGet(pGroupbyExpr->columnInfo, i); if (pColIndex->flag == TSDB_COL_NORMAL) { - /* - * make sure the normal column locates at the second position if tbname exists in group by clause - */ + //make sure the normal column locates at the second position if tbname exists in group by clause if (pGroupbyExpr->numOfGroupCols > 1) { assert(pColIndex->colIndex > 0); } @@ -294,6 +292,17 @@ bool isSelectivityWithTagsQuery(SQuery *pQuery) { return false; } +bool isProjQuery(SQuery *pQuery) { + for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { + int32_t functId = pQuery->pSelectExpr[i].base.functionId; + if (functId != TSDB_FUNC_PRJ && functId != TSDB_FUNC_TAGPRJ) { + return false; + } + } + + return true; +} + bool isTSCompQuery(SQuery *pQuery) { return pQuery->pSelectExpr[0].base.functionId == TSDB_FUNC_TS_COMP; } static bool limitResults(SQueryRuntimeEnv* pRuntimeEnv) { @@ -470,7 +479,7 @@ static int32_t addNewWindowResultBuf(SWindowResult *pWindowRes, SDiskbasedResult pData = getNewDataBuf(pResultBuf, sid, &pageId); } else { pageId = getLastPageId(list); - pData = GET_RES_BUF_PAGE_BY_ID(pResultBuf, pageId); + pData = getResBufPage(pResultBuf, pageId); if (pData->num >= numOfRowsPerPage) { pData = getNewDataBuf(pResultBuf, sid, &pageId); @@ -1003,7 +1012,6 @@ static int32_t setGroupResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, char *pDat case TSDB_DATA_TYPE_BIGINT: v = GET_INT64_VAL(pData); break; } -// assert(pRuntimeEnv->windowResInfo.hashList->size <= 2); SWindowResult *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, &pRuntimeEnv->windowResInfo, pData, bytes, true); if (pWindowRes == NULL) { return -1; @@ -1201,9 +1209,8 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS continue; } - // interval window query + // interval window query, decide the time window according to the primary timestamp if (QUERY_IS_INTERVAL_QUERY(pQuery)) { - // decide the time window according to the primary timestamp int64_t ts = tsCols[offset]; STimeWindow win = getActiveTimeWindow(pWindowResInfo, ts, pQuery); @@ -1225,8 +1232,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS while (1) { GET_NEXT_TIMEWINDOW(pQuery, &nextWin); - if (/*pWindowResInfo->startTime > nextWin.skey ||*/ - (nextWin.skey > pQuery->window.ekey && QUERY_IS_ASC_QUERY(pQuery)) || + if ((nextWin.skey > pQuery->window.ekey && QUERY_IS_ASC_QUERY(pQuery)) || (nextWin.skey < pQuery->window.ekey && !QUERY_IS_ASC_QUERY(pQuery))) { break; } @@ -1484,6 +1490,8 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order goto _clean; } + qDebug("QInfo:%p setup runtime env1", GET_QINFO_ADDR(pRuntimeEnv)); + pRuntimeEnv->offset[0] = 0; for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { SSqlFuncMsg *pSqlFuncMsg = &pQuery->pSelectExpr[i].base; @@ -1528,6 +1536,8 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order } } + qDebug("QInfo:%p setup runtime env2", GET_QINFO_ADDR(pRuntimeEnv)); + // set the order information for top/bottom query int32_t functionId = pCtx->functionId; @@ -1548,17 +1558,25 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order } } + qDebug("QInfo:%p setup runtime env3", GET_QINFO_ADDR(pRuntimeEnv)); + char* buf = (char*) pRuntimeEnv->resultInfo + sizeof(SResultInfo) * pQuery->numOfOutput; // set the intermediate result output buffer setWindowResultInfo(pRuntimeEnv->resultInfo, pQuery, pRuntimeEnv->stableQuery, buf); + qDebug("QInfo:%p setup runtime env4", GET_QINFO_ADDR(pRuntimeEnv)); + // if it is group by normal column, do not set output buffer, the output buffer is pResult - if (!isGroupbyNormalCol(pQuery->pGroupbyExpr) && !pRuntimeEnv->stableQuery) { + if (!pRuntimeEnv->groupbyNormalCol && !pRuntimeEnv->stableQuery) { resetCtxOutputBuf(pRuntimeEnv); } + qDebug("QInfo:%p setup runtime env5", GET_QINFO_ADDR(pRuntimeEnv)); + setCtxTagColumnInfo(pRuntimeEnv, pRuntimeEnv->pCtx); + + qDebug("QInfo:%p init completed", GET_QINFO_ADDR(pRuntimeEnv)); return TSDB_CODE_SUCCESS; _clean: @@ -1910,9 +1928,20 @@ static int32_t getInitialPageNum(SQInfo *pQInfo) { return num; } -static FORCE_INLINE int32_t getNumOfRowsInResultPage(SQuery *pQuery, bool topBotQuery, bool isSTableQuery) { - int32_t rowSize = pQuery->rowSize * GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, topBotQuery, isSTableQuery); - return (DEFAULT_INTERN_BUF_PAGE_SIZE - sizeof(tFilePage)) / rowSize; +static void getIntermediateBufInfo(SQueryRuntimeEnv* pRuntimeEnv, int32_t* ps, int32_t* rowsize) { + SQuery* pQuery = pRuntimeEnv->pQuery; + + *rowsize = pQuery->rowSize * GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, pRuntimeEnv->topBotQuery, pRuntimeEnv->stableQuery); + int32_t overhead = sizeof(tFilePage); + + // one page contains at least two rows + *ps = DEFAULT_INTERN_BUF_PAGE_SIZE; + while(((*rowsize) * 2) > (*ps) - overhead) { + *ps = (*ps << 1u); + } + + pRuntimeEnv->numOfRowsPerPage = ((*ps) - sizeof(tFilePage)) / (*rowsize); + } #define IS_PREFILTER_TYPE(_t) ((_t) != TSDB_DATA_TYPE_BINARY && (_t) != TSDB_DATA_TYPE_NCHAR) @@ -2038,8 +2067,7 @@ int32_t loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle, } else { // check if this data block is required to load // Calculate all time windows that are overlapping or contain current data block. - // If current data block is contained by all possible time window, loading current - // data block is not needed. + // If current data block is contained by all possible time window, do not load current data block. if (QUERY_IS_INTERVAL_QUERY(pQuery) && overlapWithTimeWindow(pQuery, pBlockInfo)) { status = BLK_DATA_ALL_NEEDED; } @@ -2424,7 +2452,7 @@ static void doMerge(SQueryRuntimeEnv *pRuntimeEnv, int64_t timestamp, SWindowRes pCtx[i].aOutputBuf = pCtx[i].aOutputBuf + pCtx[i].outputBytes; pCtx[i].currentStage = FIRST_STAGE_MERGE; - resetResultInfo(pCtx[i].resultInfo); + RESET_RESULT_INFO(pCtx[i].resultInfo); aAggs[functionId].init(&pCtx[i]); } @@ -2661,7 +2689,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) { int32_t size = taosArrayGetSize(list); for (int32_t i = 0; i < size; ++i) { int32_t* pgId = taosArrayGet(list, i); - tFilePage *pData = GET_RES_BUF_PAGE_BY_ID(pResultBuf, *pgId); + tFilePage *pData = getResBufPage(pResultBuf, *pgId); total += pData->num; } @@ -2670,7 +2698,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) { int32_t offset = 0; for (int32_t j = 0; j < size; ++j) { int32_t* pgId = taosArrayGet(list, j); - tFilePage *pData = GET_RES_BUF_PAGE_BY_ID(pResultBuf, *pgId); + tFilePage *pData = getResBufPage(pResultBuf, *pgId); for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { int32_t bytes = pRuntimeEnv->pCtx[i].outputBytes; @@ -2860,10 +2888,10 @@ int32_t flushFromResultBuf(SQInfo *pQInfo) { SQuery * pQuery = pRuntimeEnv->pQuery; SDiskbasedResultBuf *pResultBuf = pRuntimeEnv->pResultBuf; - int32_t capacity = (DEFAULT_INTERN_BUF_PAGE_SIZE - sizeof(tFilePage)) / pQuery->rowSize; // the base value for group result, since the maximum number of table for each vnode will not exceed 100,000. int32_t pageId = -1; + int32_t capacity = pResultBuf->numOfRowsPerPage; int32_t remain = pQuery->sdata[0]->num; int32_t offset = 0; @@ -3033,7 +3061,7 @@ void resetCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv) { * set the output buffer information and intermediate buffer * not all queries require the interResultBuf, such as COUNT/TAGPRJ/PRJ/TAG etc. */ - resetResultInfo(&pRuntimeEnv->resultInfo[i]); + RESET_RESULT_INFO(&pRuntimeEnv->resultInfo[i]); pCtx->resultInfo = &pRuntimeEnv->resultInfo[i]; // set the timestamp output buffer for top/bottom/diff query @@ -3072,7 +3100,7 @@ void forwardCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, int64_t output) { pRuntimeEnv->pCtx[j].ptsOutputBuf += TSDB_KEYSIZE * output; } - resetResultInfo(pRuntimeEnv->pCtx[j].resultInfo); + RESET_RESULT_INFO(pRuntimeEnv->pCtx[j].resultInfo); } } @@ -3326,8 +3354,8 @@ void scanOneTableDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start) { if (pRuntimeEnv->pSecQueryHandle == NULL) { longjmp(pRuntimeEnv->env, terrno); } - pRuntimeEnv->windowResInfo.curIndex = qstatus.windowIndex; + pRuntimeEnv->windowResInfo.curIndex = qstatus.windowIndex; setQueryStatus(pQuery, QUERY_NOT_COMPLETED); pRuntimeEnv->scanFlag = REPEAT_SCAN; @@ -3462,7 +3490,6 @@ void setExecutionContext(SQInfo *pQInfo, int32_t groupIndex, TSKEY nextKey) { return; } - int32_t GROUPRESULTID = 1; SWindowResult *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, pWindowResInfo, (char *)&groupIndex, sizeof(groupIndex), true); if (pWindowRes == NULL) { @@ -3474,7 +3501,7 @@ void setExecutionContext(SQInfo *pQInfo, int32_t groupIndex, TSKEY nextKey) { * all group belong to one result set, and each group result has different group id so set the id to be one */ if (pWindowRes->pos.pageId == -1) { - if (addNewWindowResultBuf(pWindowRes, pRuntimeEnv->pResultBuf, GROUPRESULTID, pRuntimeEnv->numOfRowsPerPage) != + if (addNewWindowResultBuf(pWindowRes, pRuntimeEnv->pResultBuf, groupIndex, pRuntimeEnv->numOfRowsPerPage) != TSDB_CODE_SUCCESS) { return; } @@ -4170,7 +4197,6 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) { } int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bool isSTableQuery) { - qDebug("QInfo:%p start to init qhandle", pQInfo); SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv; int32_t code = TSDB_CODE_SUCCESS; @@ -4209,33 +4235,42 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo return code; } - pRuntimeEnv->numOfRowsPerPage = getNumOfRowsInResultPage(pQuery, pRuntimeEnv->topBotQuery, isSTableQuery); + int32_t ps = DEFAULT_PAGE_SIZE; + int32_t rowsize = 0; + getIntermediateBufInfo(pRuntimeEnv, &ps, &rowsize); if (isSTableQuery && !onlyQueryTags(pRuntimeEnv->pQuery)) { - int32_t rows = getInitialPageNum(pQInfo); - code = createDiskbasedResultBuffer(&pRuntimeEnv->pResultBuf, rows, pQuery->rowSize, pQInfo); + int32_t numOfPages = getInitialPageNum(pQInfo); + code = createDiskbasedResultBuffer(&pRuntimeEnv->pResultBuf, numOfPages, rowsize, ps, numOfPages, pQInfo); if (code != TSDB_CODE_SUCCESS) { return code; } if (!QUERY_IS_INTERVAL_QUERY(pQuery)) { int16_t type = TSDB_DATA_TYPE_NULL; + int32_t threshold = 0; if (pRuntimeEnv->groupbyNormalCol) { // group by columns not tags; type = getGroupbyColumnType(pQuery, pQuery->pGroupbyExpr); + threshold = 4000; } else { type = TSDB_DATA_TYPE_INT; // group id + threshold = GET_NUM_OF_TABLEGROUP(pQInfo); + if (threshold < 8) { + threshold = 8; + } } - code = initWindowResInfo(&pRuntimeEnv->windowResInfo, pRuntimeEnv, 32, 4096, type); + code = initWindowResInfo(&pRuntimeEnv->windowResInfo, pRuntimeEnv, 8, threshold, type); if (code != TSDB_CODE_SUCCESS) { return code; } } - } else if (pRuntimeEnv->groupbyNormalCol || QUERY_IS_INTERVAL_QUERY(pQuery)) { - int32_t rows = getInitialPageNum(pQInfo); - code = createDiskbasedResultBuffer(&pRuntimeEnv->pResultBuf, rows, pQuery->rowSize, pQInfo); + int32_t numOfResultRows = getInitialPageNum(pQInfo); + getIntermediateBufInfo(pRuntimeEnv, &ps, &rowsize); + + code = createDiskbasedResultBuffer(&pRuntimeEnv->pResultBuf, numOfResultRows, rowsize, ps, numOfResultRows, pQInfo); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -4247,7 +4282,7 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo type = TSDB_DATA_TYPE_TIMESTAMP; } - code = initWindowResInfo(&pRuntimeEnv->windowResInfo, pRuntimeEnv, rows, 4096, type); + code = initWindowResInfo(&pRuntimeEnv->windowResInfo, pRuntimeEnv, numOfResultRows, 4096, type); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -5705,13 +5740,18 @@ static void calResultBufSize(SQuery* pQuery) { const int32_t RESULT_MSG_MIN_ROWS = 8192; const float RESULT_THRESHOLD_RATIO = 0.85; - int32_t numOfRes = RESULT_MSG_MIN_SIZE / pQuery->rowSize; - if (numOfRes < RESULT_MSG_MIN_ROWS) { - numOfRes = RESULT_MSG_MIN_ROWS; - } + if (isProjQuery(pQuery)) { + int32_t numOfRes = RESULT_MSG_MIN_SIZE / pQuery->rowSize; + if (numOfRes < RESULT_MSG_MIN_ROWS) { + numOfRes = RESULT_MSG_MIN_ROWS; + } - pQuery->rec.capacity = numOfRes; - pQuery->rec.threshold = numOfRes * RESULT_THRESHOLD_RATIO; + pQuery->rec.capacity = numOfRes; + pQuery->rec.threshold = numOfRes * RESULT_THRESHOLD_RATIO; + } else { // in case of non-prj query, a smaller output buffer will be used. + pQuery->rec.capacity = 4096; + pQuery->rec.threshold = pQuery->rec.capacity * RESULT_THRESHOLD_RATIO; + } } static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList, SSqlGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs, @@ -5723,6 +5763,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList, if (pQInfo == NULL) { goto _cleanup_qinfo; } + // to make sure third party won't overwrite this structure pQInfo->signature = pQInfo; pQInfo->tableGroupInfo = *pTableGroupInfo; @@ -6041,11 +6082,10 @@ static void freeQInfo(SQInfo *pQInfo) { tfree(pQuery->sdata); tfree(pQuery); + pQInfo->signature = 0; qDebug("QInfo:%p QInfo is freed", pQInfo); - // destroy signature, in order to avoid the query process pass the object safety check - memset(pQInfo, 0, sizeof(SQInfo)); tfree(pQInfo); } @@ -6209,7 +6249,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, vo } int64_t el = taosGetTimestampUs() - st; - qDebug("qmsg:%p tag filter completed, elapsed time:%"PRId64"us", pQueryMsg, el); + qDebug("qmsg:%p tag filter completed, numOfTables:%zu, elapsed time:%"PRId64"us", pQueryMsg, tableGroupInfo.numOfTables, el); } else { assert(0); } diff --git a/src/query/src/qFilterFunc.c b/src/query/src/qfilterfunc.c similarity index 99% rename from src/query/src/qFilterFunc.c rename to src/query/src/qfilterfunc.c index 1a95b9fe21..6f336434e1 100644 --- a/src/query/src/qFilterFunc.c +++ b/src/query/src/qfilterfunc.c @@ -16,7 +16,7 @@ #define _DEFAULT_SOURCE #include "os.h" -#include "qExecutor.h" +#include "qexecutor.h" #include "taosmsg.h" #include "tcompare.h" #include "tsqlfunction.h" @@ -554,5 +554,3 @@ __filter_func_t* getValueFilterFuncArray(int32_t type) { default: return NULL; } } - -bool supportPrefilter(int32_t type) { return type != TSDB_DATA_TYPE_BINARY && type != TSDB_DATA_TYPE_NCHAR; } diff --git a/src/query/src/qresultBuf.c b/src/query/src/qresultBuf.c deleted file mode 100644 index ae1a95179b..0000000000 --- a/src/query/src/qresultBuf.c +++ /dev/null @@ -1,184 +0,0 @@ -#include "qresultBuf.h" -#include "hash.h" -#include "qextbuffer.h" -#include "taoserror.h" -#include "queryLog.h" - -int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t size, int32_t rowSize, void* handle) { - *pResultBuf = calloc(1, sizeof(SDiskbasedResultBuf)); - SDiskbasedResultBuf* pResBuf = *pResultBuf; - if (pResBuf == NULL) { - return TSDB_CODE_COM_OUT_OF_MEMORY; - } - - pResBuf->numOfRowsPerPage = (DEFAULT_INTERN_BUF_PAGE_SIZE - sizeof(tFilePage)) / rowSize; - pResBuf->numOfPages = size; - - pResBuf->totalBufSize = pResBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE; - pResBuf->incStep = 4; - - // init id hash table - pResBuf->idsTable = taosHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false); - pResBuf->list = taosArrayInit(size, POINTER_BYTES); - - char path[4096] = {0}; - getTmpfilePath("tsdb_qbuf", path); - pResBuf->path = strdup(path); - - pResBuf->fd = open(pResBuf->path, O_CREAT | O_RDWR, 0666); - if (!FD_VALID(pResBuf->fd)) { - qError("failed to create tmp file: %s on disk. %s", pResBuf->path, strerror(errno)); - return TAOS_SYSTEM_ERROR(errno); - } - - int32_t ret = ftruncate(pResBuf->fd, pResBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE); - if (ret != TSDB_CODE_SUCCESS) { - qError("failed to create tmp file: %s on disk. %s", pResBuf->path, strerror(errno)); - return TAOS_SYSTEM_ERROR(errno); - } - - pResBuf->pBuf = mmap(NULL, pResBuf->totalBufSize, PROT_READ | PROT_WRITE, MAP_SHARED, pResBuf->fd, 0); - if (pResBuf->pBuf == MAP_FAILED) { - qError("QInfo:%p failed to map temp file: %s. %s", handle, pResBuf->path, strerror(errno)); - return TAOS_SYSTEM_ERROR(errno); - } - - qDebug("QInfo:%p create tmp file for output result:%s, %" PRId64 "bytes", handle, pResBuf->path, - pResBuf->totalBufSize); - - return TSDB_CODE_SUCCESS; -} - -int32_t getNumOfResultBufGroupId(SDiskbasedResultBuf* pResultBuf) { return taosHashGetSize(pResultBuf->idsTable); } - -int32_t getResBufSize(SDiskbasedResultBuf* pResultBuf) { return pResultBuf->totalBufSize; } - -static int32_t extendDiskFileSize(SDiskbasedResultBuf* pResultBuf, int32_t numOfPages) { - assert(pResultBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE == pResultBuf->totalBufSize); - - int32_t ret = munmap(pResultBuf->pBuf, pResultBuf->totalBufSize); - pResultBuf->numOfPages += numOfPages; - - /* - * disk-based output buffer is exhausted, try to extend the disk-based buffer, the available disk space may - * be insufficient - */ - ret = ftruncate(pResultBuf->fd, pResultBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE); - if (ret != 0) { - // dError("QInfo:%p failed to create intermediate result output file:%s. %s", pQInfo, pSupporter->extBufFile, - // strerror(errno)); - return TSDB_CODE_QRY_NO_DISKSPACE; - } - - pResultBuf->totalBufSize = pResultBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE; - pResultBuf->pBuf = mmap(NULL, pResultBuf->totalBufSize, PROT_READ | PROT_WRITE, MAP_SHARED, pResultBuf->fd, 0); - - if (pResultBuf->pBuf == MAP_FAILED) { - // dError("QInfo:%p failed to map temp file: %s. %s", pQInfo, pSupporter->extBufFile, strerror(errno)); - return TSDB_CODE_QRY_OUT_OF_MEMORY; - } - - return TSDB_CODE_SUCCESS; -} - -static FORCE_INLINE bool noMoreAvailablePages(SDiskbasedResultBuf* pResultBuf) { - return (pResultBuf->allocateId == pResultBuf->numOfPages - 1); -} - -static FORCE_INLINE int32_t getGroupIndex(SDiskbasedResultBuf* pResultBuf, int32_t groupId) { - assert(pResultBuf != NULL); - - char* p = taosHashGet(pResultBuf->idsTable, (const char*)&groupId, sizeof(int32_t)); - if (p == NULL) { // it is a new group id - return -1; - } - - int32_t slot = GET_INT32_VAL(p); - assert(slot >= 0 && slot < taosHashGetSize(pResultBuf->idsTable)); - - return slot; -} - -static int32_t addNewGroupId(SDiskbasedResultBuf* pResultBuf, int32_t groupId) { - int32_t num = getNumOfResultBufGroupId(pResultBuf); // the num is the newest allocated group id slot - taosHashPut(pResultBuf->idsTable, (const char*)&groupId, sizeof(int32_t), &num, sizeof(int32_t)); - - SArray* pa = taosArrayInit(1, sizeof(int32_t)); - taosArrayPush(pResultBuf->list, &pa); - - assert(taosArrayGetSize(pResultBuf->list) == taosHashGetSize(pResultBuf->idsTable)); - return num; -} - -static void registerPageId(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int32_t pageId) { - int32_t slot = getGroupIndex(pResultBuf, groupId); - if (slot < 0) { - slot = addNewGroupId(pResultBuf, groupId); - } - - SIDList pList = taosArrayGetP(pResultBuf->list, slot); - taosArrayPush(pList, &pageId); -} - -tFilePage* getNewDataBuf(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int32_t* pageId) { - if (noMoreAvailablePages(pResultBuf)) { - if (extendDiskFileSize(pResultBuf, pResultBuf->incStep) != TSDB_CODE_SUCCESS) { - return NULL; - } - } - - // register new id in this group - *pageId = (pResultBuf->allocateId++); - registerPageId(pResultBuf, groupId, *pageId); - - tFilePage* page = GET_RES_BUF_PAGE_BY_ID(pResultBuf, *pageId); - - // clear memory for the new page - memset(page, 0, DEFAULT_INTERN_BUF_PAGE_SIZE); - - return page; -} - -int32_t getNumOfRowsPerPage(SDiskbasedResultBuf* pResultBuf) { return pResultBuf->numOfRowsPerPage; } - -SIDList getDataBufPagesIdList(SDiskbasedResultBuf* pResultBuf, int32_t groupId) { - int32_t slot = getGroupIndex(pResultBuf, groupId); - if (slot < 0) { - return taosArrayInit(1, sizeof(int32_t)); - } else { - return taosArrayGetP(pResultBuf->list, slot); - } -} - -void destroyResultBuf(SDiskbasedResultBuf* pResultBuf, void* handle) { - if (pResultBuf == NULL) { - return; - } - - if (FD_VALID(pResultBuf->fd)) { - close(pResultBuf->fd); - } - - qDebug("QInfo:%p disk-based output buffer closed, %" PRId64 " bytes, file:%s", handle, pResultBuf->totalBufSize, pResultBuf->path); - munmap(pResultBuf->pBuf, pResultBuf->totalBufSize); - unlink(pResultBuf->path); - - tfree(pResultBuf->path); - - size_t size = taosArrayGetSize(pResultBuf->list); - for (int32_t i = 0; i < size; ++i) { - SArray* pa = taosArrayGetP(pResultBuf->list, i); - taosArrayDestroy(pa); - } - - taosArrayDestroy(pResultBuf->list); - taosHashCleanup(pResultBuf->idsTable); - - tfree(pResultBuf); -} - -int32_t getLastPageId(SIDList pList) { - size_t size = taosArrayGetSize(pList); - return *(int32_t*) taosArrayGet(pList, size - 1); -} - diff --git a/src/query/src/qresultbuf.c b/src/query/src/qresultbuf.c new file mode 100644 index 0000000000..646d8b7986 --- /dev/null +++ b/src/query/src/qresultbuf.c @@ -0,0 +1,218 @@ +#include "qresultbuf.h" +#include "hash.h" +#include "qextbuffer.h" +#include "queryLog.h" +#include "taoserror.h" + +int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t numOfPages, int32_t rowSize, + int32_t pagesize, int32_t inMemPages, void* handle) { + + *pResultBuf = calloc(1, sizeof(SDiskbasedResultBuf)); + SDiskbasedResultBuf* pResBuf = *pResultBuf; + if (pResBuf == NULL) { + return TSDB_CODE_COM_OUT_OF_MEMORY; + } + + pResBuf->pageSize = pagesize; + pResBuf->numOfPages = inMemPages; // all pages are in buffer in the first place + pResBuf->inMemPages = inMemPages; + assert(inMemPages <= numOfPages); + + pResBuf->numOfRowsPerPage = (pagesize - sizeof(tFilePage)) / rowSize; + + pResBuf->totalBufSize = pResBuf->numOfPages * pagesize; + pResBuf->incStep = 4; + pResBuf->allocateId = -1; + + pResBuf->iBuf = calloc(pResBuf->inMemPages, pResBuf->pageSize); + + // init id hash table + pResBuf->idsTable = taosHashInit(numOfPages, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false); + pResBuf->list = taosArrayInit(numOfPages, POINTER_BYTES); + + char path[PATH_MAX] = {0}; + getTmpfilePath("tsdb_qbuf", path); + pResBuf->path = strdup(path); + + pResBuf->fd = FD_INITIALIZER; + pResBuf->pBuf = NULL; + + qDebug("QInfo:%p create resBuf for output, page size:%d, initial pages:%d, %" PRId64 "bytes", handle, + pResBuf->pageSize, pResBuf->numOfPages, pResBuf->totalBufSize); + + return TSDB_CODE_SUCCESS; +} + +int32_t getNumOfResultBufGroupId(SDiskbasedResultBuf* pResultBuf) { return taosHashGetSize(pResultBuf->idsTable); } + +int32_t getResBufSize(SDiskbasedResultBuf* pResultBuf) { return pResultBuf->totalBufSize; } + +#define NUM_OF_PAGES_ON_DISK(_r) ((_r)->numOfPages - (_r)->inMemPages) +#define FILE_SIZE_ON_DISK(_r) (NUM_OF_PAGES_ON_DISK(_r) * (_r)->pageSize) + +static int32_t createDiskResidesBuf(SDiskbasedResultBuf* pResultBuf) { + pResultBuf->fd = open(pResultBuf->path, O_CREAT | O_RDWR, 0666); + if (!FD_VALID(pResultBuf->fd)) { + qError("failed to create tmp file: %s on disk. %s", pResultBuf->path, strerror(errno)); + return TAOS_SYSTEM_ERROR(errno); + } + + assert(pResultBuf->numOfPages == pResultBuf->inMemPages); + pResultBuf->numOfPages += pResultBuf->incStep; + + int32_t ret = ftruncate(pResultBuf->fd, NUM_OF_PAGES_ON_DISK(pResultBuf) * pResultBuf->pageSize); + if (ret != TSDB_CODE_SUCCESS) { + qError("failed to create tmp file: %s on disk. %s", pResultBuf->path, strerror(errno)); + return TAOS_SYSTEM_ERROR(errno); + } + + pResultBuf->pBuf = mmap(NULL, FILE_SIZE_ON_DISK(pResultBuf), PROT_READ | PROT_WRITE, MAP_SHARED, pResultBuf->fd, 0); + if (pResultBuf->pBuf == MAP_FAILED) { + qError("QInfo:%p failed to map temp file: %s. %s", pResultBuf->handle, pResultBuf->path, strerror(errno)); + return TAOS_SYSTEM_ERROR(errno); + } + + pResultBuf->totalBufSize = pResultBuf->numOfPages * pResultBuf->pageSize; + return TSDB_CODE_SUCCESS; +} + +static int32_t extendDiskFileSize(SDiskbasedResultBuf* pResultBuf, int32_t incNumOfPages) { + assert(pResultBuf->numOfPages * pResultBuf->pageSize == pResultBuf->totalBufSize); + int32_t ret = TSDB_CODE_SUCCESS; + + if (pResultBuf->pBuf == NULL) { + assert(pResultBuf->fd == FD_INITIALIZER); + + if ((ret = createDiskResidesBuf(pResultBuf)) != TSDB_CODE_SUCCESS) { + return ret; + } + } else { + ret = munmap(pResultBuf->pBuf, FILE_SIZE_ON_DISK(pResultBuf)); + pResultBuf->numOfPages += incNumOfPages; + + /* + * disk-based output buffer is exhausted, try to extend the disk-based buffer, the available disk space may + * be insufficient + */ + ret = ftruncate(pResultBuf->fd, NUM_OF_PAGES_ON_DISK(pResultBuf) * pResultBuf->pageSize); + if (ret != TSDB_CODE_SUCCESS) { + // dError("QInfo:%p failed to create intermediate result output file:%s. %s", pQInfo, pSupporter->extBufFile, + // strerror(errno)); + return TSDB_CODE_QRY_NO_DISKSPACE; + } + + pResultBuf->totalBufSize = pResultBuf->numOfPages * pResultBuf->pageSize; + pResultBuf->pBuf = mmap(NULL, FILE_SIZE_ON_DISK(pResultBuf), PROT_READ | PROT_WRITE, MAP_SHARED, pResultBuf->fd, 0); + + if (pResultBuf->pBuf == MAP_FAILED) { + // dError("QInfo:%p failed to map temp file: %s. %s", pQInfo, pSupporter->extBufFile, strerror(errno)); + return TSDB_CODE_QRY_OUT_OF_MEMORY; + } + } + + return TSDB_CODE_SUCCESS; +} + +#define NO_AVAILABLE_PAGES(_b) ((_b)->allocateId == (_b)->numOfPages - 1) + +static FORCE_INLINE int32_t getGroupIndex(SDiskbasedResultBuf* pResultBuf, int32_t groupId) { + assert(pResultBuf != NULL); + + char* p = taosHashGet(pResultBuf->idsTable, (const char*)&groupId, sizeof(int32_t)); + if (p == NULL) { // it is a new group id + return -1; + } + + int32_t slot = GET_INT32_VAL(p); + assert(slot >= 0 && slot < taosHashGetSize(pResultBuf->idsTable)); + + return slot; +} + +static int32_t addNewGroupId(SDiskbasedResultBuf* pResultBuf, int32_t groupId) { + int32_t num = getNumOfResultBufGroupId(pResultBuf); // the num is the newest allocated group id slot + taosHashPut(pResultBuf->idsTable, (const char*)&groupId, sizeof(int32_t), &num, sizeof(int32_t)); + + SArray* pa = taosArrayInit(1, sizeof(int32_t)); + taosArrayPush(pResultBuf->list, &pa); + + assert(taosArrayGetSize(pResultBuf->list) == taosHashGetSize(pResultBuf->idsTable)); + return num; +} + +static void registerPageId(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int32_t pageId) { + int32_t slot = getGroupIndex(pResultBuf, groupId); + if (slot < 0) { + slot = addNewGroupId(pResultBuf, groupId); + } + + SIDList pList = taosArrayGetP(pResultBuf->list, slot); + taosArrayPush(pList, &pageId); +} + +tFilePage* getNewDataBuf(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int32_t* pageId) { + if (NO_AVAILABLE_PAGES(pResultBuf)) { + if (extendDiskFileSize(pResultBuf, pResultBuf->incStep) != TSDB_CODE_SUCCESS) { + return NULL; + } + } + + // register new id in this group + *pageId = (++pResultBuf->allocateId); + registerPageId(pResultBuf, groupId, *pageId); + + // clear memory for the new page + tFilePage* page = getResBufPage(pResultBuf, *pageId); + memset(page, 0, pResultBuf->pageSize); + + return page; +} + +int32_t getNumOfRowsPerPage(SDiskbasedResultBuf* pResultBuf) { return pResultBuf->numOfRowsPerPage; } + +SIDList getDataBufPagesIdList(SDiskbasedResultBuf* pResultBuf, int32_t groupId) { + int32_t slot = getGroupIndex(pResultBuf, groupId); + if (slot < 0) { + return taosArrayInit(1, sizeof(int32_t)); + } else { + return taosArrayGetP(pResultBuf->list, slot); + } +} + +void destroyResultBuf(SDiskbasedResultBuf* pResultBuf, void* handle) { + if (pResultBuf == NULL) { + return; + } + + if (FD_VALID(pResultBuf->fd)) { + qDebug("QInfo:%p disk-based output buffer closed, total:%" PRId64 " bytes, file created:%s, file size:%d", handle, + pResultBuf->totalBufSize, pResultBuf->path, FILE_SIZE_ON_DISK(pResultBuf)); + + close(pResultBuf->fd); + munmap(pResultBuf->pBuf, FILE_SIZE_ON_DISK(pResultBuf)); + pResultBuf->pBuf = NULL; + } else { + qDebug("QInfo:%p disk-based output buffer closed, total:%" PRId64 " bytes, no file created", handle, + pResultBuf->totalBufSize); + } + + unlink(pResultBuf->path); + tfree(pResultBuf->path); + + size_t size = taosArrayGetSize(pResultBuf->list); + for (int32_t i = 0; i < size; ++i) { + SArray* pa = taosArrayGetP(pResultBuf->list, i); + taosArrayDestroy(pa); + } + + taosArrayDestroy(pResultBuf->list); + taosHashCleanup(pResultBuf->idsTable); + + tfree(pResultBuf); +} + +int32_t getLastPageId(SIDList pList) { + size_t size = taosArrayGetSize(pList); + return *(int32_t*) taosArrayGet(pList, size - 1); +} + diff --git a/src/query/src/qUtil.c b/src/query/src/qutil.c similarity index 98% rename from src/query/src/qUtil.c rename to src/query/src/qutil.c index 4e71de830f..4c3595b0be 100644 --- a/src/query/src/qUtil.c +++ b/src/query/src/qutil.c @@ -19,19 +19,17 @@ #include "taosmsg.h" #include "ttime.h" -#include "qExecutor.h" -#include "qUtil.h" +#include "qexecutor.h" +#include "qutil.h" int32_t getOutputInterResultBufSize(SQuery* pQuery) { int32_t size = 0; for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { - assert(pQuery->pSelectExpr[i].interBytes <= DEFAULT_INTERN_BUF_PAGE_SIZE); size += pQuery->pSelectExpr[i].interBytes; } assert(size > 0); - return size; } @@ -243,7 +241,7 @@ void clearTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pWindow size_t size = pRuntimeEnv->pQuery->pSelectExpr[i].bytes; memset(s, 0, size); - resetResultInfo(pResultInfo); + RESET_RESULT_INFO(pResultInfo); } pWindowRes->numOfRows = 0; diff --git a/src/query/tests/resultBufferTest.cpp b/src/query/tests/resultBufferTest.cpp index 9270c6e458..bf6f3bb50f 100644 --- a/src/query/tests/resultBufferTest.cpp +++ b/src/query/tests/resultBufferTest.cpp @@ -2,8 +2,8 @@ #include #include +#include "qresultbuf.h" #include "taos.h" -#include "qresultBuf.h" #include "tsdb.h" namespace { diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c index 8463e791db..5cb0dffd46 100644 --- a/src/tsdb/src/tsdbRead.c +++ b/src/tsdb/src/tsdbRead.c @@ -1376,7 +1376,7 @@ static int32_t createDataBlocksInfo(STsdbQueryHandle* pQueryHandle, int32_t numO * } */ - tsdbDebug("%p %d data blocks sort completed", pQueryHandle, cnt); + tsdbDebug("%p %d data blocks sort completed, %p", pQueryHandle, cnt, pQueryHandle->qinfo); cleanBlockOrderSupporter(&sup, numOfTables); free(pTree); diff --git a/src/util/src/tutil.c b/src/util/src/tutil.c index 3764df4afc..7b73b1e17c 100644 --- a/src/util/src/tutil.c +++ b/src/util/src/tutil.c @@ -522,7 +522,7 @@ int32_t taosFileRename(char *fullPath, char *suffix, char delimiter, char **dstP void getTmpfilePath(const char *fileNamePrefix, char *dstPath) { const char* tdengineTmpFileNamePrefix = "tdengine-"; - char tmpPath[PATH_MAX] = {0}; + char tmpPath[PATH_MAX]; #ifdef WINDOWS char *tmpDir = getenv("tmp"); -- GitLab