taosdata / TDengine / #5049

coverage: 73.313% (+0.09%) from 73.222%
#5049

push

travis-ci

web-flow 
feat: refactor taosdump code to improve backup speed and compression ratio (#35292)

6625 of 8435 new or added lines in 28 files covered. (78.54%)

2491 existing lines in 142 files now uncovered.

281233 of 383605 relevant lines covered (73.31%)

132489999.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.1
/source/libs/executor/src/eventwindowoperator.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 









16



#include "executorInt.h"










17



#include "filter.h"










18



#include "function.h"










19



#include "functionMgt.h"










20



#include "operator.h"










21



#include "querytask.h"










22



#include "tcommon.h"










23



#include "tcompare.h"










24



#include "tdatablock.h"










25



#include "ttime.h"










26












27



static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** pRes);










28



static void    destroyEWindowOperatorInfo(void* param);










29



static int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock);










30



void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo);










31












32



static int32_t resetEventWindowOperState(SOperatorInfo* pOper) {




1,854✔





33



  SEventWindowOperatorInfo* pEvent = pOper->info;




1,854✔





34



  SExecTaskInfo*           pTaskInfo = pOper->pTaskInfo;




1,854✔





35



  SEventWinodwPhysiNode* pPhynode = (SEventWinodwPhysiNode*)pOper->pPhyNode;




1,854✔





36



  pOper->status = OP_NOT_OPENED;




1,854✔





37












38



  resetBasicOperatorState(&pEvent->binfo);




1,854✔





39



  cleanupResultInfoInEventWindow(pOper, pEvent);




1,854✔





40



  taosMemoryFreeClear(pEvent->pRow);




1,854✔





41












42



  pEvent->groupId = 0;




1,854✔





43



  pEvent->pPreDataBlock = NULL;




1,854✔





44



  pEvent->inWindow = false;




1,854✔





45



  pEvent->winSup.lastTs = INT64_MIN;




1,854✔





46



  resetIndefRowsRuntime(&pEvent->indefRows, pOper);




1,854✔





47












48



  colDataDestroy(&pEvent->twAggSup.timeWindowData);




1,854✔





49



  int32_t code = initExecTimeWindowInfo(&pEvent->twAggSup.timeWindowData, &pTaskInfo->window);




1,854✔





50












51



  if (code == 0) {




1,854✔





52



    code = resetAggSup(&pOper->exprSupp, &pEvent->aggSup, pTaskInfo, pPhynode->window.pFuncs, NULL,




3,708✔





53



                       sizeof(int64_t) * 2 + POINTER_BYTES, pTaskInfo->id.str, NULL,




1,854✔





54



                       &pTaskInfo->storageAPI.functionStore);










55



  }










56



  if (code == 0) {




1,854✔





57



    code = resetExprSupp(&pEvent->scalarSup, pTaskInfo, pPhynode->window.pExprs, NULL,




1,854✔





58



                         &pTaskInfo->storageAPI.functionStore);










59



  }










60



  return code;




1,854✔





61



}










62












63



int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,




355,324✔





64



                                             SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {










65



  QRY_PARAM_CHECK(pOptrInfo);




355,324✔





66












67



  int32_t                   code = TSDB_CODE_SUCCESS;




355,324✔





68



  int32_t                   lino = 0;




355,324✔





69



  SEventWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SEventWindowOperatorInfo));




355,324✔





70



  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));




355,324✔





71



  if (pInfo == NULL || pOperator == NULL) {




355,324✔






UNCOV

72



    code = terrno;




×







73



    goto _error;




×







74



  }










75



  initOperatorCostInfo(pOperator);




355,324✔





76












77



  pOperator->pPhyNode = physiNode;




355,324✔





78



  pOperator->exprSupp.hasWindowOrGroup = true;




355,324✔





79



  pOperator->exprSupp.hasWindow = true;




355,324✔





80












81



  SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;




355,324✔





82












83



  int32_t tsSlotId = ((SColumnNode*)pEventWindowNode->window.pTspk)->slotId;




355,324✔





84



  code = filterInitFromNode((SNode*)pEventWindowNode->pStartCond, &pInfo->pStartCondInfo, 0,




355,324✔





85



                            pTaskInfo->pStreamRuntimeInfo);




355,324✔





86



  QUERY_CHECK_CODE(code, lino, _error);




355,324✔





87



  filterSetExecContext(pInfo->pStartCondInfo, pTaskInfo, isTaskKilled);




355,324✔





88












89



  code = filterInitFromNode((SNode*)pEventWindowNode->pEndCond, &pInfo->pEndCondInfo, 0,




355,324✔





90



                            pTaskInfo->pStreamRuntimeInfo);




355,324✔





91



  QUERY_CHECK_CODE(code, lino, _error);




355,324✔





92



  filterSetExecContext(pInfo->pEndCondInfo, pTaskInfo, isTaskKilled);




355,324✔





93












94



  if (pEventWindowNode->window.pExprs != NULL) {




355,324✔





95



    int32_t    numOfScalarExpr = 0;




72,466✔





96



    SExprInfo* pScalarExprInfo = NULL;




72,466✔





97












98



    code = createExprInfo(pEventWindowNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);




72,466✔





99



    QUERY_CHECK_CODE(code, lino, _error);




72,466✔





100



    code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);




72,466✔





101



    QUERY_CHECK_CODE(code, lino, _error);




72,466✔





102



  }










103












104



  code = filterInitFromNode((SNode*)pEventWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,




355,324✔





105



                            pTaskInfo->pStreamRuntimeInfo);




355,324✔





106



  QUERY_CHECK_CODE(code, lino, _error);




355,324✔





107



  filterSetExecContext(pOperator->exprSupp.pFilterInfo, pTaskInfo, isTaskKilled);




355,324✔





108












109



  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;




355,324✔





110












111



  int32_t    num = 0;




355,324✔





112



  SExprInfo* pExprInfo = NULL;




355,324✔





113



  code = createExprInfo(pEventWindowNode->window.pFuncs, NULL, &pExprInfo, &num);




355,324✔





114



  QUERY_CHECK_CODE(code, lino, _error);




355,324✔





115












116



  initResultSizeInfo(&pOperator->resultInfo, 4096);




355,324✔





117












118



  code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,




355,324✔





119



                    NULL, &pTaskInfo->storageAPI.functionStore);










120



  QUERY_CHECK_CODE(code, lino, _error);




355,324✔





121












122



  pInfo->indefRowsMode = pEventWindowNode->window.indefRowsFunc;




355,324✔





123



  if (pInfo->indefRowsMode) {




355,324✔





124



    code = initIndefRowsRuntime(&pInfo->indefRows, pOperator->exprSupp.pCtx, num, pOperator->resultInfo.capacity);




10,968✔





125



    QUERY_CHECK_CODE(code, lino, _error);




10,968✔





126



  }










127












128



  SSDataBlock* pResBlock = createDataBlockFromDescNode(pEventWindowNode->window.node.pOutputDataBlockDesc);




355,324✔





129



  QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);




355,324✔





130



  initBasicInfo(&pInfo->binfo, pResBlock);




355,324✔





131












132



  code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);




355,324✔





133



  QUERY_CHECK_CODE(code, lino, _error);




355,324✔





134












135



  initResultRowInfo(&pInfo->binfo.resultRowInfo);




355,324✔





136



  pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;




355,324✔





137



  pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;




355,324✔





138



  pInfo->winSup.lastTs = INT64_MIN;




355,324✔





139












140



  code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);




355,324✔





141



  QUERY_CHECK_CODE(code, lino, _error);




355,324✔





142












143



  pInfo->tsSlotId = tsSlotId;




355,324✔





144



  pInfo->pPreDataBlock = NULL;




355,324✔





145



  pInfo->pOperator = pOperator;




355,324✔





146



  pInfo->trueForInfo.trueForType = pEventWindowNode->trueForType;




355,324✔





147



  pInfo->trueForInfo.count = pEventWindowNode->trueForCount;




355,324✔





148



  pInfo->trueForInfo.duration = pEventWindowNode->trueForDuration;




355,324✔





149












150



  setOperatorInfo(pOperator, "EventWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT, true, OP_NOT_OPENED, pInfo,




355,324✔





151



                  pTaskInfo);










152



  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, eventWindowAggregateNext, NULL, destroyEWindowOperatorInfo,




355,324✔





153



                                         optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);










154












155



  setOperatorResetStateFn(pOperator, resetEventWindowOperState);




355,324✔





156



  code = appendDownstream(pOperator, &downstream, 1);




355,324✔





157



  if (code != TSDB_CODE_SUCCESS) {




355,324✔





158



    goto _error;




×







159



  }










160












161



  *pOptrInfo = pOperator;




355,324✔





162



  return TSDB_CODE_SUCCESS;




355,324✔





163












164



_error:




×







165



  if (pInfo != NULL) {




×







166



    destroyEWindowOperatorInfo(pInfo);




×







167



  }










168












169



  destroyOperatorAndDownstreams(pOperator, &downstream, 1);




×







170



  pTaskInfo->code = code;




×







171



  return code;




×







172



}










173












174



void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo) {




357,178✔





175



  if (pInfo == NULL || pInfo->pRow == NULL || pOperator == NULL) {




357,178✔





176



    return;




108,956✔





177



  }










178



  SExprSupp*       pSup = &pOperator->exprSupp;




248,222✔





179



  for (int32_t j = 0; j < pSup->numOfExprs; ++j) {




248,222✔





180



    pSup->pCtx[j].resultInfo = getResultEntryInfo(pInfo->pRow, j, pSup->rowEntryInfoOffset);




×







181



    if (pSup->pCtx[j].fpSet.cleanup) {




×







182



      pSup->pCtx[j].fpSet.cleanup(&pSup->pCtx[j]);




×







183



    }










184



  }










185



}










186












187



void destroyEWindowOperatorInfo(void* param) {




355,324✔





188



  SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;




355,324✔





189



  if (pInfo == NULL) {




355,324✔





190



    return;




×







191



  }










192












193



  // First cleanup function contexts that may reference result buffers/state.










194



  // This must happen before freeing any buffers that those cleanups might touch.










195



  cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);




355,324✔





196



  cleanupIndefRowsRuntime(&pInfo->indefRows, pInfo->pOperator);




355,324✔





197












198



  if (pInfo->pRow != NULL) {




355,324✔





199



    taosMemoryFree(pInfo->pRow);




246,368✔





200



    pInfo->pRow = NULL;




246,368✔





201



  }










202












203



  if (pInfo->pStartCondInfo != NULL) {




355,324✔





204



    filterFreeInfo(pInfo->pStartCondInfo);




355,324✔





205



    pInfo->pStartCondInfo = NULL;




355,324✔





206



  }










207












208



  if (pInfo->pEndCondInfo != NULL) {




355,324✔





209



    filterFreeInfo(pInfo->pEndCondInfo);




355,324✔





210



    pInfo->pEndCondInfo = NULL;




355,324✔





211



  }










212












213



  cleanupBasicInfo(&pInfo->binfo);




355,324✔





214



  colDataDestroy(&pInfo->twAggSup.timeWindowData);




355,324✔





215



  pInfo->pOperator = NULL;




355,324✔





216



  cleanupAggSup(&pInfo->aggSup);




355,324✔





217



  cleanupExprSupp(&pInfo->scalarSup);




355,324✔





218



  taosMemoryFreeClear(param);




355,324✔





219



}










220












221



static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {




795,539✔





222



  int32_t                   code = TSDB_CODE_SUCCESS;




795,539✔





223



  int32_t                   lino = 0;




795,539✔





224



  SEventWindowOperatorInfo* pInfo = pOperator->info;




795,539✔





225



  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;




795,539✔





226












227



  if (pOperator->status == OP_EXEC_DONE) {




795,539✔





228



    *ppRes = NULL;




×







229



    return code;




×







230



  }










231












232



  SExprSupp* pSup = &pOperator->exprSupp;




795,539✔





233



  int32_t    order = pInfo->binfo.inputTsOrder;




795,539✔





234












235



  SSDataBlock* pRes = pInfo->binfo.pRes;




795,539✔





236












237



  if (pInfo->indefRowsMode) {




795,539✔





238



    (*ppRes) = getNextIndefRowsResultBlock(&pInfo->indefRows, pOperator);




39,024✔





239



    if ((*ppRes) != NULL) {




39,024✔





240



      return code;




6,720✔





241



    }










242



  }










243












244



  blockDataCleanup(pRes);




788,819✔





245












246



  SOperatorInfo* downstream = pOperator->pDownstream[0];




788,819✔





247



  while (1) {




3,399,782✔





248



    SSDataBlock* pBlock = NULL;




4,188,601✔





249



    if (pInfo->pPreDataBlock == NULL) {




4,188,601✔





250



      pBlock = getNextBlockFromDownstream(pOperator, 0);




3,864,153✔





251



    } else {










252



      pBlock = pInfo->pPreDataBlock;




324,448✔





253



      pInfo->pPreDataBlock = NULL;




324,448✔





254



    }










255












256



    if (pBlock == NULL) {




4,187,929✔





257



      break;




373,056✔





258



    }










259












260



    pRes->info.scanFlag = pBlock->info.scanFlag;




3,814,873✔





261



    pRes->info.dataLoad = 1;




3,814,873✔





262



    code = setInputDataBlock(pSup, pBlock, order, pBlock->info.scanFlag, true);




3,814,873✔





263



    QUERY_CHECK_CODE(code, lino, _end);




3,814,873✔





264












265



    code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);




3,814,873✔





266



    QUERY_CHECK_CODE(code, lino, _end);




3,814,873✔





267












268



    // there is an scalar expression that needs to be calculated right before apply the group aggregation.










269



    if (pInfo->scalarSup.pExprInfo != NULL) {




3,814,873✔





270



      code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,




126,298✔





271



                                   pInfo->scalarSup.numOfExprs, NULL, GET_STM_RTINFO(pOperator->pTaskInfo), pOperator->pTaskInfo);




63,149✔





272



      QUERY_CHECK_CODE(code, lino, _end);




63,149✔





273



    }










274












275



    code = eventWindowAggImpl(pOperator, pInfo, pBlock);




3,798,592✔





276



    QUERY_CHECK_CODE(code, lino, _end);




3,798,592✔





277












278



    if (pInfo->indefRowsMode) {




3,780,496✔





279



      (*ppRes) = getNextIndefRowsResultBlock(&pInfo->indefRows, pOperator);




33,144✔





280



      if ((*ppRes) != NULL) {




33,144✔





281



        return code;




21,336✔





282



      }










283



      continue;




11,808✔





284



    }










285












286



    code = doFilter(pRes, pSup->pFilterInfo, NULL, NULL);




3,747,352✔





287



    QUERY_CHECK_CODE(code, lino, _end);




3,747,352✔





288












289



    if (pRes->info.rows >= pOperator->resultInfo.threshold ||




3,747,352✔





290



        (pRes->info.id.groupId != pInfo->groupId && pRes->info.rows > 0)) {




3,669,556✔





291



      (*ppRes) = pRes;




359,378✔





292



      return code;




359,378✔





293



    }










294



  }










295












296



  if (pInfo->indefRowsMode) {




373,056✔





297



    dropAllIndefRowsWindowStates(pOperator, &pInfo->indefRows);




10,968✔





298



    pInfo->inWindow = false;




10,968✔





299



    (*ppRes) = getNextIndefRowsResultBlock(&pInfo->indefRows, pOperator);




10,968✔





300



    if ((*ppRes) == NULL) {




10,968✔





301



      setOperatorCompleted(pOperator);




10,968✔





302



    }










303



    return code;




10,968✔





304



  }










305












306



_end:




362,088✔





307



  if (code != TSDB_CODE_SUCCESS) {




396,465✔





308



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));




34,377✔





309



    pTaskInfo->code = code;




34,377✔





310



    T_LONG_JMP(pTaskInfo->env, code);




34,377✔





311



  }










312



  (*ppRes) =  pRes->info.rows == 0 ? NULL : pRes;




362,088✔





313



  return code;




362,088✔





314



}










315












316



static int32_t setSingleOutputTupleBufv1(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,




554,096,320✔





317



                                         SExprSupp* pExprSup, SAggSupporter* pAggSup) {










318



  if (*pResult == NULL) {




554,096,320✔





319



    SResultRow* p = taosMemoryCalloc(1, pAggSup->resultRowSize);




248,222✔





320



    if (!p) {




248,222✔





321



      return terrno;




×







322



    }










323



    pResultRowInfo->cur = (SResultRowPosition){.pageId = p->pageId, .offset = p->offset};




248,222✔





324



    *pResult = p;




248,222✔





325



  }










326












327



  (*pResult)->win = *win;




554,096,320✔





328












329



  return setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);




554,096,320✔





330



}










331












332



static int32_t doEventWindowAggImpl(SEventWindowOperatorInfo* pInfo, SExprSupp* pSup, int32_t startIndex,




554,124,376✔





333



                                    int32_t endIndex, const SSDataBlock* pBlock, int64_t* tsList,










334



                                    SExecTaskInfo* pTaskInfo) {










335



  int32_t code = TSDB_CODE_SUCCESS;




554,124,376✔





336



  SWindowRowsSup* pRowSup = &pInfo->winSup;




554,124,376✔





337












338



  int32_t numOfOutput = pSup->numOfExprs;




554,124,376✔





339



  int32_t numOfRows = endIndex - startIndex + 1;




554,124,376✔





340












341



  doKeepTuple(pRowSup, tsList[endIndex], endIndex, pBlock->info.id.groupId);




554,124,376✔





342












343



  if (pInfo->indefRowsMode) {




554,124,376✔





344



    SIndefRowsWindowState* pState = NULL;




28,056✔





345



    return applyIndefRowsFuncOnWindowState(pInfo->pOperator, &pInfo->indefRows, &pState, pInfo->binfo.pRes,




56,112✔





346



                                           pBlock->info.id.groupId, &pRowSup->win, (SSDataBlock*)pBlock, startIndex,




28,056✔





347



                                           numOfRows, pInfo->binfo.inputTsOrder, pInfo->aggSup.resultRowSize);










348



  }










349












350



  code = setSingleOutputTupleBufv1(&pInfo->binfo.resultRowInfo, &pRowSup->win, &pInfo->pRow, pSup, &pInfo->aggSup);




554,096,320✔





351



  if (code != TSDB_CODE_SUCCESS) {  // null data, too many state code




554,096,320✔





352



    qError("failed to set single output tuple buffer, code:%d", code);




×







353



    return code;




×







354



  }










355












356



  updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, 0);




554,096,320✔





357



  pInfo->pRow->nOrigRows += numOfRows;




554,096,320✔





358



  code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startIndex, numOfRows,




554,096,320✔





359



                                         pBlock->info.rows, numOfOutput);




554,096,320✔





360



  return code;




554,096,320✔





361



}










362












363



int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {




3,798,592✔





364



  int32_t          code = TSDB_CODE_SUCCESS;




3,798,592✔





365



  int32_t          lino = 0;




3,798,592✔





366



  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;




3,798,592✔





367



  SExprSupp*       pSup = &pOperator->exprSupp;




3,798,592✔





368



  SSDataBlock*     pRes = pInfo->binfo.pRes;




3,798,592✔





369



  int64_t          gid = pBlock->info.id.groupId;




3,798,592✔





370



  SColumnInfoData *ps = NULL, *pe = NULL;




3,798,592✔





371



  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);




3,798,592✔





372



  QUERY_CHECK_NULL(pColInfoData, code, lino, _return, terrno);




3,798,592✔





373



  TSKEY*           tsList = (TSKEY*)pColInfoData->pData;




3,798,592✔





374



  SWindowRowsSup*  pRowSup = &pInfo->winSup;




3,798,592✔





375



  int32_t          rowIndex = 0;




3,798,592✔





376



  STrueForInfo*    pTrueForInfo = getTrueForInfo(pOperator);




3,798,592✔





377












378



  pRowSup->numOfRows = 0;




3,798,592✔





379



  if (pInfo->groupId == 0) {




3,798,592✔





380



    pInfo->groupId = gid;




3,117,442✔





381



  } else if (pInfo->groupId != gid) {




681,150✔





382



    // this is a new group, reset the info










383



    if (pInfo->indefRowsMode) {




345,992✔





384



      dropAllIndefRowsWindowStates(pOperator, &pInfo->indefRows);




11,328✔





385



    }










386



    pInfo->inWindow = false;




345,992✔





387



    pInfo->groupId = gid;




345,992✔





388



    pInfo->winSup.lastTs = INT64_MIN;




345,992✔





389



    pInfo->pPreDataBlock = pBlock;




345,992✔





390



    goto _return;




345,992✔





391



  }










392



  pRes->info.id.groupId = pInfo->groupId;




3,452,600✔





393












394



  SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};




3,452,600✔





395












396



  code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &param1);




3,452,600✔





397



  QUERY_CHECK_CODE(code, lino, _return);




3,452,600✔





398












399



  int32_t status1 = 0;




3,452,600✔





400



  code = filterExecute(pInfo->pStartCondInfo, pBlock, &ps, NULL, param1.numOfCols, &status1);




3,452,600✔





401



  QUERY_CHECK_CODE(code, lino, _return);




3,452,600✔





402












403



  SFilterColumnParam param2 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};




3,452,600✔





404



  code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &param2);




3,452,600✔





405



  QUERY_CHECK_CODE(code, lino, _return);




3,452,600✔





406












407



  int32_t status2 = 0;




3,452,600✔





408



  code = filterExecute(pInfo->pEndCondInfo, pBlock, &pe, NULL, param2.numOfCols, &status2);




3,452,600✔





409



  QUERY_CHECK_CODE(code, lino, _return);




3,452,600✔





410












411



  for (int32_t i = 0; i < pBlock->info.rows; ++i) {




2,147,483,647✔





412



    if (pBlock->info.scanFlag != PRE_SCAN) {




2,147,483,647✔





413



      if (pInfo->winSup.lastTs == INT64_MIN) {




2,147,483,647✔





414



        pInfo->winSup.lastTs = tsList[i];




621,864✔





415



      } else {










416



        if (tsList[i] == pInfo->winSup.lastTs) {




2,147,483,647✔





417



          qError("duplicate timestamp found in event window operator, groupId: %" PRId64 ", timestamp: %" PRId64,




18,096✔





418



                 gid, tsList[i]);










419



          code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;




18,096✔





420



          QUERY_CHECK_CODE(code, lino, _return);




18,096✔





421



        } else {










422



          pInfo->winSup.lastTs = tsList[i];




2,147,483,647✔





423



        }










424



      }










425



    }










426



  }










427



  int32_t startIndex = pInfo->inWindow ? 0 : -1;




3,434,504✔





428



  while (rowIndex < pBlock->info.rows) {




1,109,693,146✔





429



    if (pInfo->inWindow) {  // let's find the first end value




1,106,950,568✔





430



      for (rowIndex = startIndex; rowIndex < pBlock->info.rows; ++rowIndex) {




2,147,483,647✔





431



        if (((bool*)pe->pData)[rowIndex]) {




2,147,483,647✔





432



          break;




551,867,592✔





433



        }










434



      }










435












436



      if (rowIndex < pBlock->info.rows) {




554,124,376✔





437



        code = doEventWindowAggImpl(pInfo, pSup, startIndex, rowIndex, pBlock, tsList, pTaskInfo);




551,867,592✔





438



        QUERY_CHECK_CODE(code, lino, _return);




551,867,592✔





439



        if (pInfo->indefRowsMode) {




551,867,592✔





440



          SIndefRowsWindowState* pState = findIndefRowsWindowState(&pInfo->indefRows, pInfo->groupId, pRowSup->win.skey);




27,576✔





441



          if (pState == NULL) {




27,576✔





442



            code = TSDB_CODE_QRY_WINDOW_STATE_NOT_EXIST;




×







443



            QUERY_CHECK_CODE(code, lino, _return);




×







444



          }










445












446



          if (!isTrueForSatisfied(pTrueForInfo, pState->win.skey, pState->win.ekey, pState->pRow->nOrigRows)) {




27,576✔





447



            qDebug("skip small window, groupId: %" PRId64 ", skey: %" PRId64 ", ekey: %" PRId64 ", nrows: %u",




×







448



                   pInfo->groupId, pState->win.skey, pState->win.ekey, pState->pRow->nOrigRows);










449



            dropIndefRowsWindowState(pOperator, &pInfo->indefRows, pState);




×







450



          } else {










451



            code = closeIndefRowsWindowState(pOperator, &pInfo->indefRows, pState);




27,576✔





452



            QUERY_CHECK_CODE(code, lino, _return);




27,576✔





453



          }










454



        } else {










455



          doUpdateNumOfRows(pSup->pCtx, pInfo->pRow, pSup->numOfExprs, pSup->rowEntryInfoOffset);




551,840,016✔





456












457



          if (!isTrueForSatisfied(pTrueForInfo, pRowSup->win.skey, pRowSup->win.ekey, pInfo->pRow->nOrigRows)) {




551,840,016✔





458



            qDebug("skip small window, groupId: %" PRId64 ", skey: %" PRId64 ", ekey: %" PRId64 ", nrows: %u",




151,139,100✔





459



                   pInfo->groupId, pRowSup->win.skey, pRowSup->win.ekey, pInfo->pRow->nOrigRows);










460



          } else {










461



            // check buffer size










462



            if (pRes->info.rows + pInfo->pRow->numOfRows >= pRes->info.capacity) {




400,700,916✔





463



              int32_t newSize = pRes->info.rows + pInfo->pRow->numOfRows;




775,584✔





464



              code = blockDataEnsureCapacity(pRes, newSize);




775,584✔





465



              QUERY_CHECK_CODE(code, lino, _return);




775,584✔





466



            }










467












468



            code = copyResultrowToDataBlock(pSup->pExprInfo, pSup->numOfExprs, pInfo->pRow, pSup->pCtx, pRes,




400,700,916✔





469



                                            pSup->rowEntryInfoOffset, pTaskInfo);




400,700,916✔





470



            QUERY_CHECK_CODE(code, lino, _return);




400,700,916✔





471












472



            pRes->info.rows += pInfo->pRow->numOfRows;




400,700,916✔





473



          }










474



          pInfo->pRow->numOfRows = 0;




551,840,016✔





475



          pInfo->pRow->nOrigRows = 0;




551,840,016✔





476



        }










477












478



        pInfo->inWindow = false;




551,867,592✔





479



        rowIndex += 1;




551,867,592✔





480



      } else {










481



        code = doEventWindowAggImpl(pInfo, pSup, startIndex, pBlock->info.rows - 1, pBlock, tsList, pTaskInfo);




2,256,784✔





482



        QUERY_CHECK_CODE(code, lino, _return);




2,256,784✔





483



      }










484



    } else {  // find the first start value that is fulfill for the start condition










485



      for (; rowIndex < pBlock->info.rows; ++rowIndex) {




1,500,912,740✔





486



        if (((bool*)ps->pData)[rowIndex]) {




1,500,220,814✔





487



          doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);




552,134,266✔





488



          pInfo->inWindow = true;




552,134,266✔





489



          startIndex = rowIndex;




552,134,266✔





490



          if (!pInfo->indefRowsMode && pInfo->pRow != NULL) {




552,134,266✔





491



            clearResultRowInitFlag(pSup->pCtx, pSup->numOfExprs);




551,858,468✔





492



          }










493



          break;




552,134,266✔





494



        }










495



      }










496












497



      if (pInfo->inWindow) {




552,826,192✔





498



        continue;  // try to find the end position




552,134,266✔





499



      } else {










500



        break;  // no valid start position, quit




691,926✔





501



      }










502



    }










503



  }










504












505



_return:




3,798,592✔





506












507



  if (code != TSDB_CODE_SUCCESS) {




3,798,592✔





508



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));




18,096✔





509



  }










510



  colDataDestroy(ps);




3,798,592✔





511



  taosMemoryFree(ps);




3,798,592✔





512



  colDataDestroy(pe);




3,798,592✔





513



  taosMemoryFree(pe);




3,798,592✔





514












515



  return code;




3,798,592✔





516



}
























    

  • 

    •