// Copyright 2016 PingCAP, Inc.

//

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// See the License for the specific language governing permissions and

// limitations under the License.

package executor

import (

"bytes"

"fmt"

"strings"

"github.com/juju/errors"

"github.com/pingcap/tidb/ast"

"github.com/pingcap/tidb/context"

"github.com/pingcap/tidb/expression"

"github.com/pingcap/tidb/kv"

"github.com/pingcap/tidb/mysql"

"github.com/pingcap/tidb/table"

"github.com/pingcap/tidb/table/tables"

"github.com/pingcap/tidb/types"

"github.com/pingcap/tidb/util/chunk"

log "github.com/sirupsen/logrus"

goctx "golang.org/x/net/context"

)

var (

_ Executor = &UpdateExec{}

_ Executor = &DeleteExec{}

_ Executor = &InsertExec{}

_ Executor = &ReplaceExec{}

_ Executor = &LoadData{}

)

// updateRecord updates the row specified by the handle `h`, from `oldData` to `newData`.

// `modified` means which columns are really modified. It's used for secondary indices.

// Length of `oldData` and `newData` equals to length of `t.WritableCols()`.

// ignoreErr indicate that update statement has the `IGNORE` modifier, in this situation, update statement will not update

// the keys which cause duplicate conflicts and ignore the error.

func updateRecord(ctx context.Context, h int64, oldData, newData []types.Datum, modified []bool, t table.Table, onDup, ignoreErr bool) (bool, error) {

var sc = ctx.GetSessionVars().StmtCtx

var changed, handleChanged = false, false

// onUpdateSpecified is for "UPDATE SET ts_field = old_value", the

// timestamp field is explicitly set, but not changed in fact.

var onUpdateSpecified = make(map[int]bool)

var newHandle int64

// We can iterate on public columns not writable columns,

// because all of them are sorted by their `Offset`, which

// causes all writable columns are after public columns.

for i, col := range t.Cols() {

if modified[i] {

// Cast changed fields with respective columns.

v, err := table.CastValue(ctx, newData[i], col.ToInfo())

if err != nil {

return false, errors.Trace(err)

}

newData[i] = v

}

// Rebase auto increment id if the field is changed.

if mysql.HasAutoIncrementFlag(col.Flag) {

if newData[i].IsNull() {

return false, errors.Errorf("Column '%v' cannot be null", col.Name.O)

}

val, errTI := newData[i].ToInt64(sc)

if errTI != nil {

return false, errors.Trace(errTI)

}

err := t.RebaseAutoID(ctx, val, true)

if err != nil {

return false, errors.Trace(err)

}

}

cmp, err := newData[i].CompareDatum(sc, &oldData[i])

if err != nil {

return false, errors.Trace(err)

}

if cmp != 0 {

changed = true

modified[i] = true

if col.IsPKHandleColumn(t.Meta()) {

handleChanged = true

newHandle = newData[i].GetInt64()

}

} else {

if mysql.HasOnUpdateNowFlag(col.Flag) && modified[i] {

// It's for "UPDATE t SET ts = ts" and ts is a timestamp.

onUpdateSpecified[i] = true

}

modified[i] = false

}

}

// Check the not-null constraints.

err := table.CheckNotNull(t.Cols(), newData)

if err != nil {

return false, errors.Trace(err)

}

if !changed {

// See https://dev.mysql.com/doc/refman/5.7/en/mysql-real-connect.html CLIENT_FOUND_ROWS

if ctx.GetSessionVars().ClientCapability&mysql.ClientFoundRows > 0 {

sc.AddAffectedRows(1)

}

return false, nil

}

// Fill values into on-update-now fields, only if they are really changed.

for i, col := range t.Cols() {

if mysql.HasOnUpdateNowFlag(col.Flag) && !modified[i] && !onUpdateSpecified[i] {

v, errGT := expression.GetTimeValue(ctx, strings.ToUpper(ast.CurrentTimestamp), col.Tp, col.Decimal)

if errGT != nil {

return false, errors.Trace(errGT)

}

newData[i] = v

modified[i] = true

}

}

if handleChanged {

skipHandleCheck := false

if ignoreErr {

// if the new handle exists. `UPDATE IGNORE` will avoid removing record, and do nothing.

if err = tables.CheckHandleExists(ctx, t, newHandle); err != nil {

return false, errors.Trace(err)

}

skipHandleCheck = true

}

err = t.RemoveRecord(ctx, h, oldData)

if err != nil {

return false, errors.Trace(err)

}

_, err = t.AddRecord(ctx, newData, skipHandleCheck)

} else {

// Update record to new value and update index.

err = t.UpdateRecord(ctx, h, oldData, newData, modified)

}

if err != nil {

return false, errors.Trace(err)

}

tid := t.Meta().ID

ctx.StmtAddDirtyTableOP(DirtyTableDeleteRow, tid, h, nil)

ctx.StmtAddDirtyTableOP(DirtyTableAddRow, tid, h, newData)

if onDup {

sc.AddAffectedRows(2)

} else {

sc.AddAffectedRows(1)

}

ctx.GetSessionVars().TxnCtx.UpdateDeltaForTable(t.Meta().ID, 0, 1)

return true, nil

}

// DeleteExec represents a delete executor.

// See https://dev.mysql.com/doc/refman/5.7/en/delete.html

type DeleteExec struct {

baseExecutor

SelectExec Executor

Tables []*ast.TableName

IsMultiTable bool

tblID2Table map[int64]table.Table

// tblMap is the table map value is an array which contains table aliases.

// Table ID may not be unique for deleting multiple tables, for statements like

// `delete from t as t1, t as t2`, the same table has two alias, we have to identify a table

// by its alias instead of ID.

tblMap map[int64][]*ast.TableName

finished bool

}

// Next implements the Executor Next interface.

func (e *DeleteExec) Next(goCtx goctx.Context) (Row, error) {

if e.finished {

return nil, nil

}

defer func() {

e.finished = true

}()

if e.IsMultiTable {

return nil, e.deleteMultiTables(goCtx)

}

return nil, e.deleteSingleTable(goCtx)

}

// NextChunk implements the Executor NextChunk interface.

func (e *DeleteExec) NextChunk(goCtx goctx.Context, chk *chunk.Chunk) error {

chk.Reset()

if e.finished {

return nil

}

defer func() {

e.finished = true

}()

if e.IsMultiTable {

return errors.Trace(e.deleteMultiTablesByChunk(goCtx))

}

return errors.Trace(e.deleteSingleTableByChunk(goCtx))

}

type tblColPosInfo struct {

tblID int64

colBeginIndex int

colEndIndex int

handleIndex int

}

// tableRowMapType is a map for unique (Table, Row) pair. key is the tableID.

// the key in map[int64]Row is the joined table handle, which represent a unique reference row.

// the value in map[int64]Row is the deleting row.

type tableRowMapType map[int64]map[int64]Row

func (e *DeleteExec) deleteMultiTables(goCtx goctx.Context) error {

if len(e.Tables) == 0 {

return nil

}

e.initialMultiTableTblMap()

colPosInfos := e.getColPosInfos(e.children[0].Schema())

tblRowMap := make(tableRowMapType)

for {

joinedRow, err := e.SelectExec.Next(goCtx)

if err != nil {

return errors.Trace(err)

}

if joinedRow == nil {

break

}

e.composeTblRowMap(tblRowMap, colPosInfos, joinedRow)

}

return errors.Trace(e.removeRowsInTblRowMap(tblRowMap))

}

// matchingDeletingTable checks whether this column is from the table which is in the deleting list.

func (e *DeleteExec) matchingDeletingTable(tableID int64, col *expression.Column) bool {

names, ok := e.tblMap[tableID]

if !ok {

return false

}

for _, n := range names {

if (col.DBName.L == "" || col.DBName.L == n.Schema.L) && col.TblName.L == n.Name.L {

return true

}

}

return false

}

func (e *DeleteExec) deleteSingleTable(goCtx goctx.Context) error {

var (

id int64

tbl table.Table

handleCol *expression.Column

rowCount int

)

for i, t := range e.tblID2Table {

id, tbl = i, t

handleCol = e.SelectExec.Schema().TblID2Handle[id][0]

break

}

// If tidb_batch_delete is ON and not in a transaction, we could use BatchDelete mode.

batchDelete := e.ctx.GetSessionVars().BatchDelete && !e.ctx.GetSessionVars().InTxn()

batchSize := e.ctx.GetSessionVars().DMLBatchSize

for {

if batchDelete && rowCount >= batchSize {

e.ctx.StmtCommit()

if err := e.ctx.NewTxn(); err != nil {

// We should return a special error for batch insert.

return ErrBatchInsertFail.Gen("BatchDelete failed with error: %v", err)

}

rowCount = 0

}

row, err := e.SelectExec.Next(goCtx)

if err != nil {

return errors.Trace(err)

}

if row == nil {

break

}

err = e.deleteOneRow(tbl, handleCol, row)

if err != nil {

return errors.Trace(err)

}

rowCount++

}

return nil

}

func (e *DeleteExec) deleteOneRow(tbl table.Table, handleCol *expression.Column, row Row) error {

end := len(row)

if handleIsExtra(handleCol) {

end--

}

handle := row[handleCol.Index].GetInt64()

err := e.removeRow(e.ctx, tbl, handle, row[:end])

if err != nil {

return errors.Trace(err)

}

return nil

}

func (e *DeleteExec) deleteSingleTableByChunk(goCtx goctx.Context) error {

var (

id int64

tbl table.Table

handleCol *expression.Column

rowCount int

)

for i, t := range e.tblID2Table {

id, tbl = i, t

handleCol = e.children[0].Schema().TblID2Handle[id][0]

break

}

// If tidb_batch_delete is ON and not in a transaction, we could use BatchDelete mode.

batchDelete := e.ctx.GetSessionVars().BatchDelete && !e.ctx.GetSessionVars().InTxn()

batchDMLSize := e.ctx.GetSessionVars().DMLBatchSize

fields := e.children[0].retTypes()

for {

chk := e.children[0].newChunk()

iter := chunk.NewIterator4Chunk(chk)

err := e.children[0].NextChunk(goCtx, chk)

if err != nil {

return errors.Trace(err)

}

if chk.NumRows() == 0 {

break

}

for chunkRow := iter.Begin(); chunkRow != iter.End(); chunkRow = iter.Next() {

if batchDelete && rowCount >= batchDMLSize {

e.ctx.StmtCommit()

if err = e.ctx.NewTxn(); err != nil {

// We should return a special error for batch insert.

return ErrBatchInsertFail.Gen("BatchDelete failed with error: %v", err)

}

rowCount = 0

}

datumRow := chunkRow.GetDatumRow(fields)

err = e.deleteOneRow(tbl, handleCol, datumRow)

if err != nil {

return errors.Trace(err)

}

rowCount++

}

}

return nil

}

func (e *DeleteExec) initialMultiTableTblMap() {

e.tblMap = make(map[int64][]*ast.TableName, len(e.Tables))

for _, t := range e.Tables {

e.tblMap[t.TableInfo.ID] = append(e.tblMap[t.TableInfo.ID], t)

}

}

func (e *DeleteExec) getColPosInfos(schema *expression.Schema) []tblColPosInfo {

var colPosInfos []tblColPosInfo

// Extract the columns' position information of this table in the delete's schema, together with the table id

// and its handle's position in the schema.

for id, cols := range schema.TblID2Handle {

tbl := e.tblID2Table[id]

for _, col := range cols {

if !e.matchingDeletingTable(id, col) {

continue

}

offset := getTableOffset(schema, col)

end := offset + len(tbl.Cols())

colPosInfos = append(colPosInfos, tblColPosInfo{tblID: id, colBeginIndex: offset, colEndIndex: end, handleIndex: col.Index})

}

}

return colPosInfos

}

func (e *DeleteExec) composeTblRowMap(tblRowMap tableRowMapType, colPosInfos []tblColPosInfo, joinedRow Row) {

// iterate all the joined tables, and got the copresonding rows in joinedRow.

for _, info := range colPosInfos {

if tblRowMap[info.tblID] == nil {

tblRowMap[info.tblID] = make(map[int64]Row)

}

handle := joinedRow[info.handleIndex].GetInt64()

// tblRowMap[info.tblID][handle] hold the row datas binding to this table and this handle.

tblRowMap[info.tblID][handle] = joinedRow[info.colBeginIndex:info.colEndIndex]

}

}

func (e *DeleteExec) deleteMultiTablesByChunk(goCtx goctx.Context) error {

if len(e.Tables) == 0 {

return nil

}

e.initialMultiTableTblMap()

colPosInfos := e.getColPosInfos(e.children[0].Schema())

tblRowMap := make(tableRowMapType)

fields := e.children[0].retTypes()

for {

chk := e.children[0].newChunk()

iter := chunk.NewIterator4Chunk(chk)

err := e.children[0].NextChunk(goCtx, chk)

if err != nil {

return errors.Trace(err)

}

if chk.NumRows() == 0 {

break

}

for joinedChunkRow := iter.Begin(); joinedChunkRow != iter.End(); joinedChunkRow = iter.Next() {

joinedDatumRow := joinedChunkRow.GetDatumRow(fields)

e.composeTblRowMap(tblRowMap, colPosInfos, joinedDatumRow)

}

}

return errors.Trace(e.removeRowsInTblRowMap(tblRowMap))

}

func (e *DeleteExec) removeRowsInTblRowMap(tblRowMap tableRowMapType) error {

for id, rowMap := range tblRowMap {

for handle, data := range rowMap {

err := e.removeRow(e.ctx, e.tblID2Table[id], handle, data)

if err != nil {

return errors.Trace(err)

}

}

}

return nil

}

const (

// DirtyTableAddRow is the constant for dirty table operation type.

DirtyTableAddRow = iota

// DirtyTableDeleteRow is the constant for dirty table operation type.

DirtyTableDeleteRow

// DirtyTableTruncate is the constant for dirty table operation type.

DirtyTableTruncate

)

func (e *DeleteExec) removeRow(ctx context.Context, t table.Table, h int64, data []types.Datum) error {

err := t.RemoveRecord(ctx, h, data)

if err != nil {

return errors.Trace(err)

}

ctx.StmtAddDirtyTableOP(DirtyTableDeleteRow, t.Meta().ID, h, nil)

ctx.GetSessionVars().StmtCtx.AddAffectedRows(1)

ctx.GetSessionVars().TxnCtx.UpdateDeltaForTable(t.Meta().ID, -1, 1)

return nil

}

// Close implements the Executor Close interface.

func (e *DeleteExec) Close() error {

return e.SelectExec.Close()

}

// Open implements the Executor Open interface.

func (e *DeleteExec) Open(goCtx goctx.Context) error {

return e.SelectExec.Open(goCtx)

}

// NewLoadDataInfo returns a LoadDataInfo structure, and it's only used for tests now.

func NewLoadDataInfo(row []types.Datum, ctx context.Context, tbl table.Table, cols []*table.Column) *LoadDataInfo {

insertVal := &InsertValues{baseExecutor: newBaseExecutor(nil, ctx, "InsertValues"), Table: tbl}

return &LoadDataInfo{

row: row,

insertVal: insertVal,

Table: tbl,

Ctx: ctx,

columns: cols,

}

}

// LoadDataInfo saves the information of loading data operation.

type LoadDataInfo struct {

row []types.Datum

insertVal *InsertValues

Path string

Table table.Table

FieldsInfo *ast.FieldsClause

LinesInfo *ast.LinesClause

Ctx context.Context

columns []*table.Column

}

// SetMaxRowsInBatch sets the max number of rows to insert in a batch.

func (e *LoadDataInfo) SetMaxRowsInBatch(limit uint64) {

e.insertVal.maxRowsInBatch = limit

}

// getValidData returns prevData and curData that starts from starting symbol.

// If the data doesn't have starting symbol, prevData is nil and curData is curData[len(curData)-startingLen+1:].

// If curData size less than startingLen, curData is returned directly.

func (e *LoadDataInfo) getValidData(prevData, curData []byte) ([]byte, []byte) {

startingLen := len(e.LinesInfo.Starting)

if startingLen == 0 {

return prevData, curData

}

prevLen := len(prevData)

if prevLen > 0 {

// starting symbol in the prevData

idx := strings.Index(string(prevData), e.LinesInfo.Starting)

if idx != -1 {

return prevData[idx:], curData

}

// starting symbol in the middle of prevData and curData

restStart := curData

if len(curData) >= startingLen {

restStart = curData[:startingLen-1]

}

prevData = append(prevData, restStart...)

idx = strings.Index(string(prevData), e.LinesInfo.Starting)

if idx != -1 {

return prevData[idx:prevLen], curData

}

}

// starting symbol in the curData

idx := strings.Index(string(curData), e.LinesInfo.Starting)

if idx != -1 {

return nil, curData[idx:]

}

// no starting symbol

if len(curData) >= startingLen {

curData = curData[len(curData)-startingLen+1:]

}

return nil, curData

}

// getLine returns a line, curData, the next data start index and a bool value.

// If it has starting symbol the bool is true, otherwise is false.

func (e *LoadDataInfo) getLine(prevData, curData []byte) ([]byte, []byte, bool) {

startingLen := len(e.LinesInfo.Starting)

prevData, curData = e.getValidData(prevData, curData)

if prevData == nil && len(curData) < startingLen {

return nil, curData, false

}

prevLen := len(prevData)

terminatedLen := len(e.LinesInfo.Terminated)

curStartIdx := 0

if prevLen < startingLen {

curStartIdx = startingLen - prevLen

}

endIdx := -1

if len(curData) >= curStartIdx {

endIdx = strings.Index(string(curData[curStartIdx:]), e.LinesInfo.Terminated)

}

if endIdx == -1 {

// no terminated symbol

if len(prevData) == 0 {

return nil, curData, true

}

// terminated symbol in the middle of prevData and curData

curData = append(prevData, curData...)

endIdx = strings.Index(string(curData[startingLen:]), e.LinesInfo.Terminated)

if endIdx != -1 {

nextDataIdx := startingLen + endIdx + terminatedLen

return curData[startingLen : startingLen+endIdx], curData[nextDataIdx:], true

}

// no terminated symbol

return nil, curData, true

}

// terminated symbol in the curData

nextDataIdx := curStartIdx + endIdx + terminatedLen

if len(prevData) == 0 {

return curData[curStartIdx : curStartIdx+endIdx], curData[nextDataIdx:], true

}

// terminated symbol in the curData

prevData = append(prevData, curData[:nextDataIdx]...)

endIdx = strings.Index(string(prevData[startingLen:]), e.LinesInfo.Terminated)

if endIdx >= prevLen {

return prevData[startingLen : startingLen+endIdx], curData[nextDataIdx:], true

}

// terminated symbol in the middle of prevData and curData

lineLen := startingLen + endIdx + terminatedLen

return prevData[startingLen : startingLen+endIdx], curData[lineLen-prevLen:], true

}

// InsertData inserts data into specified table according to the specified format.

// If it has the rest of data isn't completed the processing, then is returns without completed data.

// If the number of inserted rows reaches the batchRows, then the second return value is true.

// If prevData isn't nil and curData is nil, there are no other data to deal with and the isEOF is true.

func (e *LoadDataInfo) InsertData(prevData, curData []byte) ([]byte, bool, error) {

// TODO: support enclosed and escape.

if len(prevData) == 0 && len(curData) == 0 {

return nil, false, nil

}

var line []byte

var isEOF, hasStarting, reachLimit bool

if len(prevData) > 0 && len(curData) == 0 {

isEOF = true

prevData, curData = curData, prevData

}

rows := make([][]types.Datum, 0, e.insertVal.maxRowsInBatch)

for len(curData) > 0 {

line, curData, hasStarting = e.getLine(prevData, curData)

prevData = nil

// If it doesn't find the terminated symbol and this data isn't the last data,

// the data can't be inserted.

if line == nil && !isEOF {

break

}

// If doesn't find starting symbol, this data can't be inserted.

if !hasStarting {

if isEOF {

curData = nil

}

break

}

if line == nil && isEOF {

line = curData[len(e.LinesInfo.Starting):]

curData = nil

}

cols, err := GetFieldsFromLine(line, e.FieldsInfo)

if err != nil {

return nil, false, errors.Trace(err)

}

rows = append(rows, e.colsToRow(cols))

e.insertVal.rowCount++

if e.insertVal.maxRowsInBatch != 0 && e.insertVal.rowCount%e.insertVal.maxRowsInBatch == 0 {

reachLimit = true

log.Infof("This insert rows has reached the batch %d, current total rows %d",

e.insertVal.maxRowsInBatch, e.insertVal.rowCount)

break

}

}

rows, err := batchMarkDupRows(e.Ctx, e.Table, rows)

if err != nil {

return nil, reachLimit, errors.Trace(err)

}

for _, row := range rows {

e.insertData(row)

}

if e.insertVal.lastInsertID != 0 {

e.insertVal.ctx.GetSessionVars().SetLastInsertID(e.insertVal.lastInsertID)

}

return curData, reachLimit, nil

}

// GetFieldsFromLine splits line according to fieldsInfo, this function is exported for testing.

func GetFieldsFromLine(line []byte, fieldsInfo *ast.FieldsClause) ([]string, error) {

var sep []byte

if fieldsInfo.Enclosed != 0 {

if line[0] != fieldsInfo.Enclosed || line[len(line)-1] != fieldsInfo.Enclosed {

return nil, errors.Errorf("line %s should begin and end with %c", string(line), fieldsInfo.Enclosed)

}

line = line[1 : len(line)-1]

sep = make([]byte, 0, len(fieldsInfo.Terminated)+2)

sep = append(sep, fieldsInfo.Enclosed)

sep = append(sep, fieldsInfo.Terminated...)

sep = append(sep, fieldsInfo.Enclosed)

} else {

sep = []byte(fieldsInfo.Terminated)

}

rawCols := bytes.Split(line, sep)

cols := escapeCols(rawCols)

return cols, nil

}

func escapeCols(strs [][]byte) []string {

ret := make([]string, len(strs))

for i, v := range strs {

output := escape(v)

ret[i] = string(output)

}

return ret

}

// escape handles escape characters when running load data statement.

// TODO: escape need to be improved, it should support ESCAPED BY to specify

// the escape character and handle \N escape.

// See http://dev.mysql.com/doc/refman/5.7/en/load-data.html

func escape(str []byte) []byte {

pos := 0

for i := 0; i < len(str); i++ {

c := str[i]

if c == '\\' && i+1 < len(str) {

c = escapeChar(str[i+1])

i++

}

str[pos] = c

pos++

}

return str[:pos]

}

func escapeChar(c byte) byte {

switch c {

case '0':

return 0

case 'b':

return '\b'

case 'n':

return '\n'

case 'r':

return '\r'

case 't':

return '\t'

case 'Z':

return 26

case '\\':

return '\\'

}

return c

}

func (e *LoadDataInfo) colsToRow(cols []string) types.DatumRow {

for i := 0; i < len(e.row); i++ {

if i >= len(cols) {

e.row[i].SetString("")

continue

}

e.row[i].SetString(cols[i])

}

row, err := e.insertVal.fillRowData(e.columns, e.row, true)

if err != nil {

warnLog := fmt.Sprintf("Load Data: insert data:%v failed:%v", e.row, errors.ErrorStack(err))

e.insertVal.handleLoadDataWarnings(err, warnLog)

return nil

}

return row

}

func (e *LoadDataInfo) insertData(row types.DatumRow) {

if row == nil {

return

}

_, err := e.Table.AddRecord(e.insertVal.ctx, row, false)

if err != nil {

warnLog := fmt.Sprintf("Load Data: insert data:%v failed:%v", row, errors.ErrorStack(err))

e.insertVal.handleLoadDataWarnings(err, warnLog)

}

}

func (e *InsertValues) handleLoadDataWarnings(err error, logInfo string) {

sc := e.ctx.GetSessionVars().StmtCtx

sc.AppendWarning(err)

log.Warn(logInfo)

}

// LoadData represents a load data executor.

type LoadData struct {

baseExecutor

IsLocal bool

loadDataInfo *LoadDataInfo

}

// loadDataVarKeyType is a dummy type to avoid naming collision in context.

type loadDataVarKeyType int

// String defines a Stringer function for debugging and pretty printing.

func (k loadDataVarKeyType) String() string {

return "load_data_var"

}

// LoadDataVarKey is a variable key for load data.

const LoadDataVarKey loadDataVarKeyType = 0

func (e *LoadData) exec(goCtx goctx.Context) (Row, error) {

// TODO: support load data without local field.

if !e.IsLocal {

return nil, errors.New("Load Data: don't support load data without local field")

}

// TODO: support lines terminated is "".

if len(e.loadDataInfo.LinesInfo.Terminated) == 0 {

return nil, errors.New("Load Data: don't support load data terminated is nil")

}

ctx := e.loadDataInfo.insertVal.ctx

val := ctx.Value(LoadDataVarKey)

if val != nil {

ctx.SetValue(LoadDataVarKey, nil)

return nil, errors.New("Load Data: previous load data option isn't closed normal")

}

if e.loadDataInfo.Path == "" {

return nil, errors.New("Load Data: infile path is empty")

}

ctx.SetValue(LoadDataVarKey, e.loadDataInfo)

return nil, nil

}

// Next implements the Executor Next interface.

func (e *LoadData) Next(goCtx goctx.Context) (Row, error) {

return e.exec(goCtx)

}

// NextChunk implements the Executor NextChunk interface.

func (e *LoadData) NextChunk(goCtx goctx.Context, chk *chunk.Chunk) error {

chk.Reset()

_, err := e.exec(goCtx)

return errors.Trace(err)

}

// Close implements the Executor Close interface.

func (e *LoadData) Close() error {

return nil

}

// Open implements the Executor Open interface.

func (e *LoadData) Open(goCtx goctx.Context) error {

return nil

}

type defaultVal struct {

val types.Datum

// We evaluate the default value lazily. The valid indicates whether the val is evaluated.

valid bool

}

// InsertValues is the data to insert.

type InsertValues struct {

baseExecutor

rowCount uint64

maxRowsInBatch uint64

lastInsertID uint64

needFillDefaultValues bool

SelectExec Executor

Table table.Table

Columns []*ast.ColumnName

Lists [][]expression.Expression

Setlist []*expression.Assignment

IsPrepare bool

GenColumns []*ast.ColumnName

GenExprs []expression.Expression

// colDefaultVals is used to store casted default value.

// Because not every insert statement needs colDefaultVals, so we will init the buffer lazily.

colDefaultVals []defaultVal

}

// InsertExec represents an insert executor.

type InsertExec struct {

*InsertValues

OnDuplicate []*expression.Assignment

Priority mysql.PriorityEnum

IgnoreErr bool

finished bool

}

func (e *InsertExec) exec(goCtx goctx.Context, rows [][]types.Datum) (Row, error) {

// If tidb_batch_insert is ON and not in a transaction, we could use BatchInsert mode.

sessVars := e.ctx.GetSessionVars()

defer sessVars.CleanBuffers()

batchInsert := sessVars.BatchInsert && !sessVars.InTxn()

batchSize := sessVars.DMLBatchSize

txn := e.ctx.Txn()

rowCount := 0

writeBufs := sessVars.GetWriteStmtBufs()

if !sessVars.ImportingData {

writeBufs.BufStore = kv.NewBufferStore(txn, kv.TempTxnMemBufCap)

}

// If you use the IGNORE keyword, duplicate-key error that occurs while executing the INSERT statement are ignored.

// For example, without IGNORE, a row that duplicates an existing UNIQUE index or PRIMARY KEY value in

// the table causes a duplicate-key error and the statement is aborted. With IGNORE, the row is discarded and no error occurs.

// Using BatchGet in insert ignore to mark rows as duplicated before we add records to the table.

if e.IgnoreErr {

var err error

rows, err = batchMarkDupRows(e.ctx, e.Table, rows)

if err != nil {

return nil, errors.Trace(err)

}

}

for _, row := range rows {

// duplicate row will be marked as nil in batchMarkDupRows if

// IgnoreErr is true. For IgnoreErr is false, it is a protection.

if row == nil {

continue

}

if batchInsert && rowCount >= batchSize {

e.ctx.StmtCommit()

if err := e.ctx.NewTxn(); err != nil {

// We should return a special error for batch insert.

return nil, ErrBatchInsertFail.Gen("BatchInsert failed with error: %v", err)

}

txn = e.ctx.Txn()

rowCount = 0

if !sessVars.ImportingData {

writeBufs.BufStore = kv.NewBufferStore(txn, kv.TempTxnMemBufCap)

}

}

if len(e.OnDuplicate) == 0 && !e.IgnoreErr {

txn.SetOption(kv.PresumeKeyNotExists, nil)

}

h, err := e.Table.AddRecord(e.ctx, row, false)

txn.DelOption(kv.PresumeKeyNotExists)

if err == nil {

if !sessVars.ImportingData {

e.ctx.StmtAddDirtyTableOP(DirtyTableAddRow, e.Table.Meta().ID, h, row)

}

rowCount++

continue

}

if kv.ErrKeyExists.Equal(err) {

if len(e.OnDuplicate) > 0 {

if err = e.onDuplicateUpdate(row, h, e.OnDuplicate); err != nil {

return nil, errors.Trace(err)

}

rowCount++

continue

}

}

return nil, errors.Trace(err)

}

if e.lastInsertID != 0 {

sessVars.SetLastInsertID(e.lastInsertID)

}

e.finished = true

return nil, nil

}

type keyWithDupError struct {

key kv.Key

dupErr error

}

func getRecordIDs(ctx context.Context, t table.Table, rows [][]types.Datum) ([]int64, error) {

recordIDs := make([]int64, 0, len(rows))

if t.Meta().PKIsHandle {

var handleCol *table.Column

for _, col := range t.Cols() {

if col.IsPKHandleColumn(t.Meta()) {

handleCol = col

break

}

}

for _, row := range rows {

recordIDs = append(recordIDs, row[handleCol.Offset].GetInt64())

}

} else {

for range rows {

recordID, err := t.AllocAutoID(ctx)

if err != nil {

return nil, errors.Trace(err)

}

recordIDs = append(recordIDs, recordID)

}

}

return recordIDs, nil

}

// getKeysNeedCheck gets keys converted from to-be-insert rows to record keys and unique index keys,

// which need to be checked whether they are duplicate keys.

func getKeysNeedCheck(ctx context.Context, t table.Table, rows [][]types.Datum) ([][]keyWithDupError, error) {

nUnique := 0

for _, v := range t.WritableIndices() {

if v.Meta().Unique {

nUnique++

}

}

rowWithKeys := make([][]keyWithDupError, 0, len(rows))

// get recordIDs

recordIDs, err := getRecordIDs(ctx, t, rows)

if err != nil {

return nil, errors.Trace(err)

}