executor: fix wrong partition boundary for window functions (#11637)

executor/window.go

@@ -62,7 +62,7 @@ func (e *WindowExec) Next(ctx context.Context, chk *chunk.Chunk) error {
  defer func() { e.runtimeStats.Record(time.Now().Sub(start), chk.NumRows()) }()
  }
  chk.Reset()
- if e.meetNewGroup && e.remainingRowsInGroup > 0 {
+ if (e.executed || e.meetNewGroup) && e.remainingRowsInGroup > 0 {
  err := e.appendResult2Chunk(chk)
  if err != nil {
  return err
@@ -88,22 +88,16 @@ func (e *WindowExec) consumeOneGroup(ctx context.Context, chk *chunk.Chunk) erro
  if err != nil {
  return errors.Trace(err)
  }
- if e.meetNewGroup {
+ if e.meetNewGroup && e.remainingRowsInGroup > 0 {
  err := e.consumeGroupRows()
  if err != nil {
  return errors.Trace(err)
  }
  err = e.appendResult2Chunk(chk)
- if err != nil {
- return errors.Trace(err)
- }
+ return err
  }
  e.remainingRowsInGroup++
  e.groupRows = append(e.groupRows, e.inputRow)
- if e.meetNewGroup {
- e.inputRow = e.inputIter.Next()
- return nil
- }
  }
  return nil
 }

executor/window_test.go

@@ -166,4 +166,8 @@ func (s *testSuite4) TestWindowFunctions(c *C) {
  result.Check(testkit.Rows("1 3", "2 3", "3 6", "4 6"))
  result = tk.MustQuery("select row_number() over w, sum(b) over w from t window w as (rows between 1 preceding and 1 following)")
  result.Check(testkit.Rows("1 3", "2 4", "3 5", "4 3"))
+
+ tk.Se.GetSessionVars().MaxChunkSize = 1
+ result = tk.MustQuery("select a, row_number() over (partition by a) from t")
+ result.Check(testkit.Rows("1 1", "1 2", "2 1", "2 2"))
 }

util/chunk/chunk.go

@@ -77,24 +77,30 @@ func New(fields []*types.FieldType, cap, maxChunkSize int) *Chunk {
  return chk
 }
 
-// Renew creates a new Chunk based on an existing Chunk. The newly created Chunk
-// has the same data schema with the old Chunk. The capacity of the new Chunk
-// might be doubled based on the capacity of the old Chunk and the maxChunkSize.
-// chk: old chunk(often used in previous call).
-// maxChunkSize: the limit for the max number of rows.
-func Renew(chk *Chunk, maxChunkSize int) *Chunk {
+// renewWithCapacity creates a new Chunk based on an existing Chunk with capacity. The newly
+// created Chunk has the same data schema with the old Chunk.
+func renewWithCapacity(chk *Chunk, cap, maxChunkSize int) *Chunk {
  newChk := new(Chunk)
  if chk.columns == nil {
  return newChk
  }
- newCap := reCalcCapacity(chk, maxChunkSize)
- newChk.columns = renewColumns(chk.columns, newCap)
+ newChk.columns = renewColumns(chk.columns, cap)
  newChk.numVirtualRows = 0
- newChk.capacity = newCap
+ newChk.capacity = cap
  newChk.requiredRows = maxChunkSize
  return newChk
 }
 
+// Renew creates a new Chunk based on an existing Chunk. The newly created Chunk
+// has the same data schema with the old Chunk. The capacity of the new Chunk
+// might be doubled based on the capacity of the old Chunk and the maxChunkSize.
+// chk: old chunk(often used in previous call).
+// maxChunkSize: the limit for the max number of rows.
+func Renew(chk *Chunk, maxChunkSize int) *Chunk {
+ newCap := reCalcCapacity(chk, maxChunkSize)
+ return renewWithCapacity(chk, newCap, maxChunkSize)
+}
+
 // renewColumns creates the columns of a Chunk. The capacity of the newly
 // created columns is equal to cap.
 func renewColumns(oldCol []*column, cap int) []*column {

util/chunk/row.go

@@ -222,3 +222,10 @@ func (r Row) GetDatum(colIdx int, tp *types.FieldType) types.Datum {
 func (r Row) IsNull(colIdx int) bool {
  return r.c.columns[colIdx].isNull(r.idx)
 }
+
+// CopyConstruct creates a new row and copies this row's data into it.
+func (r Row) CopyConstruct() Row {
+ newChk := renewWithCapacity(r.c, 1, 1)
+ newChk.AppendRow(r)
+ return newChk.GetRow(0)
+}