Added statement boundaries

Author: Hydrocharged

enginetest/memory_engine_test.go

@@ -373,6 +373,12 @@ func TestTruncate(t *testing.T) {
 }
 
 func TestScripts(t *testing.T) {
+	//TODO: when foreign keys are implemented in the memory table, we can do the following test
+	for i := len(enginetest.ScriptTests) - 1; i >= 0; i-- {
+		if enginetest.ScriptTests[i].Name == "failed statements data validation for DELETE, REPLACE" {
+			enginetest.ScriptTests = append(enginetest.ScriptTests[:i], enginetest.ScriptTests[i+1:]...)
+		}
+	}
 	enginetest.TestScripts(t, enginetest.NewMemoryHarness("default", 1, testNumPartitions, true, mergableIndexDriver))
 }
 

enginetest/script_queries.go

@@ -55,6 +55,66 @@ type ScriptTestAssertion struct {
 // Unlike other engine tests, ScriptTests must be self-contained. No other tables are created outside the definition of
 // the tests.
 var ScriptTests = []ScriptTest{
+	{
+		Name: "failed statements data validation for INSERT, UPDATE",
+		SetUpScript: []string{
+			"CREATE TABLE test (pk BIGINT PRIMARY KEY, v1 BIGINT, INDEX (v1));",
+			"INSERT INTO test VALUES (1,1), (4,4), (5,5);",
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query:          "INSERT INTO test VALUES (2,2), (3,3), (1,1);",
+				ExpectedErrStr: "duplicate primary key given: [1]",
+			},
+			{
+				Query:    "SELECT * FROM test;",
+				Expected: []sql.Row{{1, 1}, {4, 4}, {5, 5}},
+			},
+			{
+				Query:          "UPDATE test SET pk = pk + 1;",
+				ExpectedErrStr: "duplicate primary key given: [5]",
+			},
+			{
+				Query:    "SELECT * FROM test;",
+				Expected: []sql.Row{{1, 1}, {4, 4}, {5, 5}},
+			},
+		},
+	},
+	{
+		Name: "failed statements data validation for DELETE, REPLACE",
+		SetUpScript: []string{
+			"CREATE TABLE test (pk BIGINT PRIMARY KEY, v1 BIGINT, INDEX (v1));",
+			"INSERT INTO test VALUES (1,1), (4,4), (5,5);",
+			"CREATE TABLE test2 (pk BIGINT PRIMARY KEY, CONSTRAINT fk_test FOREIGN KEY (pk) REFERENCES test (v1));",
+			"INSERT INTO test2 VALUES (4);",
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query:       "DELETE FROM test WHERE pk > 0;",
+				ExpectedErr: sql.ErrForeignKeyChildViolation,
+			},
+			{
+				Query:    "SELECT * FROM test;",
+				Expected: []sql.Row{{1, 1}, {4, 4}, {5, 5}},
+			},
+			{
+				Query:    "SELECT * FROM test2;",
+				Expected: []sql.Row{{4}},
+			},
+			{
+				Query:       "REPLACE INTO test VALUES (1,7), (4,8), (5,9);",
+				ExpectedErr: sql.ErrForeignKeyChildViolation,
+			},
+			{
+				Query:    "SELECT * FROM test;",
+				Expected: []sql.Row{{1, 1}, {4, 4}, {5, 5}},
+			},
+			{
+				Query:    "SELECT * FROM test2;",
+				Expected: []sql.Row{{4}},
+			},
+		},
+	},
 	{
 		Name: "delete with in clause",
 		SetUpScript: []string{

go.sum

@@ -14,8 +14,6 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81 h1:7/v8q9XGFa6q5Ap4Z/OhNkAMBaK5YeuEzwJt+NZdhiE=
 github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81/go.mod h1:siLfyv2c92W1eN/R4QqG/+RjjX5W2+gCTRjZxBjI3TY=
-github.com/dolthub/vitess v0.0.0-20210428165934-5801b1103b04 h1:yvfUh1EqwPu10H9VpaUxBEWzqKUmKxsW689ufJWNLsg=
-github.com/dolthub/vitess v0.0.0-20210428165934-5801b1103b04/go.mod h1:hUE8oSk2H5JZnvtlLBhJPYC8WZCA5AoSntdLTcBvdBM=
 github.com/dolthub/vitess v0.0.0-20210524220733-7b048c544267 h1:g3KWBmLtSWlEbwUF4NV4a4jzE5aec8n2ZHWwSDy9IGY=
 github.com/dolthub/vitess v0.0.0-20210524220733-7b048c544267/go.mod h1:hUE8oSk2H5JZnvtlLBhJPYC8WZCA5AoSntdLTcBvdBM=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=

memory/table.go

@@ -377,38 +377,65 @@ func EncodeIndexValue(value *IndexValue) ([]byte, error) {
 }
 
 type tableEditor struct {
-	table *Table
+	table             *Table
+	initialAutoIncVal interface{}
+	initialPartitions map[string][]sql.Row
+	initialInsert     int
 }
 
 var _ sql.RowReplacer = (*tableEditor)(nil)
 var _ sql.RowUpdater = (*tableEditor)(nil)
 var _ sql.RowInserter = (*tableEditor)(nil)
 var _ sql.RowDeleter = (*tableEditor)(nil)
 
-func (t tableEditor) Close(*sql.Context) error {
+func (t *tableEditor) Close(*sql.Context) error {
 	// TODO: it would be nice to apply all pending updates here at once, rather than directly in the Insert / Update
 	//  / Delete methods.
 	return nil
 }
 
+func (t *tableEditor) StatementBegin(ctx *sql.Context) {
+	t.initialInsert = t.table.insert
+	t.initialAutoIncVal = t.table.autoIncVal
+	t.initialPartitions = make(map[string][]sql.Row)
+	for partStr, rowSlice := range t.table.partitions {
+		newRowSlice := make([]sql.Row, len(rowSlice))
+		for i, row := range rowSlice {
+			newRowSlice[i] = row.Copy()
+		}
+		t.initialPartitions[partStr] = newRowSlice
+	}
+}
+
+func (t *tableEditor) DiscardChanges(ctx *sql.Context, errorEncountered error) error {
+	t.table.insert = t.initialInsert
+	t.table.autoIncVal = t.initialAutoIncVal
+	t.table.partitions = t.initialPartitions
+	return nil
+}
+
+func (t *tableEditor) StatementComplete(ctx *sql.Context) error {
+	return nil
+}
+
 func (t *Table) Inserter(*sql.Context) sql.RowInserter {
-	return &tableEditor{t}
+	return &tableEditor{t, nil, nil, 0}
 }
 
 func (t *Table) Updater(*sql.Context) sql.RowUpdater {
-	return &tableEditor{t}
+	return &tableEditor{t, nil, nil, 0}
 }
 
 func (t *Table) Replacer(*sql.Context) sql.RowReplacer {
-	return &tableEditor{t}
+	return &tableEditor{t, nil, nil, 0}
 }
 
 func (t *Table) Deleter(*sql.Context) sql.RowDeleter {
-	return &tableEditor{t}
+	return &tableEditor{t, nil, nil, 0}
 }
 
 func (t *Table) AutoIncrementSetter(*sql.Context) sql.AutoIncrementSetter {
-	return &tableEditor{t}
+	return &tableEditor{t, nil, nil, 0}
 }
 
 func (t *Table) Truncate(ctx *sql.Context) (int, error) {

sql/core.go

@@ -341,6 +341,20 @@ type CheckAlterableTable interface {
 	DropCheck(ctx *Context, chName string) error
 }
 
+// TableEditor is the base interface for sub interfaces that can update rows in a table during an INSERT, REPLACE,
+// UPDATE, or DELETE statement.
+type TableEditor interface {
+	// StatementBegin is called before the first operation of a statement. Integrators should mark the state of the data
+	// in some way that it may be returned to in the case of an error.
+	StatementBegin(ctx *Context)
+	// DiscardChanges is called if a statement encounters an error, and all current changes since the statement beginning
+	// should be discarded.
+	DiscardChanges(ctx *Context, errorEncountered error) error
+	// StatementComplete is called after the last operation of the statement, indicating that it has successfully completed.
+	// The mark set in StatementBegin may be removed, and a new one should be created on the next StatementBegin.
+	StatementComplete(ctx *Context) error
+}
+
 // InsertableTable is a table that can process insertion of new rows.
 type InsertableTable interface {
 	Table
@@ -351,6 +365,7 @@ type InsertableTable interface {
 
 // RowInserter is an insert cursor that can insert one or more values to a table.
 type RowInserter interface {
+	TableEditor
 	// Insert inserts the row given, returning an error if it cannot. Insert will be called once for each row to process
 	// for the insert operation, which may involve many rows. After all rows in an operation have been processed, Close
 	// is called.
@@ -369,6 +384,7 @@ type DeletableTable interface {
 
 // RowDeleter is a delete cursor that can delete one or more rows from a table.
 type RowDeleter interface {
+	TableEditor
 	// Delete deletes the given row. Returns ErrDeleteRowNotFound if the row was not found. Delete will be called once for
 	// each row to process for the delete operation, which may involve many rows. After all rows have been processed,
 	// Close is called.
@@ -419,6 +435,7 @@ type Closer interface {
 // TODO: We can't embed those interfaces because go 1.13 doesn't allow for overlapping interfaces (they both declare
 //  Close). Go 1.14 fixes this problem, but we aren't ready to drop support for 1.13 yet.
 type RowReplacer interface {
+	TableEditor
 	// Insert inserts the row given, returning an error if it cannot. Insert will be called once for each row to process
 	// for the replace operation, which may involve many rows. After all rows in an operation have been processed, Close
 	// is called.
@@ -449,6 +466,7 @@ type UpdatableTable interface {
 
 // RowUpdater is an update cursor that can update one or more rows in a table.
 type RowUpdater interface {
+	TableEditor
 	// Update the given row. Provides both the old and new rows.
 	Update(ctx *Context, old Row, new Row) error
 	// Close finalizes the update operation, persisting the result.

sql/plan/delete.go

@@ -138,8 +138,13 @@ func (d *deleteIter) Close(ctx *sql.Context) error {
 	return nil
 }
 
-func newDeleteIter(childIter sql.RowIter, deleter sql.RowDeleter, schema sql.Schema, ctx *sql.Context) *deleteIter {
-	return &deleteIter{deleter: deleter, childIter: childIter, schema: schema, ctx: ctx}
+func newDeleteIter(childIter sql.RowIter, deleter sql.RowDeleter, schema sql.Schema, ctx *sql.Context) sql.RowIter {
+	return NewTableEditorIter(ctx, deleter, &deleteIter{
+		deleter:   deleter,
+		childIter: childIter,
+		schema:    schema,
+		ctx:       ctx,
+	})
 }
 
 // WithChildren implements the Node interface.

sql/plan/insert.go

@@ -166,7 +166,7 @@ func newInsertIter(
 	checks sql.CheckConstraints,
 	row sql.Row,
 	ignore bool,
-) (*insertIter, error) {
+) (sql.RowIter, error) {
 	dstSchema := table.Schema()
 
 	insertable, err := GetInsertable(table)
@@ -194,8 +194,7 @@ func newInsertIter(
 	}
 
 	insertExpressions := getInsertExpressions(values)
-
-	return &insertIter{
+	insertIter := &insertIter{
 		schema:      dstSchema,
 		tableNode:   table,
 		inserter:    inserter,
@@ -207,7 +206,13 @@ func newInsertIter(
 		checks:      checks,
 		ctx:         ctx,
 		ignore:      ignore,
-	}, nil
+	}
+
+	if replacer != nil {
+		return NewTableEditorIter(ctx, replacer, insertIter), nil
+	} else {
+		return NewTableEditorIter(ctx, inserter, insertIter), nil
+	}
 }
 
 func getInsertExpressions(values sql.Node) []sql.Expression {

sql/plan/table_editor.go

@@ -0,0 +1,73 @@
+// Copyright 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.
+// 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,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package plan
+
+import (
+	"io"
+	"sync"
+
+	"github.com/dolthub/go-mysql-server/sql"
+)
+
+// tableEditorIter wraps the given iterator and calls the Begin and Complete functions on the given table.
+type tableEditorIter struct {
+	once             *sync.Once
+	onceCtx          *sql.Context
+	editor           sql.TableEditor
+	inner            sql.RowIter
+	errorEncountered error
+}
+
+var _ sql.RowIter = (*tableEditorIter)(nil)
+
+// NewTableEditorIter returns a new *tableEditorIter by wrapping the given iterator. If the
+// "statement_boundaries" session variable is set to false, then the original iterator is returned.
+func NewTableEditorIter(ctx *sql.Context, table sql.TableEditor, wrappedIter sql.RowIter) sql.RowIter {
+	return &tableEditorIter{
+		once:             &sync.Once{},
+		onceCtx:          ctx,
+		editor:           table,
+		inner:            wrappedIter,
+		errorEncountered: nil,
+	}
+}
+
+// Next implements the interface sql.RowIter.
+func (s *tableEditorIter) Next() (sql.Row, error) {
+	s.once.Do(func() {
+		s.editor.StatementBegin(s.onceCtx)
+	})
+	row, err := s.inner.Next()
+	if err != nil && err != io.EOF && !ErrInsertIgnore.Is(err) {
+		s.errorEncountered = err
+	}
+	return row, err
+}
+
+// Close implements the interface sql.RowIter.
+func (s *tableEditorIter) Close(ctx *sql.Context) error {
+	var err error
+	if s.errorEncountered != nil {
+		err = s.editor.DiscardChanges(ctx, s.errorEncountered)
+	} else {
+		err = s.editor.StatementComplete(ctx)
+	}
+	if err != nil {
+		_ = s.inner.Close(ctx)
+	} else {
+		err = s.inner.Close(ctx)
+	}
+	return err
+}

sql/plan/update.go

@@ -209,14 +209,14 @@ func newUpdateIter(
 	schema sql.Schema,
 	updater sql.RowUpdater,
 	checks sql.CheckConstraints,
-) *updateIter {
-	return &updateIter{
+) sql.RowIter {
+	return NewTableEditorIter(ctx, updater, &updateIter{
 		childIter: childIter,
 		updater:   updater,
 		schema:    schema,
 		checks:    checks,
 		ctx:       ctx,
-	}
+	})
 }
 
 // RowIter implements the Node interface.

sql/row.go

@@ -99,6 +99,7 @@ func RowIterToRows(ctx *Context, i RowIter) ([]Row, error) {
 		}
 
 		if err != nil {
+			_ = i.Close(ctx)
 			return nil, err
 		}