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.
330 lines
9.0 KiB
330 lines
9.0 KiB
// 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 (
|
|
"context"
|
|
"runtime/trace"
|
|
"sync"
|
|
|
|
"github.com/pingcap/parser/model"
|
|
"github.com/pingcap/parser/mysql"
|
|
"github.com/pingcap/tidb/expression"
|
|
"github.com/pingcap/tidb/sessionctx"
|
|
"github.com/pingcap/tidb/table"
|
|
"github.com/pingcap/tidb/types"
|
|
"github.com/pingcap/tidb/util/chunk"
|
|
)
|
|
|
|
// DirtyDB stores uncommitted write operations for a transaction.
|
|
// It is stored and retrieved by context.Value and context.SetValue method.
|
|
type DirtyDB struct {
|
|
sync.Mutex
|
|
|
|
// tables is a map whose key is tableID.
|
|
tables map[int64]*DirtyTable
|
|
}
|
|
|
|
// GetDirtyTable gets the DirtyTable by id from the DirtyDB.
|
|
func (udb *DirtyDB) GetDirtyTable(tid int64) *DirtyTable {
|
|
// The index join access the tables map parallelly.
|
|
// But the map throws panic in this case. So it's locked.
|
|
udb.Lock()
|
|
dt, ok := udb.tables[tid]
|
|
if !ok {
|
|
dt = &DirtyTable{
|
|
tid: tid,
|
|
addedRows: make(map[int64]struct{}),
|
|
deletedRows: make(map[int64]struct{}),
|
|
}
|
|
udb.tables[tid] = dt
|
|
}
|
|
udb.Unlock()
|
|
return dt
|
|
}
|
|
|
|
// DirtyTable stores uncommitted write operation for a transaction.
|
|
type DirtyTable struct {
|
|
tid int64
|
|
// addedRows ...
|
|
// the key is handle.
|
|
addedRows map[int64]struct{}
|
|
deletedRows map[int64]struct{}
|
|
}
|
|
|
|
// AddRow adds a row to the DirtyDB.
|
|
func (dt *DirtyTable) AddRow(handle int64) {
|
|
dt.addedRows[handle] = struct{}{}
|
|
}
|
|
|
|
// DeleteRow deletes a row from the DirtyDB.
|
|
func (dt *DirtyTable) DeleteRow(handle int64) {
|
|
delete(dt.addedRows, handle)
|
|
dt.deletedRows[handle] = struct{}{}
|
|
}
|
|
|
|
// IsEmpty checks whether the table is empty.
|
|
func (dt *DirtyTable) IsEmpty() bool {
|
|
return len(dt.addedRows)+len(dt.deletedRows) == 0
|
|
}
|
|
|
|
// GetDirtyDB returns the DirtyDB bind to the context.
|
|
func GetDirtyDB(ctx sessionctx.Context) *DirtyDB {
|
|
var udb *DirtyDB
|
|
x := ctx.GetSessionVars().TxnCtx.DirtyDB
|
|
if x == nil {
|
|
udb = &DirtyDB{tables: make(map[int64]*DirtyTable)}
|
|
ctx.GetSessionVars().TxnCtx.DirtyDB = udb
|
|
} else {
|
|
udb = x.(*DirtyDB)
|
|
}
|
|
return udb
|
|
}
|
|
|
|
// UnionScanExec merges the rows from dirty table and the rows from distsql request.
|
|
type UnionScanExec struct {
|
|
baseExecutor
|
|
|
|
dirty *DirtyTable
|
|
// usedIndex is the column offsets of the index which Src executor has used.
|
|
usedIndex []int
|
|
desc bool
|
|
conditions []expression.Expression
|
|
conditionsWithVirCol []expression.Expression
|
|
columns []*model.ColumnInfo
|
|
table table.Table
|
|
// belowHandleIndex is the handle's position of the below scan plan.
|
|
belowHandleIndex int
|
|
|
|
addedRows [][]types.Datum
|
|
cursor4AddRows int
|
|
sortErr error
|
|
snapshotRows [][]types.Datum
|
|
cursor4SnapshotRows int
|
|
snapshotChunkBuffer *chunk.Chunk
|
|
mutableRow chunk.MutRow
|
|
// virtualColumnIndex records all the indices of virtual columns and sort them in definition
|
|
// to make sure we can compute the virtual column in right order.
|
|
virtualColumnIndex []int
|
|
}
|
|
|
|
// Open implements the Executor Open interface.
|
|
func (us *UnionScanExec) Open(ctx context.Context) error {
|
|
if err := us.baseExecutor.Open(ctx); err != nil {
|
|
return err
|
|
}
|
|
return us.open(ctx)
|
|
}
|
|
|
|
func (us *UnionScanExec) open(ctx context.Context) error {
|
|
var err error
|
|
reader := us.children[0]
|
|
|
|
// If the push-downed condition contains virtual column, we may build a selection upon reader. Since unionScanExec
|
|
// has already contained condition, we can ignore the selection.
|
|
if sel, ok := reader.(*SelectionExec); ok {
|
|
reader = sel.children[0]
|
|
}
|
|
|
|
defer trace.StartRegion(ctx, "UnionScanBuildRows").End()
|
|
// 1. select without virtual columns
|
|
// 2. build virtual columns and select with virtual columns
|
|
switch x := reader.(type) {
|
|
case *TableReaderExecutor:
|
|
us.addedRows, err = buildMemTableReader(us, x).getMemRows()
|
|
case *IndexReaderExecutor:
|
|
us.addedRows, err = buildMemIndexReader(us, x).getMemRows()
|
|
case *IndexLookUpExecutor:
|
|
us.addedRows, err = buildMemIndexLookUpReader(us, x).getMemRows()
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
us.snapshotChunkBuffer = newFirstChunk(us)
|
|
return nil
|
|
}
|
|
|
|
// Next implements the Executor Next interface.
|
|
func (us *UnionScanExec) Next(ctx context.Context, req *chunk.Chunk) error {
|
|
req.GrowAndReset(us.maxChunkSize)
|
|
mutableRow := chunk.MutRowFromTypes(retTypes(us))
|
|
for i, batchSize := 0, req.Capacity(); i < batchSize; i++ {
|
|
row, err := us.getOneRow(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// no more data.
|
|
if row == nil {
|
|
return nil
|
|
}
|
|
mutableRow.SetDatums(row...)
|
|
|
|
for _, idx := range us.virtualColumnIndex {
|
|
datum, err := us.schema.Columns[idx].EvalVirtualColumn(mutableRow.ToRow())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Because the expression might return different type from
|
|
// the generated column, we should wrap a CAST on the result.
|
|
castDatum, err := table.CastValue(us.ctx, datum, us.columns[idx], false, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Handle the bad null error.
|
|
if (mysql.HasNotNullFlag(us.columns[idx].Flag) || mysql.HasPreventNullInsertFlag(us.columns[idx].Flag)) && castDatum.IsNull() {
|
|
castDatum = table.GetZeroValue(us.columns[idx])
|
|
}
|
|
mutableRow.SetDatum(idx, castDatum)
|
|
}
|
|
|
|
matched, _, err := expression.EvalBool(us.ctx, us.conditionsWithVirCol, mutableRow.ToRow())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if matched {
|
|
req.AppendRow(mutableRow.ToRow())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Close implements the Executor Close interface.
|
|
func (us *UnionScanExec) Close() error {
|
|
us.cursor4AddRows = 0
|
|
us.cursor4SnapshotRows = 0
|
|
us.addedRows = us.addedRows[:0]
|
|
us.snapshotRows = us.snapshotRows[:0]
|
|
return us.children[0].Close()
|
|
}
|
|
|
|
// getOneRow gets one result row from dirty table or child.
|
|
func (us *UnionScanExec) getOneRow(ctx context.Context) ([]types.Datum, error) {
|
|
snapshotRow, err := us.getSnapshotRow(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
addedRow := us.getAddedRow()
|
|
var row []types.Datum
|
|
var isSnapshotRow bool
|
|
if addedRow == nil {
|
|
row = snapshotRow
|
|
isSnapshotRow = true
|
|
} else if snapshotRow == nil {
|
|
row = addedRow
|
|
} else {
|
|
isSnapshotRow, err = us.shouldPickFirstRow(snapshotRow, addedRow)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if isSnapshotRow {
|
|
row = snapshotRow
|
|
} else {
|
|
row = addedRow
|
|
}
|
|
}
|
|
if row == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
if isSnapshotRow {
|
|
us.cursor4SnapshotRows++
|
|
} else {
|
|
us.cursor4AddRows++
|
|
}
|
|
return row, nil
|
|
}
|
|
|
|
func (us *UnionScanExec) getSnapshotRow(ctx context.Context) ([]types.Datum, error) {
|
|
if us.cursor4SnapshotRows < len(us.snapshotRows) {
|
|
return us.snapshotRows[us.cursor4SnapshotRows], nil
|
|
}
|
|
var err error
|
|
us.cursor4SnapshotRows = 0
|
|
us.snapshotRows = us.snapshotRows[:0]
|
|
for len(us.snapshotRows) == 0 {
|
|
err = Next(ctx, us.children[0], us.snapshotChunkBuffer)
|
|
if err != nil || us.snapshotChunkBuffer.NumRows() == 0 {
|
|
return nil, err
|
|
}
|
|
iter := chunk.NewIterator4Chunk(us.snapshotChunkBuffer)
|
|
for row := iter.Begin(); row != iter.End(); row = iter.Next() {
|
|
snapshotHandle := row.GetInt64(us.belowHandleIndex)
|
|
if _, ok := us.dirty.deletedRows[snapshotHandle]; ok {
|
|
continue
|
|
}
|
|
if _, ok := us.dirty.addedRows[snapshotHandle]; ok {
|
|
// If src handle appears in added rows, it means there is conflict and the transaction will fail to
|
|
// commit, but for simplicity, we don't handle it here.
|
|
continue
|
|
}
|
|
us.snapshotRows = append(us.snapshotRows, row.GetDatumRow(retTypes(us.children[0])))
|
|
}
|
|
}
|
|
return us.snapshotRows[0], nil
|
|
}
|
|
|
|
func (us *UnionScanExec) getAddedRow() []types.Datum {
|
|
var addedRow []types.Datum
|
|
if us.cursor4AddRows < len(us.addedRows) {
|
|
addedRow = us.addedRows[us.cursor4AddRows]
|
|
}
|
|
return addedRow
|
|
}
|
|
|
|
// shouldPickFirstRow picks the suitable row in order.
|
|
// The value returned is used to determine whether to pick the first input row.
|
|
func (us *UnionScanExec) shouldPickFirstRow(a, b []types.Datum) (bool, error) {
|
|
var isFirstRow bool
|
|
addedCmpSrc, err := us.compare(a, b)
|
|
if err != nil {
|
|
return isFirstRow, err
|
|
}
|
|
// Compare result will never be 0.
|
|
if us.desc {
|
|
if addedCmpSrc > 0 {
|
|
isFirstRow = true
|
|
}
|
|
} else {
|
|
if addedCmpSrc < 0 {
|
|
isFirstRow = true
|
|
}
|
|
}
|
|
return isFirstRow, nil
|
|
}
|
|
|
|
func (us *UnionScanExec) compare(a, b []types.Datum) (int, error) {
|
|
sc := us.ctx.GetSessionVars().StmtCtx
|
|
for _, colOff := range us.usedIndex {
|
|
aColumn := a[colOff]
|
|
bColumn := b[colOff]
|
|
cmp, err := aColumn.CompareDatum(sc, &bColumn)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if cmp != 0 {
|
|
return cmp, nil
|
|
}
|
|
}
|
|
aHandle := a[us.belowHandleIndex].GetInt64()
|
|
bHandle := b[us.belowHandleIndex].GetInt64()
|
|
var cmp int
|
|
if aHandle == bHandle {
|
|
cmp = 0
|
|
} else if aHandle > bHandle {
|
|
cmp = 1
|
|
} else {
|
|
cmp = -1
|
|
}
|
|
return cmp, nil
|
|
}
|
|
|