Skip to content

Commit

Permalink
pkg/executor,tests: reset the parameters in the Open method of pipeli…
Browse files Browse the repository at this point in the history
…nedwindow to avoid unexpected errors caused by repeated open and close operations when pipelinedwindow is used as a child node of apply, which prevents reusing previous parameter values.
  • Loading branch information
XuHuaiyu committed Jul 18, 2024
1 parent 26378cb commit ac14840
Show file tree
Hide file tree
Showing 4 changed files with 1,043 additions and 16 deletions.
11 changes: 8 additions & 3 deletions pkg/executor/builder.go
Original file line number Diff line number Diff line change
Expand Up @@ -4913,10 +4913,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 = &plannercore.FrameBound{
Type: ast.Preceding,
Expand Down
17 changes: 4 additions & 13 deletions pkg/executor/pipelined_window.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}

Expand Down
Loading

0 comments on commit ac14840

Please sign in to comment.