diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index ebfbacaa9e..287d80759e 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -2325,9 +2325,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } } @@ -2337,6 +2334,7 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { } if (ts == pSliceInfo->current) { + blockDataEnsureCapacity(pResBlock, pResBlock->info.rows + 1); for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) { SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j]; @@ -2377,9 +2375,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } if (pSliceInfo->current > pSliceInfo->win.ekey) { @@ -2398,10 +2393,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { doSetOperatorCompleted(pOperator); break; } - - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } } else if (ts < pSliceInfo->current) { // in case of interpolation window starts and ends between two datapoints, fill(prev) need to interpolate @@ -2419,9 +2410,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } if (pSliceInfo->current > pSliceInfo->win.ekey) { @@ -2443,9 +2431,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } if (pSliceInfo->current > pSliceInfo->win.ekey) { @@ -2467,13 +2452,11 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } // add current row if timestamp match if (ts == pSliceInfo->current && pSliceInfo->current <= pSliceInfo->win.ekey) { + blockDataEnsureCapacity(pResBlock, pResBlock->info.rows + 1); for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) { SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j]; @@ -2510,9 +2493,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } if (pSliceInfo->current > pSliceInfo->win.ekey) { @@ -2528,9 +2508,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } } @@ -2549,9 +2526,6 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pResBlock); pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision); - if (pResBlock->info.rows >= pResBlock->info.capacity) { - break; - } } // restore the value