You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

245 lines
6.8 KiB

// Copyright 2018 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 (
"context"
"fmt"
"runtime/trace"
"time"
"github.com/pingcap/errors"
"github.com/pingcap/parser/mysql"
"github.com/pingcap/tidb/kv"
"github.com/pingcap/tidb/tablecodec"
"github.com/pingcap/tidb/types"
"github.com/pingcap/tidb/util/chunk"
"github.com/pingcap/tidb/util/logutil"
"github.com/pingcap/tidb/util/memory"
"go.uber.org/zap"
)
// ReplaceExec represents a replace executor.
type ReplaceExec struct {
*InsertValues
Priority int
}
// Close implements the Executor Close interface.
func (e *ReplaceExec) Close() error {
e.setMessage()
if e.SelectExec != nil {
return e.SelectExec.Close()
}
return nil
}
// Open implements the Executor Open interface.
func (e *ReplaceExec) Open(ctx context.Context) error {
e.memTracker = memory.NewTracker(e.id, -1)
e.memTracker.AttachTo(e.ctx.GetSessionVars().StmtCtx.MemTracker)
if e.SelectExec != nil {
return e.SelectExec.Open(ctx)
}
e.initEvalBuffer()
return nil
}
// removeRow removes the duplicate row and cleanup its keys in the key-value map,
// but if the to-be-removed row equals to the to-be-added row, no remove or add things to do.
func (e *ReplaceExec) removeRow(ctx context.Context, txn kv.Transaction, handle int64, r toBeCheckedRow) (bool, error) {
newRow := r.row
oldRow, err := getOldRow(ctx, e.ctx, txn, r.t, handle, e.GenExprs)
if err != nil {
logutil.BgLogger().Error("get old row failed when replace",
zap.Int64("handle", handle),
zap.String("toBeInsertedRow", types.DatumsToStrNoErr(r.row)))
if kv.IsErrNotFound(err) {
err = errors.NotFoundf("can not be duplicated row, due to old row not found. handle %d", handle)
}
return false, err
}
rowUnchanged, err := types.EqualDatums(e.ctx.GetSessionVars().StmtCtx, oldRow, newRow)
if err != nil {
return false, err
}
if rowUnchanged {
e.ctx.GetSessionVars().StmtCtx.AddAffectedRows(1)
return true, nil
}
err = r.t.RemoveRecord(e.ctx, handle, oldRow)
if err != nil {
return false, err
}
e.ctx.GetSessionVars().StmtCtx.AddAffectedRows(1)
return false, nil
}
// replaceRow removes all duplicate rows for one row, then inserts it.
func (e *ReplaceExec) replaceRow(ctx context.Context, r toBeCheckedRow) error {
txn, err := e.ctx.Txn(true)
if err != nil {
return err
}
if r.handleKey != nil {
handle, err := tablecodec.DecodeRowKey(r.handleKey.newKey)
if err != nil {
return err
}
if _, err := txn.Get(ctx, r.handleKey.newKey); err == nil {
rowUnchanged, err := e.removeRow(ctx, txn, handle, r)
if err != nil {
return err
}
if rowUnchanged {
return nil
}
} else {
if !kv.IsErrNotFound(err) {
return err
}
}
}
// Keep on removing duplicated rows.
for {
rowUnchanged, foundDupKey, err := e.removeIndexRow(ctx, txn, r)
if err != nil {
return err
}
if rowUnchanged {
return nil
}
if foundDupKey {
continue
}
break
}
// No duplicated rows now, insert the row.
_, err = e.addRecord(ctx, r.row)
if err != nil {
return err
}
return nil
}
// removeIndexRow removes the row which has a duplicated key.
// the return values:
// 1. bool: true when the row is unchanged. This means no need to remove, and then add the row.
// 2. bool: true when found the duplicated key. This only means that duplicated key was found,
// and the row was removed.
// 3. error: the error.
func (e *ReplaceExec) removeIndexRow(ctx context.Context, txn kv.Transaction, r toBeCheckedRow) (bool, bool, error) {
for _, uk := range r.uniqueKeys {
val, err := txn.Get(ctx, uk.newKey)
if err != nil {
if kv.IsErrNotFound(err) {
continue
}
return false, false, err
}
handle, err := tablecodec.DecodeHandle(val)
if err != nil {
return false, true, err
}
rowUnchanged, err := e.removeRow(ctx, txn, handle, r)
if err != nil {
return false, true, err
}
return rowUnchanged, true, nil
}
return false, false, nil
}
func (e *ReplaceExec) exec(ctx context.Context, newRows [][]types.Datum) error {
/*
* MySQL uses the following algorithm for REPLACE (and LOAD DATA ... REPLACE):
* 1. Try to insert the new row into the table
* 2. While the insertion fails because a duplicate-key error occurs for a primary key or unique index:
* 3. Delete from the table the conflicting row that has the duplicate key value
* 4. Try again to insert the new row into the table
* See http://dev.mysql.com/doc/refman/5.7/en/replace.html
*
* For REPLACE statements, the affected-rows value is 2 if the new row replaced an old row,
* because in this case, one row was inserted after the duplicate was deleted.
* See http://dev.mysql.com/doc/refman/5.7/en/mysql-affected-rows.html
*/
defer trace.StartRegion(ctx, "ReplaceExec").End()
// Get keys need to be checked.
toBeCheckedRows, err := getKeysNeedCheck(ctx, e.ctx, e.Table, newRows)
if err != nil {
return err
}
txn, err := e.ctx.Txn(true)
if err != nil {
return err
}
txnSize := txn.Size()
if e.collectRuntimeStatsEnabled() {
if snapshot := txn.GetSnapshot(); snapshot != nil {
snapshot.SetOption(kv.CollectRuntimeStats, e.stats.SnapshotRuntimeStats)
defer snapshot.DelOption(kv.CollectRuntimeStats)
}
}
prefetchStart := time.Now()
// Use BatchGet to fill cache.
// It's an optimization and could be removed without affecting correctness.
if err = prefetchDataCache(ctx, txn, toBeCheckedRows); err != nil {
return err
}
if e.stats != nil {
e.stats.Prefetch = time.Since(prefetchStart)
}
e.ctx.GetSessionVars().StmtCtx.AddRecordRows(uint64(len(newRows)))
for _, r := range toBeCheckedRows {
err = e.replaceRow(ctx, r)
if err != nil {
return err
}
}
e.memTracker.Consume(int64(txn.Size() - txnSize))
return nil
}
// Next implements the Executor Next interface.
func (e *ReplaceExec) Next(ctx context.Context, req *chunk.Chunk) error {
req.Reset()
if len(e.children) > 0 && e.children[0] != nil {
return insertRowsFromSelect(ctx, e)
}
return insertRows(ctx, e)
}
// setMessage sets info message(ERR_INSERT_INFO) generated by REPLACE statement
func (e *ReplaceExec) setMessage() {
stmtCtx := e.ctx.GetSessionVars().StmtCtx
numRecords := stmtCtx.RecordRows()
if e.SelectExec != nil || numRecords > 1 {
numWarnings := stmtCtx.WarningCount()
numDuplicates := stmtCtx.AffectedRows() - numRecords
msg := fmt.Sprintf(mysql.MySQLErrName[mysql.ErrInsertInfo].Raw, numRecords, numDuplicates, numWarnings)
stmtCtx.SetMessage(msg)
}
}