diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 6100d416b11d22df45cca1119c1ebfad50891095..bf178612bacef9fa25440e028f23b19661797894 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -388,13 +388,15 @@ typedef struct SStreamBlockScanInfo { SColumnInfo* pCols; // the output column info uint64_t numOfRows; // total scanned rows uint64_t numOfExec; // execution times - void* readerHandle; // stream block reader handle + void* streamBlockReader;// stream block reader handle SArray* pColMatchInfo; // SNode* pCondition; SArray* tsArray; SUpdateInfo* pUpdateInfo; int32_t primaryTsIndex; // primary time stamp slot id void* pDataReader; + SReadHandle readHandle; + uint64_t tableUid; // queried super table uid EStreamScanMode scanMode; SOperatorInfo* pOperatorDumy; SInterval interval; // if the upstream is an interval operator, the interval info is also kept here. @@ -706,9 +708,10 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo); SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo); -SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataReader, SSDataBlock* pResBlock, - SArray* pColList, SArray* pTableIdList, SExecTaskInfo* pTaskInfo, - SNode* pConditions, SOperatorInfo* pOperatorDumy); +SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataReader, SReadHandle* pHandle, + uint64_t uid, SSDataBlock* pResBlock, SArray* pColList, + SArray* pTableIdList, SExecTaskInfo* pTaskInfo, SNode* pCondition, + SOperatorInfo* pOperatorDumy); SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SInterval* pInterval, STimeWindow* pWindow, SSDataBlock* pResBlock, int32_t fillType, SNodeListNode* fillVal, diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 4863b03fb9f69fcba57bea23b359081fd857568f..320450eb6e5eb8703b82c399e69a6f02088c5101 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -14,6 +14,7 @@ */ #include "executor.h" +#include #include "executorimpl.h" #include "planner.h" #include "tdatablock.h" @@ -46,7 +47,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu } if (type == STREAM_DATA_TYPE_SUBMIT_BLOCK) { - if (tqReadHandleSetMsg(pInfo->readerHandle, input, 0) < 0) { + if (tqReadHandleSetMsg(pInfo->streamBlockReader, input, 0) < 0) { qError("submit msg messed up when initing stream block, %s" PRIx64, id); return TSDB_CODE_QRY_APP_ERROR; } @@ -128,7 +129,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle) { int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, SArray* tableIdList, bool isAdd) { SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; - // traverse to the streamscan node to add this table id + // traverse to the stream scanner node to add this table id SOperatorInfo* pInfo = pTaskInfo->pRoot; while (pInfo->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { pInfo = pInfo->pDownstream[0]; @@ -136,7 +137,31 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, SArray* tableIdList, bool isA SStreamBlockScanInfo* pScanInfo = pInfo->info; if (isAdd) { - int32_t code = tqReadHandleAddTbUidList(pScanInfo->readerHandle, tableIdList); + SArray* qa = taosArrayInit(4, sizeof(tb_uid_t)); + + SMetaReader mr = {0}; + metaReaderInit(&mr, pScanInfo->readHandle.meta, 0); + for(int32_t i = 0; i < taosArrayGetSize(tableIdList); ++i) { + int64_t* id = (int64_t*)taosArrayGet(tableIdList, i); + + int32_t code = metaGetTableEntryByUid(&mr, *id); + if (code != TSDB_CODE_SUCCESS) { + qError("failed to get table meta, uid:%"PRIu64" code:%s", *id, tstrerror(terrno)); + continue; + } + + ASSERT(mr.me.type == TSDB_CHILD_TABLE); + if (mr.me.ctbEntry.suid != pScanInfo->tableUid) { + continue; + } + + taosArrayPush(qa, id); + } + + metaReaderClear(&mr); + + qDebug(" %d qualified child tables added into stream scanner", (int32_t) taosArrayGetSize(qa)); + int32_t code = tqReadHandleAddTbUidList(pScanInfo->streamBlockReader, qa); if (code != TSDB_CODE_SUCCESS) { return code; } diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 688901226c1ca832603e975ec49c00a1082f7f3b..c261271728c5a4a2930e7a7d172a1752c69db271 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -155,7 +155,7 @@ SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, void operatorDummyCloseFn(void* param, int32_t numOfCols) {} static int32_t doCopyToSDataBlock(SExecTaskInfo* taskInfo, SSDataBlock* pBlock, SExprInfo* pExprInfo, SDiskbasedBuf* pBuf, SGroupResInfo* pGroupResInfo, - int32_t orderType, int32_t* rowCellOffset, SqlFunctionCtx* pCtx, int32_t numOfExprs); + int32_t* rowCellOffset, SqlFunctionCtx* pCtx, int32_t numOfExprs); static void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size); static void setResultBufSize(STaskAttr* pQueryAttr, SResultInfo* pResultInfo); @@ -2136,23 +2136,13 @@ void setExecutionContext(int32_t numOfOutput, uint64_t groupId, SExecTaskInfo* p * @param result */ int32_t doCopyToSDataBlock(SExecTaskInfo* taskInfo, SSDataBlock* pBlock, SExprInfo* pExprInfo, SDiskbasedBuf* pBuf, SGroupResInfo* pGroupResInfo, - int32_t orderType, int32_t* rowCellOffset, SqlFunctionCtx* pCtx, int32_t numOfExprs) { + int32_t* rowCellOffset, SqlFunctionCtx* pCtx, int32_t numOfExprs) { int32_t numOfRows = getNumOfTotalRes(pGroupResInfo); - int32_t numOfResult = pBlock->info.rows; // there are already exists result rows - - int32_t start = 0; - int32_t step = 1; + int32_t start = pGroupResInfo->index; // qDebug("QInfo:0x%"PRIx64" start to copy data from windowResInfo to output buf", GET_TASKID(pRuntimeEnv)); - assert(orderType == TSDB_ORDER_ASC || orderType == TSDB_ORDER_DESC); - - if (orderType == TSDB_ORDER_ASC) { - start = pGroupResInfo->index; - } else { // desc order copy all data - start = numOfRows - pGroupResInfo->index - 1; - } - for (int32_t i = start; (i < numOfRows) && (i >= 0); i += step) { + for (int32_t i = start; i < numOfRows; i += 1) { SResKeyPos* pPos = taosArrayGetP(pGroupResInfo->pRows, i); SFilePage* page = getBufPage(pBuf, pPos->pos.pageId); @@ -2162,9 +2152,7 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* taskInfo, SSDataBlock* pBlock, SExprIn continue; } - // TODO copy multiple rows? - int32_t numOfRowsToCopy = pRow->numOfRows; - if (numOfResult + numOfRowsToCopy >= pBlock->info.capacity) { + if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) { break; } @@ -2195,7 +2183,6 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* taskInfo, SSDataBlock* pBlock, SExprIn } releaseBufPage(pBuf, page); - pBlock->info.rows += pRow->numOfRows; if (pBlock->info.rows >= pBlock->info.capacity) { // output buffer is full break; @@ -2223,8 +2210,7 @@ void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SG return; } - int32_t orderType = TSDB_ORDER_ASC; - doCopyToSDataBlock(pTaskInfo, pBlock, pExprInfo, pBuf, pGroupResInfo, orderType, rowCellOffset, pCtx, numOfExprs); + doCopyToSDataBlock(pTaskInfo, pBlock, pExprInfo, pBuf, pGroupResInfo, rowCellOffset, pCtx, numOfExprs); // add condition (pBlock->info.rows >= 1) just to runtime happy blockDataUpdateTsWindow(pBlock); @@ -4717,10 +4703,11 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo SOperatorInfo* pOperatorDumy = createTableScanOperatorInfo(pTableScanNode, pDataReader, pHandle, pTaskInfo); SArray* tableIdList = extractTableIdList(pTableGroupInfo); - SSDataBlock* pResBlock = createResDataBlock(pDescNode); - SArray* pCols = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); - SOperatorInfo* pOperator = createStreamScanOperatorInfo(pHandle->reader, pDataReader, pResBlock, pCols, tableIdList, pTaskInfo, + SSDataBlock* pResBlock = createResDataBlock(pDescNode); + SArray* pCols = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); + + SOperatorInfo* pOperator = createStreamScanOperatorInfo(pHandle->reader, pDataReader, pHandle, pScanPhyNode->uid, pResBlock, pCols, tableIdList, pTaskInfo, pScanPhyNode->node.pConditions, pOperatorDumy); taosArrayDestroy(tableIdList); return pOperator; diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 1a862c94f34004089a6546f3b97899b4befd7c01..df7478264f412cbdaf4298bb9a444df65aa3e504 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -710,14 +710,14 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) { SDataBlockInfo* pBlockInfo = &pInfo->pRes->info; blockDataCleanup(pInfo->pRes); - while (tqNextDataBlock(pInfo->readerHandle)) { + while (tqNextDataBlock(pInfo->streamBlockReader)) { SArray* pCols = NULL; uint64_t groupId = 0; uint64_t uid = 0; int32_t numOfRows = 0; int16_t outputCol = 0; - int32_t code = tqRetrieveDataBlock(&pCols, pInfo->readerHandle, &groupId, &uid, &numOfRows, &outputCol); + int32_t code = tqRetrieveDataBlock(&pCols, pInfo->streamBlockReader, &groupId, &uid, &numOfRows, &outputCol); if (code != TSDB_CODE_SUCCESS || numOfRows == 0) { pTaskInfo->code = code; @@ -791,9 +791,10 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) { } } -SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataReader, - SSDataBlock* pResBlock, SArray* pColList, SArray* pTableIdList, - SExecTaskInfo* pTaskInfo, SNode* pCondition, SOperatorInfo* pOperatorDumy ) { +SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataReader, SReadHandle* pHandle, + uint64_t uid, SSDataBlock* pResBlock, SArray* pColList, + SArray* pTableIdList, SExecTaskInfo* pTaskInfo, SNode* pCondition, + SOperatorInfo* pOperatorDumy) { SStreamBlockScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamBlockScanInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); if (pInfo == NULL || pOperator == NULL) { @@ -829,37 +830,32 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataR pInfo->tsArray = taosArrayInit(4, sizeof(TSKEY)); if (pInfo->tsArray == NULL) { - taosMemoryFreeClear(pInfo); - taosMemoryFreeClear(pOperator); - return NULL; + goto _error; } pInfo->primaryTsIndex = 0; // TODO(liuyao) get it from physical plan pInfo->pUpdateInfo = updateInfoInitP(&pSTInfo->interval, 10000); // TODO(liuyao) get watermark from physical plan if (pInfo->pUpdateInfo == NULL) { - taosMemoryFreeClear(pInfo); - taosMemoryFreeClear(pOperator); - return NULL; + goto _error; } - pInfo->readerHandle = streamReadHandle; - pInfo->pRes = pResBlock; - pInfo->pCondition = pCondition; - pInfo->pDataReader = pDataReader; - pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; - pInfo->pOperatorDumy = pOperatorDumy; - pInfo->interval = pSTInfo->interval; - - pOperator->name = "StreamBlockScanOperator"; + pInfo->readHandle = *pHandle; + pInfo->tableUid = uid; + pInfo->streamBlockReader = streamReadHandle; + pInfo->pRes = pResBlock; + pInfo->pCondition = pCondition; + pInfo->pDataReader = pDataReader; + pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; + pInfo->pOperatorDumy = pOperatorDumy; + pInfo->interval = pSTInfo->interval; + + pOperator->name = "StreamBlockScanOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; pOperator->numOfExprs = pResBlock->info.numOfCols; - pOperator->fpSet._openFn = operatorDummyOpenFn; - pOperator->fpSet.getNextFn = doStreamBlockScan; - pOperator->fpSet.closeFn = operatorDummyCloseFn; - pOperator->pTaskInfo = pTaskInfo; + pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamBlockScan, NULL, NULL, operatorDummyCloseFn, NULL, NULL, NULL);