// GetTableRecordsCount returns the total number of table records from startHandle.
// If startHandle = 0, returns the total number of table records.
func GetTableRecordsCount(txn kv.Transaction, t table.Table, startHandle int64) (int64, error) {
	startKey := t.RecordKey(startHandle, nil)
	it, err := txn.Seek(startKey)
	if err != nil {
		return 0, errors.Trace(err)
	}

	var cnt int64
	prefix := t.RecordPrefix()
	for it.Valid() && it.Key().HasPrefix(prefix) {
		handle, err := tables.DecodeRecordKeyHandle(it.Key())
		if err != nil {
			return 0, errors.Trace(err)
		}

		it.Close()
		rk := t.RecordKey(handle+1, nil)
		it, err = txn.Seek(rk)
		if err != nil {
			return 0, errors.Trace(err)
		}

		cnt++
	}

	it.Close()

	return cnt, nil
}

// Next implements plan.Plan Next interface.
func (r *TableNilPlan) Next(ctx context.Context) (row *plan.Row, err error) {
	if r.iter == nil {
		var txn kv.Transaction
		txn, err = ctx.GetTxn(false)
		if err != nil {
			return nil, errors.Trace(err)
		}
		r.iter, err = txn.Seek(r.T.FirstKey())
		if err != nil {
			return nil, errors.Trace(err)
		}
	}
	if !r.iter.Valid() || !r.iter.Key().HasPrefix(r.T.RecordPrefix()) {
		return
	}
	handle, err := tables.DecodeRecordKeyHandle(r.iter.Key())
	if err != nil {
		return nil, errors.Trace(err)
	}
	rk := r.T.RecordKey(handle, nil)
	// Even though the data is nil, we should return not nil row,
	// or the iteration will stop.
	row = &plan.Row{}
	err = kv.NextUntil(r.iter, util.RowKeyPrefixFilter(rk))
	return
}

func (ts *testSuite) TestRowKeyCodec(c *C) {
	table := []struct {
		tableID int64
		h       int64
		ID      int64
	}{
		{1, 1234567890, 0},
		{2, 1, 0},
		{3, -1, 0},
		{4, -1, 1},
	}

	for _, t := range table {
		b := tables.EncodeRecordKey(t.tableID, t.h, t.ID)
		tableID, handle, columnID, err := tables.DecodeRecordKey(b)
		c.Assert(err, IsNil)
		c.Assert(tableID, Equals, t.tableID)
		c.Assert(handle, Equals, t.h)
		c.Assert(columnID, Equals, t.ID)

		handle, err = tables.DecodeRecordKeyHandle(b)
		c.Assert(err, IsNil)
		c.Assert(handle, Equals, t.h)
	}

	// test error
	tbl := []string{
		"",
		"x",
		"t1",
		"t12345678",
		"t12345678_i",
		"t12345678_r1",
		"t12345678_r1234567",
		"t12345678_r123456781",
	}

	for _, t := range tbl {
		_, err := tables.DecodeRecordKeyHandle(kv.Key(t))
		c.Assert(err, NotNil)
	}
}

func (d *ddl) getSnapshotRows(t table.Table, version uint64, seekHandle int64) ([]int64, error) {
	ver := kv.Version{Ver: version}

	snap, err := d.store.GetSnapshot(ver)
	if err != nil {
		return nil, errors.Trace(err)
	}

	defer snap.Release()

	firstKey := t.RecordKey(seekHandle, nil)

	it, err := snap.Seek(firstKey)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer it.Close()

	handles := make([]int64, 0, maxBatchSize)

	for it.Valid() {
		if !it.Key().HasPrefix(t.RecordPrefix()) {
			break
		}

		var handle int64
		handle, err = tables.DecodeRecordKeyHandle(it.Key())
		if err != nil {
			return nil, errors.Trace(err)
		}

		rk := t.RecordKey(handle, nil)

		handles = append(handles, handle)
		if len(handles) == maxBatchSize {
			seekHandle = handle + 1
			break
		}

		err = kv.NextUntil(it, util.RowKeyPrefixFilter(rk))
		if terror.ErrorEqual(err, kv.ErrNotExist) {
			break
		} else if err != nil {
			return nil, errors.Trace(err)
		}
	}

	return handles, nil
}

// Next implements Executor Next interface.
func (e *UpdateExec) Next() (*Row, error) {
	if !e.fetched {
		err := e.fetchRows()
		if err != nil {
			return nil, errors.Trace(err)
		}
		e.fetched = true
	}
	if e.cursor >= len(e.rows) {
		return nil, nil
	}
	if e.updatedRowKeys == nil {
		e.updatedRowKeys = map[string]bool{}
	}
	row := e.rows[e.cursor]
	newData := e.newRowsData[e.cursor]
	for _, entry := range row.RowKeys {
		tbl := entry.Tbl
		offset := e.getTableOffset(tbl)
		k := entry.Key
		oldData := row.Data[offset : offset+len(tbl.Cols())]
		newTableData := newData[offset : offset+len(tbl.Cols())]

		_, ok := e.updatedRowKeys[k]
		if ok {
			// Each matching row is updated once, even if it matches the conditions multiple times.
			continue
		}

		// Update row
		handle, err1 := tables.DecodeRecordKeyHandle(kv.Key(k))
		if err1 != nil {
			return nil, errors.Trace(err1)
		}

		err1 = e.updateRecord(handle, oldData, newTableData, tbl, offset, false)
		if err1 != nil {
			return nil, errors.Trace(err1)
		}
		e.updatedRowKeys[k] = true
	}
	e.cursor++
	return &Row{}, nil
}

func iterRecords(retriever kv.Retriever, t table.Table, startKey kv.Key, cols []*table.Column,
	fn table.RecordIterFunc) error {
	it, err := retriever.Seek(startKey)
	if err != nil {
		return errors.Trace(err)
	}
	defer it.Close()

	if !it.Valid() {
		return nil
	}

	log.Debugf("startKey:%q, key:%q, value:%q", startKey, it.Key(), it.Value())

	prefix := t.RecordPrefix()
	for it.Valid() && it.Key().HasPrefix(prefix) {
		// first kv pair is row lock information.
		// TODO: check valid lock
		// get row handle
		handle, err := tables.DecodeRecordKeyHandle(it.Key())
		if err != nil {
			return errors.Trace(err)
		}

		data, err := rowWithCols(retriever, t, handle, cols)
		if err != nil {
			return errors.Trace(err)
		}
		more, err := fn(handle, data, cols)
		if !more || err != nil {
			return errors.Trace(err)
		}

		rk := t.RecordKey(handle, nil)
		err = kv.NextUntil(it, util.RowKeyPrefixFilter(rk))
		if err != nil {
			return errors.Trace(err)
		}
	}

	return nil
}

// Next implements Execution Next interface.
func (e *TableScanExec) Next() (*Row, error) {
	for {
		if e.cursor >= len(e.ranges) {
			return nil, nil
		}
		ran := e.ranges[e.cursor]
		if e.seekHandle < ran.LowVal {
			e.seekHandle = ran.LowVal
		}
		if e.seekHandle > ran.HighVal {
			e.cursor++
			continue
		}
		rowKey, err := e.seek()
		if err != nil || rowKey == nil {
			return nil, errors.Trace(err)
		}
		handle, err := tables.DecodeRecordKeyHandle(rowKey)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if handle > ran.HighVal {
			// The handle is out of the current range, but may be in following ranges.
			// We seek to the range that may contains the handle, so we
			// don't need to seek key again.
			inRange := e.seekRange(handle)
			if !inRange {
				// The handle may be less than the current range low value, can not
				// return directly.
				continue
			}
		}
		row, err := e.getRow(handle, rowKey)
		if err != nil {
			return nil, errors.Trace(err)
		}
		e.seekHandle = handle + 1
		return row, nil
	}
}

func (r *TableDefaultPlan) rangeNext(ctx context.Context) (*plan.Row, error) {
	for {
		if r.cursor == len(r.spans) {
			return nil, nil
		}
		span := r.spans[r.cursor]
		if r.seekKey == nil {
			seekVal := span.seekVal
			var err error
			r.seekKey, err = r.toSeekKey(seekVal)
			if err != nil {
				return nil, errors.Trace(err)
			}
		}
		txn, err := ctx.GetTxn(false)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if r.iter != nil {
			r.iter.Close()
		}
		r.iter, err = txn.Seek(r.seekKey)
		if err != nil {
			return nil, types.EOFAsNil(err)
		}

		if !r.iter.Valid() || !r.iter.Key().HasPrefix(r.T.RecordPrefix()) {
			r.seekKey = nil
			r.cursor++
			r.skipLowCmp = false
			continue
		}
		rowKey := r.iter.Key()
		handle, err := tables.DecodeRecordKeyHandle(rowKey)
		if err != nil {
			return nil, errors.Trace(err)
		}
		r.seekKey, err = r.toSeekKey(handle + 1)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !r.skipLowCmp {
			cmp := indexCompare(handle, span.lowVal)
			if cmp < 0 || (cmp == 0 && span.lowExclude) {
				continue
			}
			r.skipLowCmp = true
		}
		cmp := indexCompare(handle, span.highVal)
		if cmp > 0 || (cmp == 0 && span.highExclude) {
			// This span has finished iteration.
			// Move to the next span.
			r.seekKey = nil
			r.cursor++
			r.skipLowCmp = false
			continue
		}
		row := &plan.Row{}
		row.Data, err = r.T.Row(ctx, handle)
		if err != nil {
			return nil, errors.Trace(err)
		}
		// Put rowKey to the tail of record row
		rke := &plan.RowKeyEntry{
			Tbl: r.T,
			Key: string(rowKey),
		}
		row.RowKeys = append(row.RowKeys, rke)
		return row, nil
	}
}

// Exec implements the stmt.Statement Exec interface.
func (s *UpdateStmt) Exec(ctx context.Context) (_ rset.Recordset, err error) {
	p, err := s.plan(ctx)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer p.Close()

	fs := p.GetFields()
	columns, err := getUpdateColumns(s.List, fs)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var records []*plan.Row
	for {
		row, err1 := p.Next(ctx)
		if err1 != nil {
			return nil, errors.Trace(err1)
		}
		if row == nil {
			break
		}
		if len(row.RowKeys) == 0 {
			// Nothing to update
			continue
		}
		records = append(records, row)
	}

	evalMap := map[interface{}]interface{}{}
	updatedRowKeys := make(map[string]bool)
	for _, row := range records {
		rowData := row.Data

		// Set ExprEvalIdentReferFunc.
		evalMap[expression.ExprEvalIdentReferFunc] = func(name string, scope int, index int) (interface{}, error) {
			return rowData[index], nil
		}

		// Update rows.
		offset := 0
		for _, entry := range row.RowKeys {
			tbl := entry.Tbl
			k := entry.Key
			lastOffset := offset
			offset += len(tbl.Cols())
			data := rowData[lastOffset:offset]

			_, ok := updatedRowKeys[k]
			if ok {
				// Each matching row is updated once, even if it matches the conditions multiple times.
				continue
			}

			// Update row
			handle, err1 := tables.DecodeRecordKeyHandle(kv.Key(k))
			if err1 != nil {
				return nil, errors.Trace(err1)
			}

			err1 = updateRecord(ctx, handle, data, tbl, columns, evalMap, lastOffset, false)
			if err1 != nil {
				return nil, errors.Trace(err1)
			}

			updatedRowKeys[k] = true
		}
	}
	return nil, nil
}

// Exec implements the stmt.Statement Exec interface.
func (s *DeleteStmt) Exec(ctx context.Context) (_ rset.Recordset, err error) {
	if s.MultiTable && len(s.TableIdents) == 0 {
		return nil, nil
	}
	p, err := s.plan(ctx)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if p == nil {
		return nil, nil
	}
	defer p.Close()
	tblIDMap := make(map[int64]bool, len(s.TableIdents))
	// Get table alias map.
	tblNames := make(map[string]string)
	if s.MultiTable {
		// Delete from multiple tables should consider table ident list.
		fs := p.GetFields()
		for _, f := range fs {
			if f.TableName != f.OrgTableName {
				tblNames[f.TableName] = f.OrgTableName
			} else {
				tblNames[f.TableName] = f.TableName
			}
		}
		for _, t := range s.TableIdents {
			// Consider DBName.
			oname, ok := tblNames[t.Name.O]
			if !ok {
				return nil, errors.Errorf("Unknown table '%s' in MULTI DELETE", t.Name.O)
			}

			t.Name.O = oname
			t.Name.L = strings.ToLower(oname)

			var tbl table.Table
			tbl, err = getTable(ctx, t)
			if err != nil {
				return nil, errors.Trace(err)
			}
			tblIDMap[tbl.TableID()] = true
		}
	}
	rowKeyMap := make(map[string]table.Table)
	for {
		row, err1 := p.Next(ctx)
		if err1 != nil {
			return nil, errors.Trace(err1)
		}
		if row == nil {
			break
		}

		for _, entry := range row.RowKeys {
			if s.MultiTable {
				tid := entry.Tbl.TableID()
				if _, ok := tblIDMap[tid]; !ok {
					continue
				}
			}
			rowKeyMap[entry.Key] = entry.Tbl
		}
	}

	for k, t := range rowKeyMap {
		handle, err := tables.DecodeRecordKeyHandle(kv.Key(k))
		if err != nil {
			return nil, errors.Trace(err)
		}
		data, err := t.Row(ctx, handle)
		if err != nil {
			return nil, errors.Trace(err)
		}
		err = removeRow(ctx, t, handle, data)
		if err != nil {
			return nil, errors.Trace(err)
		}
	}
	return nil, nil
}