// 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 }