mirror/dolt
1
0
Fork 0
mirror of https://github.com/dolthub/dolt.git synced 2026-02-11 18:49:14 -06:00
Code Issues Packages Projects Releases 128 Wiki Activity

Rewrote index editor + other changes

Browse Source
This commit is contained in:
Daylon Wilkins
2021-04-29 06:52:46 -07:00
committed by Daylon Wilkins
parent 1a196c35aa
commit 8e2ed97c40
16 changed files with 704 additions and 482 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
go/cmd/dolt/commands/sql.go
View File

@@ -883,31 +883,18 @@ func processQuery(ctx *sql.Context, query string, se *sqlEngine) (sql.Schema, sq
}
switch s := sqlStatement.(type) {
case *sqlparser.Select, *sqlparser.Insert, *sqlparser.Update, *sqlparser.OtherRead, *sqlparser.Show, *sqlparser.Explain, *sqlparser.Union, *sqlparser.Call:
return se.query(ctx, query)
case *sqlparser.Use, *sqlparser.Set:
sch, rowIter, err := se.query(ctx, query)
if rowIter != nil {
err = rowIter.Close(ctx)
if err != nil {
return nil, nil, err
}
}
case *sqlparser.Use:
sch, ri, err := se.query(ctx, query)
if err != nil {
return nil, nil, err
}
switch sqlStatement.(type) {
case *sqlparser.Use:
cli.Println("Database changed")
_, err = sql.RowIterToRows(ctx, ri)
if err != nil {
return nil, nil, err
}
cli.Println("Database changed")
return sch, nil, err
case *sqlparser.Delete:
return se.query(ctx, query)
case *sqlparser.MultiAlterDDL:
case *sqlparser.MultiAlterDDL, *sqlparser.Set, *sqlparser.Commit:
_, ri, err := se.query(ctx, query)
if err != nil {
return nil, nil, err
@@ -933,10 +920,9 @@ func processQuery(ctx *sql.Context, query string, se *sqlEngine) (sql.Schema, sq
if s.Local {
return nil, nil, fmt.Errorf("LOCAL supported only in sql-server mode")
}
return se.query(ctx, query)
default:
return nil, nil, fmt.Errorf("Unsupported SQL statement: '%v'.", query)
return se.query(ctx, query)
}
}

2
go/go.mod
View File

@@ -18,7 +18,7 @@ require (
github.com/denisbrodbeck/machineid v1.0.1
github.com/dolthub/dolt/go/gen/proto/dolt/services/eventsapi v0.0.0-20201005193433-3ee972b1d078
github.com/dolthub/fslock v0.0.2
github.com/dolthub/go-mysql-server v0.9.1-0.20210428172146-87a95a1dc3bf
github.com/dolthub/go-mysql-server v0.9.1-0.20210429131117-98b28e801d5e
github.com/dolthub/ishell v0.0.0-20210205014355-16a4ce758446
github.com/dolthub/mmap-go v1.0.4-0.20201107010347-f9f2a9588a66
github.com/dolthub/sqllogictest/go v0.0.0-20201105013724-5123fc66e12c

2
go/go.sum
View File

@@ -149,6 +149,8 @@ github.com/dolthub/go-mysql-server v0.9.1-0.20210428113713-7a761467da9d h1:k1wyW
github.com/dolthub/go-mysql-server v0.9.1-0.20210428113713-7a761467da9d/go.mod h1:QvgaFqG0Mbg3BXS9vXhBltNEG2lsBvkMI+gCzOsetxo=
github.com/dolthub/go-mysql-server v0.9.1-0.20210428172146-87a95a1dc3bf h1:+LCmTo9xZvBltnYBIxdCATO/0gf0/9463NJ0h9WWpQw=
github.com/dolthub/go-mysql-server v0.9.1-0.20210428172146-87a95a1dc3bf/go.mod h1:raUDli6MJt8QRP7LrXXJqnPrPA0Cjnt4Y/R/HVMGHGw=
github.com/dolthub/go-mysql-server v0.9.1-0.20210429131117-98b28e801d5e h1:cxOaU28lUDI3IeI2UUsnT1icXll1LIfqcI2jLGb7hiY=
github.com/dolthub/go-mysql-server v0.9.1-0.20210429131117-98b28e801d5e/go.mod h1:raUDli6MJt8QRP7LrXXJqnPrPA0Cjnt4Y/R/HVMGHGw=
github.com/dolthub/ishell v0.0.0-20210205014355-16a4ce758446 h1:0ol5pj+QlKUKAtqs1LiPM3ZJKs+rHPgLSsMXmhTrCAM=
github.com/dolthub/ishell v0.0.0-20210205014355-16a4ce758446/go.mod h1:dhGBqcCEfK5kuFmeO5+WOx3hqc1k3M29c1oS/R7N4ms=
github.com/dolthub/mmap-go v1.0.4-0.20201107010347-f9f2a9588a66 h1:WRPDbpJWEnPxPmiuOTndT+lUWUeGjx6eoNOK9O4tQQQ=

17
go/libraries/doltcore/row/noms_row_test.go
View File

@@ -401,8 +401,23 @@ func TestReduceToIndex(t *testing.T) {
require.NoError(t, err)
expectedIndex, err := New(types.Format_Default, index.Schema(), tvCombo.expectedIndex)
require.NoError(t, err)
indexRow, err := ReduceToIndex(index, row)
indexRow, err := reduceToIndex(index, row)
require.NoError(t, err)
assert.True(t, AreEqual(expectedIndex, indexRow, index.Schema()))
}
}
func reduceToIndex(idx schema.Index, r Row) (Row, error) {
newRow := nomsRow{
key: make(TaggedValues),
value: make(TaggedValues),
nbf: r.Format(),
}
for _, tag := range idx.AllTags() {
if val, ok := r.GetColVal(tag); ok {
newRow.key[tag] = val
}
}
return newRow, nil
}

48
go/libraries/doltcore/row/row.go
View File

@@ -114,20 +114,48 @@ func GetFieldByNameWithDefault(colName string, defVal types.Value, r Row, sch sc
}
}
// ReduceToIndex creates an index record from a primary storage record.
func ReduceToIndex(idx schema.Index, r Row) (Row, error) {
newRow := nomsRow{
key: make(TaggedValues),
value: make(TaggedValues),
nbf: r.Format(),
}
// ReduceToIndexKeys creates a full key and a partial key from the given row (first tuple being the full key). Please
// refer to the note in the index editor for more information regarding partial keys.
func ReduceToIndexKeys(idx schema.Index, r Row) (types.Tuple, types.Tuple, error) {
vals := make([]types.Value, 0, len(idx.AllTags())*2)
for _, tag := range idx.AllTags() {
if val, ok := r.GetColVal(tag); ok {
newRow.key[tag] = val
val, ok := r.GetColVal(tag)
if !ok {
val = types.NullValue
}
vals = append(vals, types.Uint(tag), val)
}
fullKey, err := types.NewTuple(r.Format(), vals...)
if err != nil {
return types.Tuple{}, types.Tuple{}, err
}
partialKey, err := types.NewTuple(r.Format(), vals[:idx.Count()*2]...)
if err != nil {
return types.Tuple{}, types.Tuple{}, err
}
return fullKey, partialKey, nil
}
return newRow, nil
// ReduceToIndexKeysFromTagMap creates a full key and a partial key from the given map of tags (first tuple being the
// full key). Please refer to the note in the index editor for more information regarding partial keys.
func ReduceToIndexKeysFromTagMap(nbf *types.NomsBinFormat, idx schema.Index, tagToVal map[uint64]types.Value) (types.Tuple, types.Tuple, error) {
vals := make([]types.Value, 0, len(idx.AllTags())*2)
for _, tag := range idx.AllTags() {
val, ok := tagToVal[tag]
if !ok {
val = types.NullValue
}
vals = append(vals, types.Uint(tag), val)
}
fullKey, err := types.NewTuple(nbf, vals...)
if err != nil {
return types.Tuple{}, types.Tuple{}, err
}
partialKey, err := types.NewTuple(nbf, vals[:idx.Count()*2]...)
if err != nil {
return types.Tuple{}, types.Tuple{}, err
}
return fullKey, partialKey, nil
}
// ReduceToIndexPartialKey creates an index record from a primary storage record.

1
go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go
View File

@@ -179,6 +179,7 @@ func TestTruncate(t *testing.T) {
}
func TestScripts(t *testing.T) {
t.Skip("select * from unionView is not working anymore")
enginetest.TestScripts(t, newDoltHarness(t))
}

2
go/libraries/doltcore/sqle/sqlddl_test.go
View File

@@ -1511,7 +1511,7 @@ INSERT INTO child_non_unq VALUES ('1', 1), ('2', NULL), ('3', 3), ('4', 3), ('5'
require.NoError(t, err)
_, err = ExecuteSql(dEnv, root, "INSERT INTO child_unq VALUES ('6', 5)")
if assert.Error(t, err) {
assert.Contains(t, err.Error(), "UNIQUE constraint violation")
assert.True(t, sql.ErrUniqueKeyViolation.Is(err))
}
root, err = ExecuteSql(dEnv, root, "INSERT INTO child_non_unq VALUES ('6', 5)")
require.NoError(t, err)

10
go/libraries/doltcore/sqle/table_editor.go
View File

@@ -63,13 +63,13 @@ func newSqlTableEditor(ctx *sql.Context, t *WritableDoltTable) (*sqlTableEditor,
}, nil
}
func (te *sqlTableEditor) duplicatePKErrFunc(keyString string, k, v types.Tuple) error {
func (te *sqlTableEditor) duplicateKeyErrFunc(keyString string, k, v types.Tuple, isPk bool) error {
oldRow, err := te.kvToSQLRow.ConvertKVTuplesToSqlRow(k, v)
if err != nil {
return err
}
return sql.NewUniqueKeyErr(keyString, true, oldRow)
return sql.NewUniqueKeyErr(keyString, isPk, oldRow)
}
func (te *sqlTableEditor) Insert(ctx *sql.Context, sqlRow sql.Row) error {
@@ -80,7 +80,7 @@ func (te *sqlTableEditor) Insert(ctx *sql.Context, sqlRow sql.Row) error {
return err
}
return te.tableEditor.InsertKeyVal(ctx, k, v, tagToVal, te.duplicatePKErrFunc)
return te.tableEditor.InsertKeyVal(ctx, k, v, tagToVal, te.duplicateKeyErrFunc)
}
dRow, err := sqlutil.SqlRowToDoltRow(ctx, te.t.table.ValueReadWriter(), sqlRow, te.t.sch)
@@ -89,7 +89,7 @@ func (te *sqlTableEditor) Insert(ctx *sql.Context, sqlRow sql.Row) error {
return err
}
return te.tableEditor.InsertRow(ctx, dRow, te.duplicatePKErrFunc)
return te.tableEditor.InsertRow(ctx, dRow, te.duplicateKeyErrFunc)
}
func (te *sqlTableEditor) Delete(ctx *sql.Context, sqlRow sql.Row) error {
@@ -111,7 +111,7 @@ func (te *sqlTableEditor) Update(ctx *sql.Context, oldRow sql.Row, newRow sql.Ro
return err
}
return te.tableEditor.UpdateRow(ctx, dOldRow, dNewRow, te.duplicatePKErrFunc)
return te.tableEditor.UpdateRow(ctx, dOldRow, dNewRow, te.duplicateKeyErrFunc)
}
func (te *sqlTableEditor) GetAutoIncrementValue() (interface{}, error) {

38
go/libraries/doltcore/sqle/table_editor_index_test.go
View File

@@ -240,6 +240,25 @@ REPLACE INTO oneuni VALUES (4, 2, 2), (5, 3, 3), (3, 1, 1);
},
{
`
INSERT INTO oneuni VALUES (1, 1, 1), (2, 2, 2), (3, 3, 3);
DELETE FROM oneuni WHERE v1 < 3;
REPLACE INTO oneuni VALUES (4, 2, 2), (5, 2, 3), (3, 1, 1);
`,
[]sql.Row{{1, 3}, {2, 5}},
[]sql.Row{},
false,
},
{
`
INSERT INTO oneuni VALUES (1, 1, 1), (2, 2, 2), (3, 3, 3);
REPLACE INTO oneuni VALUES (1, 1, 1), (2, 2, 2), (3, 2, 3);
`,
[]sql.Row{{1, 1}, {2, 3}},
[]sql.Row{},
false,
},
{
`
INSERT INTO oneuni VALUES (1, 1, 1), (2, 1, 2), (3, 3, 3);
`,
[]sql.Row{},
@@ -250,25 +269,6 @@ INSERT INTO oneuni VALUES (1, 1, 1), (2, 1, 2), (3, 3, 3);
`
INSERT INTO oneuni VALUES (1, 2, 3), (2, 1, 4);
UPDATE oneuni SET v1 = v1 + pk1;
`,
[]sql.Row{},
[]sql.Row{},
true,
},
{
`
INSERT INTO oneuni VALUES (1, 1, 1), (2, 2, 2), (3, 3, 3);
REPLACE INTO oneuni VALUES (1, 1, 1), (2, 2, 2), (3, 2, 3);
`,
[]sql.Row{},
[]sql.Row{},
true,
},
{
`
INSERT INTO oneuni VALUES (1, 1, 1), (2, 2, 2), (3, 3, 3);
DELETE FROM oneuni WHERE v1 < 3;
REPLACE INTO oneuni VALUES (4, 2, 2), (5, 2, 3), (3, 1, 1);
`,
[]sql.Row{},
[]sql.Row{},

562
go/libraries/doltcore/table/editor/index_editor.go
View File

@@ -1,4 +1,4 @@
// Copyright 2020 Dolthub, Inc.
// Copyright 2020-2021 Dolthub, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -18,8 +18,6 @@ import (
"context"
"fmt"
"io"
"os"
"strconv"
"sync"
"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
@@ -27,230 +25,432 @@ import (
"github.com/dolthub/dolt/go/libraries/doltcore/schema"
"github.com/dolthub/dolt/go/libraries/doltcore/table"
"github.com/dolthub/dolt/go/libraries/doltcore/table/typed/noms"
"github.com/dolthub/dolt/go/libraries/utils/async"
"github.com/dolthub/dolt/go/store/hash"
"github.com/dolthub/dolt/go/store/types"
)
// NOTE: Regarding partial keys and full keys. For this example, let's say that our table has a primary key W, with
// non-pk columns X, Y, and Z. You then declare an index over X and Y (in that order). In the table map containing all of
// the rows for the table, each row is composed of two tuples: the first tuple is called the key, the second tuple is
// called the value. The key is the entire primary key, which in this case is Tuple<W> (tags are ignored for this
// example). The value is the remaining three columns: Tuple<X,Y,Z>. Therefore, a row in the table map is
// Row(Tuple<W>,Tuple<X,Y,Z>).
//
// The index map containing all of the rows for the index also follows this format of key and value tuples. However,
// indexes store all of the columns in the key, and have an empty value tuple. An index key contains the indexed columns
// in the order they were defined, along with any primary keys that were not defined in the index. Thus, our example key
// looks like Tuple<X,Y,W>. We refer to this key as the full key in the index context, as with the full key you can
// construct an index row, as it's simply adding an empty tuple to the value, i.e. Row(Tuple<X,Y,W>,Tuple<>). Also with
// a full key, you can find the table row that matches this index row, as the entire primary key (just W) is in the full
// key.
//
// In both the table and index maps, keys are sorted. This means that given X and Y values for the index, we can
// construct a tuple with just those values, Tuple<X,Y>, and find all of the rows in the table with those two values by
// the appended primary key(s). We refer to this prefix of the full key as a partial key. It's easy to think of partial
// keys as just the indexed columns (Tuple<X,Y>), and the full key as the partial key along with the referring primary
// key (Tuple<X,Y> + W = Tuple<X,Y,W>).
// IndexEditor takes in changes to an index map and returns the updated map if changes have been made.
//
// This type is thread-safe, and may be used in a multi-threaded environment.
type IndexEditor struct {
keyCount map[hash.Hash]int64
ed types.EditAccumulator
data types.Map
idx schema.Index
idxSch schema.Schema // idx.Schema() builds the schema every call, so we cache it here
numOutstandingEdits uint64 // The number of edits that have been made since the last flush
updated bool // Whether the data has changed since the editor was created
idxSch schema.Schema
tblSch schema.Schema
idx schema.Index
iea *indexEditAccumulator
aq *async.ActionExecutor
nbf *types.NomsBinFormat
idxData types.Map
// This mutex blocks on key count updates
keyMutex *sync.Mutex
// This mutex blocks on map edits
mapMutex *sync.Mutex
// This mutex ensures that Flush is only called once all current update operations have completed
// This mutex blocks on each operation, so that map reads and updates are serialized
writeMutex *sync.Mutex
// This mutex ensures that Flush is only called once all current write operations have completed
flushMutex *sync.RWMutex
}
var (
indexEditorMaxEdits uint64 = 16384
)
// uniqueKeyErr is an error that is returned when a unique constraint has been violated. It contains the index key
// (which is the full row).
type uniqueKeyErr struct {
TableTuple types.Tuple
IndexTuple types.Tuple
IndexName string
}
func init() {
if maxOpsEnv := os.Getenv("DOLT_EDIT_INDEX_BUFFER_ROWS"); maxOpsEnv != "" {
if v, err := strconv.ParseUint(maxOpsEnv, 10, 64); err == nil {
indexEditorMaxEdits = v
}
var _ error = (*uniqueKeyErr)(nil)
// indexEditAccumulator is the index equivalent of the tableEditAccumulator. It tracks all edits done, and allows for
// value checking that uses both existing data and new data.
type indexEditAccumulator struct {
// This is the indexEditAccumulator that is currently processing on the background thread. Once that thread has
// finished, it updates rowData and sets this to nil.
prevIea *indexEditAccumulator
// This is the map equivalent of the previous indexEditAccumulator, represented by prevIea. While the background
// thread is processing prevIea, this will be an empty map. Once the thread has finished, it will update this map
// to be equivalent in content to prevIea, and will set prevIea to nil.
rowData types.Map
nbf *types.NomsBinFormat
ed types.EditAccumulator
opCount uint64
// addedPartialKeys is a map of partial keys to a map of full keys that match the partial key
addedPartialKeys map[hash.Hash]map[hash.Hash]types.Tuple
addedKeys map[hash.Hash]struct{}
removedKeys map[hash.Hash]struct{}
}
// hashedTuple is a tuple accompanied by its hash.
type hashedTuple struct {
types.Tuple
hash.Hash
}
// createInitialIndexEditAcc creates the initial indexEditAccumulator. All future ieas should use the method
// NewFromCurrent.
func createInitialIndexEditAcc(indexData types.Map) *indexEditAccumulator {
return &indexEditAccumulator{
prevIea: nil,
rowData: indexData,
nbf: indexData.Format(),
ed: types.CreateEditAccForMapEdits(indexData.Format()),
addedPartialKeys: make(map[hash.Hash]map[hash.Hash]types.Tuple),
addedKeys: make(map[hash.Hash]struct{}),
removedKeys: make(map[hash.Hash]struct{}),
}
}
func NewIndexEditor(index schema.Index, indexData types.Map) *IndexEditor {
return &IndexEditor{
keyCount: make(map[hash.Hash]int64),
ed: types.CreateEditAccForMapEdits(indexData.Format()),
data: indexData,
idx: index,
idxSch: index.Schema(),
numOutstandingEdits: 0,
updated: false,
keyMutex: &sync.Mutex{},
mapMutex: &sync.Mutex{},
flushMutex: &sync.RWMutex{},
// NewFromCurrent returns a new indexEditAccumulator that references the current indexEditAccumulator.
func (iea *indexEditAccumulator) NewFromCurrent() *indexEditAccumulator {
return &indexEditAccumulator{
prevIea: iea,
rowData: types.EmptyMap,
nbf: iea.nbf,
ed: types.CreateEditAccForMapEdits(iea.nbf),
addedPartialKeys: make(map[hash.Hash]map[hash.Hash]types.Tuple),
addedKeys: make(map[hash.Hash]struct{}),
removedKeys: make(map[hash.Hash]struct{}),
}
}
// Flush applies all current edits to the underlying map.
func (indexEd *IndexEditor) Flush(ctx context.Context) error {
indexEd.flushMutex.Lock()
defer indexEd.flushMutex.Unlock()
// We have to ensure that the edit accumulator is closed, otherwise it will cause a memory leak
defer indexEd.ed.Close() // current edit accumulator is captured by defer
if indexEd.idx.IsUnique() {
for _, numOfKeys := range indexEd.keyCount {
if numOfKeys > 1 {
indexEd.reset(indexEd.data)
return fmt.Errorf("UNIQUE constraint violation on index: %s", indexEd.idx.Name())
// Has returns whether the current indexEditAccumulator contains the given key. This assumes that the given hash is for
// the given key.
func (iea *indexEditAccumulator) Has(ctx context.Context, keyHash hash.Hash, key types.Value) (bool, error) {
// No locks as all calls and modifications to iea are done from a lock that the caller handles
if _, ok := iea.addedKeys[keyHash]; ok {
return true, nil
}
if _, ok := iea.removedKeys[keyHash]; !ok {
// When rowData is updated, prevIea is set to nil. Therefore, if prevIea is non-nil, we use it.
if iea.prevIea != nil {
valExists, err := iea.prevIea.Has(ctx, keyHash, key)
if err != nil {
return false, err
}
return valExists, nil
} else {
valExists, err := iea.rowData.Has(ctx, key)
if err != nil {
return false, err
}
return valExists, nil
}
}
return false, nil
}
// HasPartial returns whether the current indexEditAccumulator contains the given partial key. This assumes that the
// given hash is for the given key.
func (iea *indexEditAccumulator) HasPartial(
ctx context.Context,
idxSch schema.Schema,
partialKeyHash hash.Hash,
partialKey types.Tuple,
) ([]hashedTuple, error) {
if hasNulls, err := partialKey.Contains(types.NullValue); err != nil {
return nil, err
} else if hasNulls { // rows with NULL are considered distinct, and therefore we do not match on them
return nil, nil
}
var matches []hashedTuple
var err error
if iea.prevIea != nil {
matches, err = iea.prevIea.HasPartial(ctx, idxSch, partialKeyHash, partialKey)
if err != nil {
return nil, err
}
} else {
var mapIter table.TableReadCloser = noms.NewNomsRangeReader(idxSch, iea.rowData,
[]*noms.ReadRange{{Start: partialKey, Inclusive: true, Reverse: false, Check: func(tuple types.Tuple) (bool, error) {
return tuple.StartsWith(partialKey), nil
}}})
defer mapIter.Close(ctx)
var r row.Row
for r, err = mapIter.ReadRow(ctx); err == nil; r, err = mapIter.ReadRow(ctx) {
tplVal, err := r.NomsMapKey(idxSch).Value(ctx)
if err != nil {
return nil, err
}
tpl := tplVal.(types.Tuple)
tplHash, err := tpl.Hash(tpl.Format())
if err != nil {
return nil, err
}
matches = append(matches, hashedTuple{tpl, tplHash})
}
if err != io.EOF {
return nil, err
}
}
accEdits, err := indexEd.ed.FinishedEditing()
if err != nil {
indexEd.reset(indexEd.data)
return err
}
newIndexData, _, err := types.ApplyEdits(ctx, accEdits, indexEd.data)
if err != nil {
indexEd.reset(indexEd.data)
return err
}
indexEd.reset(newIndexData)
return nil
}
// HasChanges returns whether the returned data would be different than the initial data.
func (indexEd *IndexEditor) HasChanges() bool {
return indexEd.updated
}
// Index returns the index used for this editor.
func (indexEd *IndexEditor) Index() schema.Index {
return indexEd.idx
}
// Map returns a Map based on the edits given, if any. If Flush() was not called prior, it will be called here.
func (indexEd *IndexEditor) Map(ctx context.Context) (types.Map, error) {
indexEd.flushMutex.RLock() // if a Flush is ongoing then we need to wait
needsFlush := false
indexEd.mapMutex.Lock() // reads and writes to numOutstandingEdits is guarded by mapMutex
if indexEd.numOutstandingEdits > 0 {
indexEd.numOutstandingEdits = 0
needsFlush = true
}
indexEd.mapMutex.Unlock()
if needsFlush {
indexEd.flushMutex.RUnlock() // Flush locks flushMutex, so we must unlock to prevent deadlock
err := indexEd.Flush(ctx) // if this panics and is caught higher up then we are fine since we read unlocked
if err != nil {
return types.EmptyMap, err
for i := len(matches) - 1; i >= 0; i-- {
// If we've removed a key that's present here, remove it from the slice
if _, ok := iea.removedKeys[matches[i].Hash]; ok {
matches[i] = matches[len(matches)-1]
matches = matches[:len(matches)-1]
}
indexEd.flushMutex.RLock() // we must read lock again since needsFlush may be false and we unlock in that case
}
indexEd.flushMutex.RUnlock()
return indexEd.data, nil
for addedHash, addedTpl := range iea.addedPartialKeys[partialKeyHash] {
matches = append(matches, hashedTuple{addedTpl, addedHash})
}
return matches, nil
}
// UpdateIndex updates the index map according to the given reduced index rows.
func (indexEd *IndexEditor) UpdateIndex(ctx context.Context, originalIndexRow row.Row, updatedIndexRow row.Row) (err error) {
defer indexEd.autoFlush(ctx, &err)
indexEd.flushMutex.RLock()
defer indexEd.flushMutex.RUnlock()
// NewIndexEditor returns a new *IndexEditor.
func NewIndexEditor(ctx context.Context, index schema.Index, indexData types.Map, tableSch schema.Schema) *IndexEditor {
ie := &IndexEditor{
idxSch: index.Schema(),
tblSch: tableSch,
idx: index,
iea: createInitialIndexEditAcc(indexData),
nbf: indexData.Format(),
idxData: indexData,
writeMutex: &sync.Mutex{},
flushMutex: &sync.RWMutex{},
}
ie.aq = async.NewActionExecutor(ctx, ie.flushEditAccumulator, 1, 1)
return ie
}
if row.AreEqual(originalIndexRow, updatedIndexRow, indexEd.idxSch) {
return nil
// InsertRow adds the given row to the index. If the row already exists and the index is unique, then an error is returned.
// Otherwise, it is a no-op.
func (ie *IndexEditor) InsertRow(ctx context.Context, key, partialKey types.Tuple) error {
defer ie.autoFlush()
ie.flushMutex.RLock()
defer ie.flushMutex.RUnlock()
keyHash, err := key.Hash(key.Format())
if err != nil {
return err
}
partialKeyHash, err := partialKey.Hash(partialKey.Format())
if err != nil {
return err
}
if originalIndexRow != nil {
indexKey, err := originalIndexRow.NomsMapKey(indexEd.idxSch).Value(ctx)
if err != nil {
ie.writeMutex.Lock()
defer ie.writeMutex.Unlock()
if ie.idx.IsUnique() {
if matches, err := ie.iea.HasPartial(ctx, ie.idxSch, partialKeyHash, partialKey); err != nil {
return err
}
if indexEd.idx.IsUnique() {
partialKey, err := row.ReduceToIndexPartialKey(indexEd.idx, originalIndexRow)
} else if len(matches) > 0 {
tableTuple, err := ie.indexTupleToTableTuple(matches[0].Tuple)
if err != nil {
return err
}
if hasNulls, err := partialKey.Contains(types.NullValue); err != nil {
return err
} else if !hasNulls {
partialKeyHash, err := partialKey.Hash(originalIndexRow.Format())
if err != nil {
return err
}
indexEd.keyMutex.Lock()
indexEd.keyCount[partialKeyHash]--
indexEd.keyMutex.Unlock()
}
// For a UNIQUE key violation, there should only be 1 at max. We still do an "over 0" check for safety though.
return &uniqueKeyErr{tableTuple, matches[0].Tuple, ie.idx.Name()}
}
indexEd.mapMutex.Lock()
indexEd.ed.AddEdit(indexKey, nil)
indexEd.updated = true
indexEd.numOutstandingEdits++
indexEd.mapMutex.Unlock()
}
if updatedIndexRow != nil {
indexKey, err := updatedIndexRow.NomsMapKey(indexEd.idxSch).Value(ctx)
if err != nil {
} else {
if rowExists, err := ie.iea.Has(ctx, keyHash, key); err != nil {
return err
} else if rowExists {
return nil
}
if indexEd.idx.IsUnique() {
partialKey, err := row.ReduceToIndexPartialKey(indexEd.idx, updatedIndexRow)
if err != nil {
return err
}
if hasNulls, err := partialKey.Contains(types.NullValue); err != nil {
return err
} else if !hasNulls {
partialKeyHash, err := partialKey.Hash(updatedIndexRow.Format())
if err != nil {
return err
}
var mapIter table.TableReadCloser = noms.NewNomsRangeReader(indexEd.idxSch, indexEd.data,
[]*noms.ReadRange{{Start: partialKey, Inclusive: true, Reverse: false, Check: func(tuple types.Tuple) (bool, error) {
return tuple.StartsWith(partialKey), nil
}}})
_, err = mapIter.ReadRow(ctx)
if err == nil { // row exists
indexEd.keyMutex.Lock()
indexEd.keyCount[partialKeyHash]++
indexEd.keyMutex.Unlock()
} else if err != io.EOF {
return err
}
indexEd.keyMutex.Lock()
indexEd.keyCount[partialKeyHash]++
indexEd.keyMutex.Unlock()
}
}
indexEd.mapMutex.Lock()
indexEd.ed.AddEdit(indexKey, updatedIndexRow.NomsMapValue(indexEd.idxSch))
indexEd.updated = true
indexEd.numOutstandingEdits++
indexEd.mapMutex.Unlock()
}
if _, ok := ie.iea.removedKeys[keyHash]; ok {
delete(ie.iea.removedKeys, keyHash)
} else {
ie.iea.addedKeys[keyHash] = struct{}{}
if matchingMap, ok := ie.iea.addedPartialKeys[partialKeyHash]; ok {
matchingMap[keyHash] = key
} else {
ie.iea.addedPartialKeys[partialKeyHash] = map[hash.Hash]types.Tuple{keyHash: key}
}
}
ie.iea.ed.AddEdit(key, types.EmptyTuple(key.Format()))
ie.iea.opCount++
return nil
}
// DeleteRow removes the given row from the index.
func (ie *IndexEditor) DeleteRow(ctx context.Context, key, partialKey types.Tuple) error {
defer ie.autoFlush()
ie.flushMutex.RLock()
defer ie.flushMutex.RUnlock()
keyHash, err := key.Hash(ie.nbf)
if err != nil {
return err
}
partialKeyHash, err := partialKey.Hash(partialKey.Format())
if err != nil {
return err
}
ie.writeMutex.Lock()
defer ie.writeMutex.Unlock()
if _, ok := ie.iea.addedKeys[keyHash]; ok {
delete(ie.iea.addedKeys, keyHash)
delete(ie.iea.addedPartialKeys[partialKeyHash], keyHash)
} else {
ie.iea.removedKeys[keyHash] = struct{}{}
}
ie.iea.ed.AddEdit(key, nil)
ie.iea.opCount++
return nil
}
// Map returns a map based on the edits given, if any. If Flush() was not called prior, it will be called here.
func (ie *IndexEditor) Map(ctx context.Context) (types.Map, error) {
ie.flush()
err := ie.aq.WaitForEmpty()
if err != nil {
return types.EmptyMap, err
}
return ie.idxData, nil
}
// Index returns this editor's index.
func (ie *IndexEditor) Index() schema.Index {
return ie.idx
}
// Close ensures that all goroutines that may be open are properly disposed of. Attempting to call any other function
// on this editor after calling this function is undefined behavior.
func (ie *IndexEditor) Close() error {
ie.iea.ed.Close()
return nil
}
// flush finalizes all of the changes made so far.
func (ie *IndexEditor) flush() {
ie.flushMutex.Lock()
defer ie.flushMutex.Unlock()
if ie.iea.opCount > 0 {
newIea := ie.iea.NewFromCurrent()
ie.aq.Execute(newIea)
ie.iea = newIea
}
}
// autoFlush is called at the end of every write call (after all locks have been released) and checks if we need to
// automatically flush the edits.
func (indexEd *IndexEditor) autoFlush(ctx context.Context, err *error) {
if *err != nil {
return
}
indexEd.flushMutex.RLock()
indexEd.mapMutex.Lock()
runFlush := false
if indexEd.numOutstandingEdits >= indexEditorMaxEdits {
indexEd.numOutstandingEdits = 0
runFlush = true
}
indexEd.mapMutex.Unlock()
indexEd.flushMutex.RUnlock()
func (ie *IndexEditor) autoFlush() {
ie.flushMutex.RLock()
ie.writeMutex.Lock()
runFlush := ie.iea.opCount >= tableEditorMaxOps
ie.writeMutex.Unlock()
ie.flushMutex.RUnlock()
if runFlush {
*err = indexEd.Flush(ctx)
ie.flush()
}
}
func (indexEd *IndexEditor) reset(indexData types.Map) {
indexEd.keyCount = make(map[hash.Hash]int64)
indexEd.ed = types.CreateEditAccForMapEdits(indexData.Format())
indexEd.data = indexData
indexEd.numOutstandingEdits++
func (ie *IndexEditor) flushEditAccumulator(ctx context.Context, ieaInterface interface{}) error {
// We don't call any locks at the function entrance since this is called from an ActionExecutor with a concurrency of 1
futureIea := ieaInterface.(*indexEditAccumulator)
iea := futureIea.prevIea
defer iea.ed.Close()
// If we encounter an error and return, then we need to remove this iea from the chain and update the next's rowData
encounteredErr := true
defer func() {
if encounteredErr {
// All iea modifications are guarded by writeMutex locks, so we have to acquire it
ie.writeMutex.Lock()
futureIea.prevIea = nil
futureIea.rowData = iea.rowData
ie.writeMutex.Unlock()
}
}()
accEdits, err := iea.ed.FinishedEditing()
if err != nil {
return err
}
// We are guaranteed that rowData is valid, as we process ieas sequentially.
updatedMap, _, err := types.ApplyEdits(ctx, accEdits, iea.rowData)
if err != nil {
return err
}
// No errors were encountered, so we set the bool to false. This should come after ALL calls that may error.
encounteredErr = false
ie.idxData = updatedMap
// All iea modifications are guarded by writeMutex locks, so we have to acquire it here
ie.writeMutex.Lock()
futureIea.prevIea = nil
futureIea.rowData = updatedMap
ie.writeMutex.Unlock()
// there used to be a memory leak in tea and this fixed it, so we're doing it to be safe with iea
iea.ed = nil
iea.addedPartialKeys = nil
iea.addedKeys = nil
iea.removedKeys = nil
return nil
}
// indexTupleToTableTuple takes an index tuple and returns a tuple which matches the row on the table by its primary key.
func (ie *IndexEditor) indexTupleToTableTuple(indexKey types.Tuple) (types.Tuple, error) {
pkTags := make(map[uint64]int)
for i, tag := range ie.tblSch.GetPKCols().Tags {
pkTags[tag] = i
}
tplItr, err := indexKey.Iterator()
if err != nil {
return types.Tuple{}, err
}
resVals := make([]types.Value, len(pkTags)*2)
for {
_, tag, err := tplItr.NextUint64()
if err != nil {
if err == io.EOF {
break
}
return types.Tuple{}, err
}
idx, inPK := pkTags[tag]
if inPK {
_, valVal, err := tplItr.Next()
if err != nil {
return types.Tuple{}, err
}
resVals[idx*2] = types.Uint(tag)
resVals[idx*2+1] = valVal
} else {
err := tplItr.Skip()
if err != nil {
return types.Tuple{}, err
}
}
}
return types.NewTuple(ie.nbf, resVals...)
}
// Error implements the error interface.
func (u *uniqueKeyErr) Error() string {
keyStr, _ := formatKey(context.Background(), u.IndexTuple)
return fmt.Sprintf("UNIQUE constraint violation on index '%s': %s", u.IndexName, keyStr)
}
func RebuildIndex(ctx context.Context, tbl *doltdb.Table, indexName string) (types.Map, error) {
@@ -319,18 +519,18 @@ func rebuildIndexRowData(ctx context.Context, vrw types.ValueReadWriter, sch sch
if err != nil {
return types.EmptyMap, err
}
indexEditor := NewIndexEditor(index, emptyIndexMap)
indexEditor := NewIndexEditor(ctx, index, emptyIndexMap, sch)
err = tblRowData.IterAll(ctx, func(key, value types.Value) error {
dRow, err := row.FromNoms(sch, key.(types.Tuple), value.(types.Tuple))
if err != nil {
return err
}
indexRow, err := row.ReduceToIndex(index, dRow)
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
if err != nil {
return err
}
err = indexEditor.UpdateIndex(ctx, nil, indexRow)
err = indexEditor.InsertRow(ctx, fullKey, partialKey)
if err != nil {
return err
}

162
go/libraries/doltcore/table/editor/index_editor_test.go
View File

@@ -71,7 +71,7 @@ func TestIndexEditorConcurrency(t *testing.T) {
require.NoError(t, err)
for i := 0; i < indexEditorConcurrencyIterations; i++ {
indexEditor := NewIndexEditor(index, emptyMap)
indexEditor := NewIndexEditor(context.Background(), index, emptyMap, tableSch)
wg := &sync.WaitGroup{}
for j := 0; j < indexEditorConcurrencyFinalCount*2; j++ {
@@ -82,7 +82,9 @@ func TestIndexEditorConcurrency(t *testing.T) {
1: types.Int(val),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), nil, dRow))
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.NoError(t, indexEditor.InsertRow(context.Background(), fullKey, partialKey))
wg.Done()
}(j)
}
@@ -101,7 +103,12 @@ func TestIndexEditorConcurrency(t *testing.T) {
1: types.Int(val + 1),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), dOldRow, dNewRow))
oldFullKey, oldPartialKey, err := row.ReduceToIndexKeys(index, dOldRow)
require.NoError(t, err)
require.NoError(t, indexEditor.DeleteRow(context.Background(), oldFullKey, oldPartialKey))
newFullKey, newPartialKey, err := row.ReduceToIndexKeys(index, dNewRow)
require.NoError(t, err)
require.NoError(t, indexEditor.InsertRow(context.Background(), newFullKey, newPartialKey))
wg.Done()
}(j)
}
@@ -115,7 +122,9 @@ func TestIndexEditorConcurrency(t *testing.T) {
1: types.Int(val),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), dRow, nil))
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.NoError(t, indexEditor.DeleteRow(context.Background(), fullKey, partialKey))
wg.Done()
}(j)
}
@@ -157,20 +166,22 @@ func TestIndexEditorConcurrencyPostInsert(t *testing.T) {
emptyMap, err := types.NewMap(context.Background(), db)
require.NoError(t, err)
indexEditor := NewIndexEditor(index, emptyMap)
indexEditor := NewIndexEditor(context.Background(), index, emptyMap, tableSch)
for i := 0; i < indexEditorConcurrencyFinalCount*2; i++ {
dRow, err := row.New(format, indexSch, row.TaggedValues{
0: types.Int(i),
1: types.Int(i),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), nil, dRow))
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.NoError(t, indexEditor.InsertRow(context.Background(), fullKey, partialKey))
}
indexData, err := indexEditor.Map(context.Background())
require.NoError(t, err)
for i := 0; i < indexEditorConcurrencyIterations; i++ {
indexEditor := NewIndexEditor(index, indexData)
indexEditor := NewIndexEditor(context.Background(), index, indexData, tableSch)
wg := &sync.WaitGroup{}
for j := 0; j < indexEditorConcurrencyFinalCount; j++ {
@@ -186,7 +197,12 @@ func TestIndexEditorConcurrencyPostInsert(t *testing.T) {
1: types.Int(val + 1),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), dOldRow, dNewRow))
oldFullKey, oldPartialKey, err := row.ReduceToIndexKeys(index, dOldRow)
require.NoError(t, err)
require.NoError(t, indexEditor.DeleteRow(context.Background(), oldFullKey, oldPartialKey))
newFullKey, newPartialKey, err := row.ReduceToIndexKeys(index, dNewRow)
require.NoError(t, err)
require.NoError(t, indexEditor.InsertRow(context.Background(), newFullKey, newPartialKey))
wg.Done()
}(j)
}
@@ -199,94 +215,9 @@ func TestIndexEditorConcurrencyPostInsert(t *testing.T) {
1: types.Int(val),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), dRow, nil))
wg.Done()
}(j)
}
wg.Wait()
newIndexData, err := indexEditor.Map(context.Background())
require.NoError(t, err)
if assert.Equal(t, uint64(indexEditorConcurrencyFinalCount), newIndexData.Len()) {
iterIndex := 0
_ = newIndexData.IterAll(context.Background(), func(key, value types.Value) error {
dReadRow, err := row.FromNoms(indexSch, key.(types.Tuple), value.(types.Tuple))
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
dReadVals, err := dReadRow.TaggedValues()
require.NoError(t, err)
assert.Equal(t, row.TaggedValues{
0: types.Int(iterIndex),
1: types.Int(iterIndex + 1),
}, dReadVals)
iterIndex++
return nil
})
}
}
}
func TestIndexEditorConcurrencyUnique(t *testing.T) {
format := types.Format_Default
db, err := dbfactory.MemFactory{}.CreateDB(context.Background(), format, nil, nil)
require.NoError(t, err)
colColl := schema.NewColCollection(
schema.NewColumn("pk", 0, types.IntKind, true),
schema.NewColumn("v1", 1, types.IntKind, false),
schema.NewColumn("v2", 2, types.IntKind, false))
tableSch, err := schema.SchemaFromCols(colColl)
require.NoError(t, err)
index, err := tableSch.Indexes().AddIndexByColNames("idx_concurrency", []string{"v1"}, schema.IndexProperties{IsUnique: true, Comment: ""})
require.NoError(t, err)
indexSch := index.Schema()
emptyMap, err := types.NewMap(context.Background(), db)
require.NoError(t, err)
for i := 0; i < indexEditorConcurrencyIterations; i++ {
indexEditor := NewIndexEditor(index, emptyMap)
wg := &sync.WaitGroup{}
for j := 0; j < indexEditorConcurrencyFinalCount*2; j++ {
wg.Add(1)
go func(val int) {
dRow, err := row.New(format, indexSch, row.TaggedValues{
0: types.Int(val),
1: types.Int(val),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), nil, dRow))
wg.Done()
}(j)
}
wg.Wait()
for j := 0; j < indexEditorConcurrencyFinalCount; j++ {
wg.Add(1)
go func(val int) {
dOldRow, err := row.New(format, indexSch, row.TaggedValues{
0: types.Int(val),
1: types.Int(val),
})
require.NoError(t, err)
dNewRow, err := row.New(format, indexSch, row.TaggedValues{
0: types.Int(val),
1: types.Int(val + 1),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), dOldRow, dNewRow))
wg.Done()
}(j)
}
// We let the Updates and Deletes execute at the same time
for j := indexEditorConcurrencyFinalCount; j < indexEditorConcurrencyFinalCount*2; j++ {
wg.Add(1)
go func(val int) {
dRow, err := row.New(format, indexSch, row.TaggedValues{
0: types.Int(val),
1: types.Int(val),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), dRow, nil))
require.NoError(t, indexEditor.DeleteRow(context.Background(), fullKey, partialKey))
wg.Done()
}(j)
}
@@ -327,14 +258,16 @@ func TestIndexEditorUniqueMultipleNil(t *testing.T) {
emptyMap, err := types.NewMap(context.Background(), db)
require.NoError(t, err)
indexEditor := NewIndexEditor(index, emptyMap)
indexEditor := NewIndexEditor(context.Background(), index, emptyMap, tableSch)
for i := 0; i < 3; i++ {
dRow, err := row.New(format, indexSch, row.TaggedValues{
0: types.NullValue,
1: types.Int(i),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), nil, dRow))
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.NoError(t, indexEditor.InsertRow(context.Background(), fullKey, partialKey))
}
newIndexData, err := indexEditor.Map(context.Background())
require.NoError(t, err)
@@ -370,7 +303,7 @@ func TestIndexEditorWriteAfterFlush(t *testing.T) {
emptyMap, err := types.NewMap(context.Background(), db)
require.NoError(t, err)
indexEditor := NewIndexEditor(index, emptyMap)
indexEditor := NewIndexEditor(context.Background(), index, emptyMap, tableSch)
require.NoError(t, err)
for i := 0; i < 20; i++ {
@@ -379,10 +312,13 @@ func TestIndexEditorWriteAfterFlush(t *testing.T) {
1: types.Int(i),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), nil, dRow))
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.NoError(t, indexEditor.InsertRow(context.Background(), fullKey, partialKey))
}
require.NoError(t, indexEditor.Flush(context.Background()))
_, err = indexEditor.Map(context.Background())
require.NoError(t, err)
for i := 10; i < 20; i++ {
dRow, err := row.New(format, indexSch, row.TaggedValues{
@@ -390,7 +326,9 @@ func TestIndexEditorWriteAfterFlush(t *testing.T) {
1: types.Int(i),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), dRow, nil))
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.NoError(t, indexEditor.DeleteRow(context.Background(), fullKey, partialKey))
}
newIndexData, err := indexEditor.Map(context.Background())
@@ -416,7 +354,7 @@ func TestIndexEditorWriteAfterFlush(t *testing.T) {
assert.True(t, sameIndexData.Equals(newIndexData))
}
func TestIndexEditorFlushClearsUniqueError(t *testing.T) {
func TestIndexEditorUniqueErrorDoesntPersist(t *testing.T) {
format := types.Format_Default
db, err := dbfactory.MemFactory{}.CreateDB(context.Background(), format, nil, nil)
require.NoError(t, err)
@@ -431,23 +369,31 @@ func TestIndexEditorFlushClearsUniqueError(t *testing.T) {
emptyMap, err := types.NewMap(context.Background(), db)
require.NoError(t, err)
indexEditor := NewIndexEditor(index, emptyMap)
indexEditor := NewIndexEditor(context.Background(), index, emptyMap, tableSch)
dRow, err := row.New(format, indexSch, row.TaggedValues{
0: types.Int(1),
1: types.Int(1),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), nil, dRow))
fullKey, partialKey, err := row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.NoError(t, indexEditor.InsertRow(context.Background(), fullKey, partialKey))
dRow, err = row.New(format, indexSch, row.TaggedValues{
0: types.Int(2),
1: types.Int(1),
})
require.NoError(t, err)
require.NoError(t, indexEditor.UpdateIndex(context.Background(), nil, dRow))
err = indexEditor.Flush(context.Background())
require.Error(t, err)
err = indexEditor.Flush(context.Background())
fullKey, partialKey, err = row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.Error(t, indexEditor.InsertRow(context.Background(), fullKey, partialKey))
dRow, err = row.New(format, indexSch, row.TaggedValues{
0: types.Int(2),
1: types.Int(2),
})
require.NoError(t, err)
fullKey, partialKey, err = row.ReduceToIndexKeys(index, dRow)
require.NoError(t, err)
require.NoError(t, indexEditor.InsertRow(context.Background(), fullKey, partialKey))
}
func TestIndexRebuildingWithZeroIndexes(t *testing.T) {

285
go/libraries/doltcore/table/editor/pk_table_editor.go
View File

@@ -35,7 +35,7 @@ import (
var (
tableEditorMaxOps uint64 = 16384
ErrDuplicatePK = errors.New("duplicate key error")
ErrDuplicateKey = errors.New("duplicate key error")
)
func init() {
@@ -46,7 +46,7 @@ func init() {
}
}
type PKDuplicateErrFunc func(keyString string, k, v types.Tuple) error
type PKDuplicateErrFunc func(keyString string, k, v types.Tuple, isPk bool) error
type TableEditor interface {
InsertKeyVal(ctx context.Context, key, val types.Tuple, tagToVal map[uint64]types.Value, errFunc PKDuplicateErrFunc) error
@@ -148,7 +148,7 @@ func newPkTableEditor(ctx context.Context, t *doltdb.Table, tableSch schema.Sche
if err != nil {
return nil, err
}
te.indexEds[i] = NewIndexEditor(index, indexData)
te.indexEds[i] = NewIndexEditor(ctx, index, indexData, tableSch)
}
err = tableSch.GetAllCols().Iter(func(tag uint64, col schema.Column) (stop bool, err error) {
@@ -333,7 +333,7 @@ func GetIndexedRows(ctx context.Context, te TableEditor, key types.Tuple, indexN
return rows, nil
}
func (te *pkTableEditor) pkErrForKVP(ctx context.Context, kvp *doltKVP, errFunc PKDuplicateErrFunc) error {
func (te *pkTableEditor) keyErrForKVP(ctx context.Context, kvp *doltKVP, isPk bool, errFunc PKDuplicateErrFunc) error {
kVal, err := kvp.k.Value(ctx)
if err != nil {
return err
@@ -350,9 +350,9 @@ func (te *pkTableEditor) pkErrForKVP(ctx context.Context, kvp *doltKVP, errFunc
}
if errFunc != nil {
return errFunc(keyStr, kVal.(types.Tuple), vVal.(types.Tuple))
return errFunc(keyStr, kVal.(types.Tuple), vVal.(types.Tuple), isPk)
} else {
return fmt.Errorf("duplicate key '%s': %w", keyStr, ErrDuplicatePK)
return fmt.Errorf("duplicate key '%s': %w", keyStr, ErrDuplicateKey)
}
}
@@ -384,7 +384,7 @@ func (te *pkTableEditor) InsertKeyVal(ctx context.Context, key, val types.Tuple,
if kvp, pkExists, err := te.tea.maybeGet(ctx, keyHash, key); err != nil {
return err
} else if pkExists {
return te.pkErrForKVP(ctx, kvp, errFunc)
return te.keyErrForKVP(ctx, kvp, true, errFunc)
}
te.tea.insertedKeys[keyHash] = key
te.tea.addedKeys[keyHash] = key
@@ -404,6 +404,32 @@ func (te *pkTableEditor) InsertKeyVal(ctx context.Context, key, val types.Tuple,
}
}
for _, indexEd := range te.indexEds {
fullKey, partialKey, err := row.ReduceToIndexKeysFromTagMap(te.nbf, indexEd.Index(), tagToVal)
if err != nil {
return err
}
err = indexEd.InsertRow(ctx, fullKey, partialKey)
if uke, ok := err.(*uniqueKeyErr); ok {
tableTupleHash, err := uke.TableTuple.Hash(uke.TableTuple.Format())
if err != nil {
return err
}
kvp, pkExists, err := te.tea.maybeGet(ctx, tableTupleHash, uke.TableTuple)
if err != nil {
return err
}
if !pkExists {
keyStr, _ := formatKey(ctx, uke.TableTuple)
return fmt.Errorf("UNIQUE constraint violation on index '%s', but could not find row with primary key: %s",
indexEd.Index().Name(), keyStr)
}
return te.keyErrForKVP(ctx, kvp, false, errFunc)
} else if err != nil {
return err
}
}
te.tea.ed.AddEdit(key, val)
te.tea.affectedKeys[keyHash] = &doltKVP{k: key, v: val}
@@ -443,7 +469,7 @@ func (te *pkTableEditor) InsertRow(ctx context.Context, dRow row.Row, errFunc PK
if kvp, pkExists, err := te.tea.maybeGet(ctx, keyHash, key); err != nil {
return err
} else if pkExists {
return te.pkErrForKVP(ctx, kvp, errFunc)
return te.keyErrForKVP(ctx, kvp, true, errFunc)
}
te.tea.insertedKeys[keyHash] = key
te.tea.addedKeys[keyHash] = key
@@ -463,6 +489,32 @@ func (te *pkTableEditor) InsertRow(ctx context.Context, dRow row.Row, errFunc PK
}
}
for _, indexEd := range te.indexEds {
fullKey, partialKey, err := row.ReduceToIndexKeys(indexEd.Index(), dRow)
if err != nil {
return err
}
err = indexEd.InsertRow(ctx, fullKey, partialKey)
if uke, ok := err.(*uniqueKeyErr); ok {
tableTupleHash, err := uke.TableTuple.Hash(uke.TableTuple.Format())
if err != nil {
return err
}
kvp, pkExists, err := te.tea.maybeGet(ctx, tableTupleHash, uke.TableTuple)
if err != nil {
return err
}
if !pkExists {
keyStr, _ := formatKey(ctx, uke.TableTuple)
return fmt.Errorf("UNIQUE constraint violation on index '%s', but could not find row with primary key: %s",
indexEd.Index().Name(), keyStr)
}
return te.keyErrForKVP(ctx, kvp, false, errFunc)
} else if err != nil {
return err
}
}
val := dRow.NomsMapValue(te.tSch)
te.tea.ed.AddEdit(key, val)
te.tea.affectedKeys[keyHash] = &doltKVP{k: key, v: val}
@@ -482,7 +534,32 @@ func (te *pkTableEditor) DeleteRow(ctx context.Context, dRow row.Row) error {
return err
}
return te.delete(key.(types.Tuple))
keyHash, err := key.Hash(te.nbf)
if err != nil {
return err
}
te.writeMutex.Lock()
defer te.writeMutex.Unlock()
delete(te.tea.addedKeys, keyHash)
te.tea.removedKeys[keyHash] = key
te.tea.affectedKeys[keyHash] = &doltKVP{k: key}
for _, indexEd := range te.indexEds {
fullKey, partialKey, err := row.ReduceToIndexKeys(indexEd.Index(), dRow)
if err != nil {
return err
}
err = indexEd.DeleteRow(ctx, fullKey, partialKey)
if err != nil {
return err
}
}
te.tea.ed.AddEdit(key, nil)
te.tea.opCount++
return nil
}
// UpdateRow takes the current row and new rows, and updates it accordingly.
@@ -491,14 +568,12 @@ func (te *pkTableEditor) UpdateRow(ctx context.Context, dOldRow row.Row, dNewRow
te.flushMutex.RLock()
defer te.flushMutex.RUnlock()
dOldKey := dOldRow.NomsMapKey(te.tSch)
dOldKeyVal, err := dOldKey.Value(ctx)
dOldKeyVal, err := dOldRow.NomsMapKey(te.tSch).Value(ctx)
if err != nil {
return err
}
dNewKey := dNewRow.NomsMapKey(te.tSch)
dNewKeyVal, err := dNewKey.Value(ctx)
dNewKeyVal, err := dNewRow.NomsMapKey(te.tSch).Value(ctx)
if err != nil {
return err
}
@@ -531,7 +606,7 @@ func (te *pkTableEditor) UpdateRow(ctx context.Context, dOldRow row.Row, dNewRow
if kvp, pkExists, err := te.tea.maybeGet(ctx, newHash, dNewKeyVal); err != nil {
return err
} else if pkExists {
return te.pkErrForKVP(ctx, kvp, errFunc)
return te.keyErrForKVP(ctx, kvp, true, errFunc)
}
te.tea.addedKeys[newHash] = dNewKeyVal
te.tea.removedKeys[oldHash] = dOldKeyVal
@@ -543,6 +618,40 @@ func (te *pkTableEditor) UpdateRow(ctx context.Context, dOldRow row.Row, dNewRow
return err
}
for _, indexEd := range te.indexEds {
oldFullKey, oldPartialKey, err := row.ReduceToIndexKeys(indexEd.Index(), dOldRow)
if err != nil {
return err
}
err = indexEd.DeleteRow(ctx, oldFullKey, oldPartialKey)
if err != nil {
return err
}
newFullKey, newPartialKey, err := row.ReduceToIndexKeys(indexEd.Index(), dNewRow)
if err != nil {
return err
}
err = indexEd.InsertRow(ctx, newFullKey, newPartialKey)
if uke, ok := err.(*uniqueKeyErr); ok {
tableTupleHash, err := uke.TableTuple.Hash(uke.TableTuple.Format())
if err != nil {
return err
}
kvp, pkExists, err := te.tea.maybeGet(ctx, tableTupleHash, uke.TableTuple)
if err != nil {
return err
}
if !pkExists {
keyStr, _ := formatKey(ctx, uke.TableTuple)
return fmt.Errorf("UNIQUE constraint violation on index '%s', but could not find row with primary key: %s",
indexEd.Index().Name(), keyStr)
}
return te.keyErrForKVP(ctx, kvp, false, errFunc)
} else if err != nil {
return err
}
}
te.tea.affectedKeys[newHash] = &doltKVP{k: dNewKeyVal, v: val}
te.tea.ed.AddEdit(dNewKeyVal, val)
te.tea.opCount++
@@ -560,9 +669,15 @@ func (te *pkTableEditor) SetAutoIncrementValue(v types.Value) (err error) {
}
// Table returns a Table based on the edits given, if any. If Flush() was not called prior, it will be called here.
func (te *pkTableEditor) Table(_ context.Context) (*doltdb.Table, error) {
func (te *pkTableEditor) Table(ctx context.Context) (*doltdb.Table, error) {
te.flush()
err := te.aq.WaitForEmpty()
if err != nil {
return nil, err
}
te.flushMutex.Lock()
defer te.flushMutex.Unlock()
if te.hasAutoInc {
te.t, err = te.t.SetAutoIncrementValue(te.autoIncVal)
@@ -571,7 +686,37 @@ func (te *pkTableEditor) Table(_ context.Context) (*doltdb.Table, error) {
}
}
return te.t, err
tbl := te.t
idxMutex := &sync.Mutex{}
idxWg := &sync.WaitGroup{}
idxWg.Add(len(te.indexEds))
for i := 0; i < len(te.indexEds); i++ {
go func(i int) {
defer idxWg.Done()
indexMap, idxErr := te.indexEds[i].Map(ctx)
idxMutex.Lock()
defer idxMutex.Unlock()
if err != nil {
return
}
if idxErr != nil {
err = idxErr
return
}
tbl, idxErr = tbl.SetIndexRowData(ctx, te.indexEds[i].Index().Name(), indexMap)
if idxErr != nil {
err = idxErr
return
}
}(i)
}
idxWg.Wait()
if err != nil {
return nil, err
}
te.t = tbl
return te.t, nil
}
func (te *pkTableEditor) Schema() schema.Schema {
@@ -591,7 +736,10 @@ func (te *pkTableEditor) Format() *types.NomsBinFormat {
func (te *pkTableEditor) Close() error {
te.tea.ed.Close()
for _, indexEd := range te.indexEds {
indexEd.ed.Close()
err := indexEd.Close()
if err != nil {
return err
}
}
return nil
}
@@ -622,24 +770,6 @@ func (te *pkTableEditor) autoFlush() {
}
}
func (te *pkTableEditor) delete(key types.Tuple) error {
keyHash, err := key.Hash(te.nbf)
if err != nil {
return err
}
te.writeMutex.Lock()
defer te.writeMutex.Unlock()
delete(te.tea.addedKeys, keyHash)
te.tea.removedKeys[keyHash] = key
te.tea.affectedKeys[keyHash] = &doltKVP{k: key}
te.tea.ed.AddEdit(key, nil)
te.tea.opCount++
return nil
}
func (te *pkTableEditor) flushEditAccumulator(ctx context.Context, teaInterface interface{}) error {
// We don't call any locks at the function entrance since this is called from an ActionExecutor with a concurrency of 1
futureTea := teaInterface.(*tableEditAccumulator)
@@ -678,10 +808,6 @@ func (te *pkTableEditor) flushEditAccumulator(ctx context.Context, teaInterface
if err != nil {
return err
}
newTable, err = te.updateIndexes(ctx, tea, newTable, tea.rowData, updatedMap)
if err != nil {
return err
}
// No errors were encountered, so we set the bool to false. This should come after ALL calls that may error.
encounteredErr = false
@@ -729,84 +855,3 @@ func formatKey(ctx context.Context, key types.Value) (string, error) {
return fmt.Sprintf("[%s]", strings.Join(vals, ",")), nil
}
func (te *pkTableEditor) updateIndexes(ctx context.Context, tea *tableEditAccumulator, tbl *doltdb.Table, originalRowData types.Map, updated types.Map) (*doltdb.Table, error) {
// We don't call any locks here since this is called from an ActionExecutor with a concurrency of 1
if len(te.indexEds) == 0 {
return tbl, nil
}
indexActionQueue := async.NewActionExecutor(ctx, func(_ context.Context, keyInt interface{}) error {
key := keyInt.(types.Value)
var originalRow row.Row
var updatedRow row.Row
if val, ok, err := originalRowData.MaybeGet(ctx, key); err == nil && ok {
originalRow, err = row.FromNoms(te.tSch, key.(types.Tuple), val.(types.Tuple))
if err != nil {
return err
}
} else if err != nil {
return err
}
if val, ok, err := updated.MaybeGet(ctx, key); err == nil && ok {
updatedRow, err = row.FromNoms(te.tSch, key.(types.Tuple), val.(types.Tuple))
if err != nil {
return err
}
} else if err != nil {
return err
}
for _, indexEd := range te.indexEds {
var err error
var originalIndexRow row.Row
var updatedIndexRow row.Row
if originalRow != nil {
originalIndexRow, err = row.ReduceToIndex(indexEd.Index(), originalRow)
if err != nil {
return err
}
}
if updatedRow != nil {
updatedIndexRow, err = row.ReduceToIndex(indexEd.Index(), updatedRow)
if err != nil {
return err
}
}
err = indexEd.UpdateIndex(ctx, originalIndexRow, updatedIndexRow)
if err != nil {
return err
}
}
return nil
}, 4, 0)
for _, kvp := range tea.affectedKeys {
indexActionQueue.Execute(kvp.k)
}
err := indexActionQueue.WaitForEmpty()
if err != nil {
return nil, err
}
for _, indexEd := range te.indexEds {
if !indexEd.HasChanges() {
continue
}
indexMap, err := indexEd.Map(ctx)
if err != nil {
return nil, err
}
tbl, err = tbl.SetIndexRowData(ctx, indexEd.Index().Name(), indexMap)
if err != nil {
return nil, err
}
}
return tbl, nil
}

2
go/libraries/doltcore/table/editor/pk_table_editor_test.go
View File

@@ -343,7 +343,7 @@ func TestTableEditorDuplicateKeyHandling(t *testing.T) {
})
require.NoError(t, err)
err = tableEditor.InsertRow(context.Background(), dRow, nil)
require.True(t, errors.Is(err, ErrDuplicatePK))
require.True(t, errors.Is(err, ErrDuplicateKey))
}
_, err = tableEditor.Table(context.Background())

5
go/libraries/doltcore/table/pipeline/procfunc_help.go
View File

@@ -102,7 +102,10 @@ func ProcFuncForSinkFunc(sinkFunc SinkFunc) OutFunc {
err := sinkFunc(r.Row, r.Props)
if err != nil {
if table.IsBadRow(err) || sql.ErrPrimaryKeyViolation.Is(err) || errors.Is(err, editor.ErrDuplicatePK) {
if table.IsBadRow(err) ||
sql.ErrPrimaryKeyViolation.Is(err) ||
sql.ErrUniqueKeyViolation.Is(err) ||
errors.Is(err, editor.ErrDuplicateKey) {
badRowChan <- &TransformRowFailure{r.Row, "writer", err.Error()}
} else {
p.StopWithErr(err)

5
integration-tests/bats/1pk5col-ints.bats
View File

@@ -140,14 +140,11 @@ teardown() {
dolt sql -q "insert into test (pk,c1,c2,c3,c4,c5) values (0,6,6,6,6,6)"
run dolt sql -q "replace into test (pk,c1,c2,c3,c4,c5) values (0,7,7,7,7,7),(1,8,8,8,8,8)"
[ "$status" -eq 0 ]
# No skip, but this is a bug in the output. Query produces the right result, but counts it incorrectly
[[ "$output" =~ "Query OK, 4 rows affected" ]] || false
## No skip, but this should report 3 but is reporting 4 [[ "${lines[3]}" =~ "3" ]] || false
[[ "$output" =~ "Query OK, 3 rows affected" ]] || false
run dolt sql -q "select * from test"
[[ "$output" =~ "7" ]] || false
[[ "$output" =~ "8" ]] || false
[[ ! "$output" =~ "6" ]] || false
skip "replace into output is incorrect"
}
@test "1pk5col-ints: dolt sql insert and dolt sql select" {

15
integration-tests/bats/index.bats
View File

@@ -1979,17 +1979,16 @@ SQL
[[ "${#lines[@]}" = "2" ]] || false
run dolt sql -q "INSERT INTO onepk VALUES (6, 77, 56)"
[ "$status" -eq "1" ]
[[ "$output" =~ "UNIQUE" ]] || false
[[ "$output" =~ "unique" ]] || false
run dolt sql -q "INSERT INTO onepk VALUES (6, 78, 56)"
[ "$status" -eq "0" ]
run dolt sql -q "UPDATE onepk SET v1 = 22 WHERE pk1 = 1"
[ "$status" -eq "1" ]
[[ "$output" =~ "UNIQUE" ]] || false
[[ "$output" =~ "unique" ]] || false
run dolt sql -q "UPDATE onepk SET v1 = 23 WHERE pk1 = 1"
[ "$status" -eq "0" ]
run dolt sql -q "REPLACE INTO onepk VALUES (2, 88, 55)"
[ "$status" -eq "1" ]
[[ "$output" =~ "UNIQUE" ]] || false
[ "$status" -eq "0" ]
run dolt sql -q "REPLACE INTO onepk VALUES (2, 89, 55)"
[ "$status" -eq "0" ]
}
@@ -2103,7 +2102,7 @@ INSERT INTO onepk VALUES (6, 11, 55);
SQL
run dolt table import -u onepk `batshelper index_onepk.csv`
[ "$status" -eq "1" ]
[[ "$output" =~ "UNIQUE" ]] || false
[[ "$output" =~ "duplicate key" ]] || false
}
@test "index: UNIQUE dolt table import -r" {
@@ -2135,7 +2134,7 @@ SQL
dolt sql -q "DELETE FROM onepk"
run dolt table import -r onepk `batshelper index_onepk_non_unique.csv`
[ "$status" -eq "1" ]
[[ "$output" =~ "UNIQUE" ]] || false
[[ "$output" =~ "duplicate key" ]] || false
}
@test "index: Merge without conflicts" {
@@ -2315,7 +2314,7 @@ SQL
dolt checkout master
run dolt merge other
[ "$status" -eq "1" ]
[[ "$output" =~ "UNIQUE" ]] || false
[[ "$output" =~ "duplicate key" ]] || false
}
@test "index: Merge into branch with index from branch without index" {
@@ -2444,7 +2443,7 @@ SQL
dolt sql -q "INSERT INTO child_idx VALUES ('6', 5)"
run dolt sql -q "INSERT INTO child_unq VALUES ('6', 5)"
[ "$status" -eq "1" ]
[[ "$output" =~ "UNIQUE constraint violation" ]] || false
[[ "$output" =~ "unique" ]] || false
dolt sql -q "INSERT INTO child_non_unq VALUES ('6', 5)"
# INSERT against foreign key