pkg/executor,tests: reset params in the PipelineWindow.Open (#54709)

close #53600

Author: XuHuaiyu

pkg/executor/builder.go

@@ -4959,10 +4959,15 @@ func (b *executorBuilder) buildWindow(v *plannercore.PhysicalWindow) exec.Execut
 			BaseExecutor:   base,
 			groupChecker:   vecgroupchecker.NewVecGroupChecker(b.ctx.GetExprCtx().GetEvalCtx(), b.ctx.GetSessionVars().EnableVectorizedExpression, groupByItems),
 			numWindowFuncs: len(v.WindowFuncDescs),
+			windowFuncs:    windowFuncs,
+			partialResults: partialResults,
+		}
+		exec.slidingWindowFuncs = make([]aggfuncs.SlidingWindowAggFunc, len(exec.windowFuncs))
+		for i, windowFunc := range exec.windowFuncs {
+			if slidingWindowAggFunc, ok := windowFunc.(aggfuncs.SlidingWindowAggFunc); ok {
+				exec.slidingWindowFuncs[i] = slidingWindowAggFunc
+			}
 		}
-
-		exec.windowFuncs = windowFuncs
-		exec.partialResults = partialResults
 		if v.Frame == nil {
 			exec.start = &logicalop.FrameBound{
 				Type:      ast.Preceding,

pkg/executor/pipelined_window.go

@@ -84,19 +84,10 @@ func (e *PipelinedWindowExec) Close() error {
 
 // Open implements the Executor Open interface
 func (e *PipelinedWindowExec) Open(ctx context.Context) (err error) {
-	e.rowToConsume = 0
-	e.done = false
-	e.accumulated = 0
-	e.dropped = 0
-	e.data = make([]dataInfo, 0)
-	e.dataIdx = 0
-	e.slidingWindowFuncs = make([]aggfuncs.SlidingWindowAggFunc, len(e.windowFuncs))
-	for i, windowFunc := range e.windowFuncs {
-		if slidingWindowAggFunc, ok := windowFunc.(aggfuncs.SlidingWindowAggFunc); ok {
-			e.slidingWindowFuncs[i] = slidingWindowAggFunc
-		}
-	}
-	e.rows = make([]chunk.Row, 0)
+	e.done, e.newPartition, e.whole, e.initializedSlidingWindow = false, false, false, false
+	e.dataIdx, e.curRowIdx, e.dropped, e.rowToConsume, e.accumulated = 0, 0, 0, 0, 0
+	e.lastStartRow, e.lastEndRow, e.stagedStartRow, e.stagedEndRow, e.rowStart, e.rowCnt = 0, 0, 0, 0, 0, 0
+	e.rows, e.data = make([]chunk.Row, 0), make([]dataInfo, 0)
 	return e.BaseExecutor.Open(ctx)
 }