Method batchCheckAndInsert

pkg/executor/insert_common.go:1200–1334

batchCheckAndInsert checks rows with duplicate errors. All duplicate rows will be ignored and appended as duplicate warnings.

(
	ctx context.Context, rows [][]types.Datum,
	addRecord func(ctx context.Context, row []types.Datum, dupKeyCheck table.DupKeyCheckMode) error,
	replace bool,
)

// batchCheckAndInsert checks rows with duplicate errors.
// All duplicate rows will be ignored and appended as duplicate warnings.
func (e *InsertValues) batchCheckAndInsert(
	ctx context.Context, rows [][]types.Datum,
	addRecord func(ctx context.Context, row []types.Datum, dupKeyCheck table.DupKeyCheckMode) error,
	replace bool,
) error {
	defer tracing.StartRegion(ctx, "InsertValues.batchCheckAndInsert").End()
	start := time.Now()
	// Get keys need to be checked.
	toBeCheckedRows, err := getKeysNeedCheck(e.Ctx(), e.Table, rows)
	if err != nil {
		return err
	}

	txn, err := e.Ctx().Txn(true)
	if err != nil {
		return err
	}
	setOptionForTopSQL(e.Ctx().GetSessionVars().StmtCtx, txn)
	sc := e.Ctx().GetSessionVars().StmtCtx
	for _, fkc := range e.fkChecks {
		err = fkc.checkRows(ctx, sc, txn, toBeCheckedRows)
		if err != nil {
			return err
		}
	}
	prefetchStart := time.Now()
	// Fill cache using BatchGet, the following Get requests don't need to visit TiKV.
	// Temporary table need not to do prefetch because its all data are stored in the memory.
	if e.Table.Meta().TempTableType == model.TempTableNone {
		if _, err = prefetchUniqueIndices(ctx, txn, toBeCheckedRows); err != nil {
			return err
		}
	}

	if e.stats != nil {
		e.stats.FKCheckTime += prefetchStart.Sub(start)
		e.stats.Prefetch += time.Since(prefetchStart)
	}

	// append warnings and get no duplicated error rows
	for i, r := range toBeCheckedRows {
		if r.ignored {
			continue
		}
		if r.handleKey != nil {
			_, err := txn.Get(ctx, r.handleKey.newKey)
			if err == nil {
				if !replace {
					e.Ctx().GetSessionVars().StmtCtx.AppendWarning(r.handleKey.dupErr)
					if txnCtx := e.Ctx().GetSessionVars().TxnCtx; txnCtx.IsPessimistic &&
						e.Ctx().GetSessionVars().LockUnchangedKeys {
						// lock duplicated row key on insert-ignore
						txnCtx.AddUnchangedKeyForLock(r.handleKey.newKey, false)
					}
					continue
				}
				handle, err := tablecodec.DecodeRowKey(r.handleKey.newKey)
				if err != nil {