Method batchUpdateDupRows

pkg/executor/insert.go:226–322

batchUpdateDupRows updates multi-rows in batch if they are duplicate with rows in table.

(ctx context.Context, newRows [][]types.Datum)

// batchUpdateDupRows updates multi-rows in batch if they are duplicate with rows in table.
func (e *InsertExec) batchUpdateDupRows(ctx context.Context, newRows [][]types.Datum) error {
	// Get keys need to be checked.
	start := time.Now()
	toBeCheckedRows, err := getKeysNeedCheck(e.Ctx(), e.Table, newRows)
	if err != nil {
		return err
	}

	txn, err := e.Ctx().Txn(true)
	if err != nil {
		return err
	}

	prefetchStart := time.Now()
	// Use BatchGet to fill cache.
	// It's an optimization and could be removed without affecting correctness.
	if err = e.prefetchDataCache(ctx, txn, toBeCheckedRows); err != nil {
		return err
	}
	if e.stats != nil {
		e.stats.Prefetch += time.Since(prefetchStart)
	}

	// Use `optimizeDupKeyCheckForUpdate` to determine the update operation when the row meets the conflict in
	// `INSERT ... ON DUPLICATE KEY UPDATE` statement.
	// Though it is in an insert statement, `ON DUP KEY UPDATE` follows the dup-key check behavior of update.
	// For example, it will ignore variable `tidb_constraint_check_in_place`, see the test case:
	// https://github.com/pingcap/tidb/blob/3117d3fae50bbb5dabcde7b9589f92bfbbda5dc6/pkg/executor/test/writetest/write_test.go#L419-L426
	updateDupKeyCheck := optimizeDupKeyCheckForUpdate(txn, e.ignoreErr)
	// Do not use `updateDupKeyCheck` for `AddRecord` because it is not optimized for insert.
	// It seems that we can just use `DupKeyCheckSkip` here because all constraints are checked.
	// But we still use `optimizeDupKeyCheckForNormalInsert` to make the refactor same behavior with the original code.
	// TODO: just use `DupKeyCheckSkip` here.
	addRecordDupKeyCheck := optimizeDupKeyCheckForNormalInsert(e.Ctx().GetSessionVars(), txn)

	_, autoColIdx, found := findAutoIncrementColumn(e.Table)
	if !found {
		autoColIdx = -1
	}

	for i, r := range toBeCheckedRows {
		if r.handleKey != nil {
			handle, err := tablecodec.DecodeRowKey(r.handleKey.newKey)
			if err != nil {
				return err
			}

			err = e.updateDupRow(ctx, i, txn, r, handle, e.OnDuplicate, updateDupKeyCheck, autoColIdx)
			if err == nil {
				continue
			}
			if !kv.IsErrNotFound(err) {
				return err
			}
		}

		for _, uk := range r.uniqueKeys {
			_, handle, err := tables.FetchDuplicatedHandle(ctx, uk.newKey, true, txn)