Repository: src-d/go-mysql-server
Pull request #825 (2019-09-24 20:29:17)

Implemented REPLACE
This PR is a follow-up to https://github.com/src-d/go-mysql-server/pull/822 and includes that commit's changes as well.

Looking over the [MySQL REPLACE documentation](https://dev.mysql.com/doc/refman/8.0/en/replace.html), I see that it mentions a REPLACE being either a DELETE then INSERT, or just an INSERT. I figured that there were two ways to implement this: as another interface with a unique function (`Replacer` with a `Replace` function for example), or by leveraging the already-included `Insert` and `Delete` functions from their respective interfaces. I decided to go with the latter option for the below reasons.

One point is in the reduction of code needed for someone wanting to implement the library. The more that the framework handles (without appreciable loss of freedom), the better. Then there is also the consideration of triggers. If DELETE and INSERT calls should make use of triggers (which are probably already built into the implementing `Delete` and `Insert` functions), then simply calling those _as normal_ will allow for correct behavior by default.

The biggest downside that I could see is that `Delete` is now forced to check if the given row exists, and return the appropriate error if not. For some implementations, this will not make an impact on the written code nor performance, but for others it definitely will. I felt that the trade off was worth it, but I'm interesting in hearing your opinions as well.

A third potential option is to allow for a custom `Replace` function if provided, otherwise defaulting to DELETE + INSERT from earlier. For example:
```go
type Replacer interface {
	Deleter
	Inserter
}

type CustomReplacer interface {
	Replace(*Context, Row) error
}
```
This would solve both of the problems as far as I can tell, but would not follow in the pattern established for every other keyword, and thus I decided against this.


Patch:
diff --git a/engine.go b/engine.go
index ed1fe1c..ae11ef7 100644
--- a/engine.go
+++ b/engine.go
@@ -120,7 +120,7 @@ func (e *Engine) Query(
 	case *plan.CreateIndex:
 		typ = sql.CreateIndexProcess
 		perm = auth.ReadPerm | auth.WritePerm
-	case *plan.InsertInto, *plan.DropIndex, *plan.UnlockTables, *plan.LockTables:
+	case *plan.InsertInto, *plan.DeleteFrom, *plan.DropIndex, *plan.UnlockTables, *plan.LockTables:
 		perm = auth.ReadPerm | auth.WritePerm
 	}
 
diff --git a/memory/table.go b/memory/table.go
index b6420af..a73eced 100644
--- a/memory/table.go
+++ b/memory/table.go
@@ -257,6 +257,39 @@ func (t *Table) Insert(ctx *sql.Context, row sql.Row) error {
 	return nil
 }
 
+// Delete the given row from the table.
+func (t *Table) Delete(ctx *sql.Context, row sql.Row) error {
+	if err := checkRow(t.schema, row); err != nil {
+		return err
+	}
+
+	matches := false
+	for partitionIndex, partition := range t.partitions {
+		for partitionRowIndex, partitionRow := range partition {
+			matches = true
+			for rIndex, val := range row {
+				if val != partitionRow[rIndex] {
+					matches = false
+					break
+				}
+			}
+			if matches {
+				t.partitions[partitionIndex] = append(partition[:partitionRowIndex], partition[partitionRowIndex+1:]...)
+				break
+			}
+		}
+		if matches {
+			break
+		}
+	}
+
+	if !matches {
+		return sql.ErrDeleteRowNotFound
+	}
+
+	return nil
+}
+
 func checkRow(schema sql.Schema, row sql.Row) error {
 	if len(row) != len(schema) {
 		return sql.ErrUnexpectedRowLength.New(len(schema), len(row))
diff --git a/sql/analyzer/pushdown.go b/sql/analyzer/pushdown.go
index 4c500ff..34312b1 100644
--- a/sql/analyzer/pushdown.go
+++ b/sql/analyzer/pushdown.go
@@ -20,7 +20,7 @@ func pushdown(ctx *sql.Context, a *Analyzer, n sql.Node) (sql.Node, error) {
 
 	// don't do pushdown on certain queries
 	switch n.(type) {
-	case *plan.InsertInto, *plan.CreateIndex:
+	case *plan.InsertInto, *plan.DeleteFrom, *plan.CreateIndex:
 		return n, nil
 	}
 
diff --git a/sql/core.go b/sql/core.go
index d676372..37f7655 100644
--- a/sql/core.go
+++ b/sql/core.go
@@ -29,6 +29,9 @@ var (
 	// ErrInvalidChildrenNumber is returned when the WithChildren method of a
 	// node or expression is called with an invalid number of arguments.
 	ErrInvalidChildrenNumber = errors.NewKind("%T: invalid children number, got %d, expected %d")
+
+	// ErrDeleteRowNotFound
+	ErrDeleteRowNotFound = errors.NewKind("row was not found when attempting to delete").New()
 )
 
 // Nameable is something that has a name.
@@ -202,6 +205,18 @@ type Inserter interface {
 	Insert(*Context, Row) error
 }
 
+// Deleter allow rows to be deleted from them.
+type Deleter interface {
+	// Delete the given row. Returns ErrDeleteRowNotFound if the row was not found.
+	Delete(*Context, Row) error
+}
+
+// Replacer allows rows to be replaced through a Delete (if applicable) then Insert.
+type Replacer interface {
+	Deleter
+	Inserter
+}
+
 // Database represents the database.
 type Database interface {
 	Nameable
diff --git a/sql/parse/parse.go b/sql/parse/parse.go
index 4f05916..72a125e 100644
--- a/sql/parse/parse.go
+++ b/sql/parse/parse.go
@@ -148,6 +148,8 @@ func convert(ctx *sql.Context, stmt sqlparser.Statement, query string) (sql.Node
 		return convertUse(n)
 	case *sqlparser.Rollback:
 		return plan.NewRollback(), nil
+	case *sqlparser.Delete:
+		return convertDelete(ctx, n)
 	}
 }
 
@@ -371,6 +373,8 @@ func convertInsert(ctx *sql.Context, i *sqlparser.Insert) (sql.Node, error) {
 		return nil, ErrUnsupportedSyntax.New(i)
 	}
 
+	isReplace := i.Action == sqlparser.ReplaceStr
+
 	src, err := insertRowsToNode(ctx, i.Rows)
 	if err != nil {
 		return nil, err
@@ -379,10 +383,49 @@ func convertInsert(ctx *sql.Context, i *sqlparser.Insert) (sql.Node, error) {
 	return plan.NewInsertInto(
 		plan.NewUnresolvedTable(i.Table.Name.String(), i.Table.Qualifier.String()),
 		src,
+		isReplace,
 		columnsToStrings(i.Columns),
 	), nil
 }
 
+func convertDelete(ctx *sql.Context, d *sqlparser.Delete) (sql.Node, error) {
+	node, err := tableExprsToTable(ctx, d.TableExprs)
+	if err != nil {
+		return nil, err
+	}
+
+	if d.Where != nil {
+		node, err = whereToFilter(d.Where, node)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if len(d.OrderBy) != 0 {
+		node, err = orderByToSort(d.OrderBy, node)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// Limit must wrap offset, and not vice-versa, so that skipped rows don't count toward the returned row count.
+	if d.Limit != nil && d.Limit.Offset != nil {
+		node, err = offsetToOffset(ctx, d.Limit.Offset, node)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if d.Limit != nil {
+		node, err = limitToLimit(ctx, d.Limit.Rowcount, node)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return plan.NewDeleteFrom(node), nil
+}
+
 func columnDefinitionToSchema(colDef []*sqlparser.ColumnDefinition) (sql.Schema, error) {
 	var schema sql.Schema
 	for _, cd := range colDef {
diff --git a/sql/plan/delete.go b/sql/plan/delete.go
new file mode 100644
index 0000000..408d832
--- /dev/null
+++ b/sql/plan/delete.go
@@ -0,0 +1,125 @@
+package plan
+
+import (
+	"github.com/src-d/go-mysql-server/sql"
+	"gopkg.in/src-d/go-errors.v1"
+	"io"
+)
+
+var ErrDeleteFromNotSupported = errors.NewKind("table doesn't support DELETE FROM")
+
+// DeleteFrom is a node describing a deletion from some table.
+type DeleteFrom struct {
+	sql.Node
+}
+
+// NewDeleteFrom creates a DeleteFrom node.
+func NewDeleteFrom(n sql.Node) *DeleteFrom {
+	return &DeleteFrom{n}
+}
+
+// Schema implements the Node interface.
+func (p *DeleteFrom) Schema() sql.Schema {
+	return sql.Schema{{
+		Name:     "updated",
+		Type:     sql.Int64,
+		Default:  int64(0),
+		Nullable: false,
+	}}
+}
+
+// Resolved implements the Resolvable interface.
+func (p *DeleteFrom) Resolved() bool {
+	return p.Node.Resolved()
+}
+
+func (p *DeleteFrom) Children() []sql.Node {
+	return []sql.Node{p.Node}
+}
+
+func getDeletable(node sql.Node) (sql.Deleter, error) {
+	switch node := node.(type) {
+	case sql.Deleter:
+		return node, nil
+	case *ResolvedTable:
+		return getDeletableTable(node.Table)
+	}
+	for _, child := range node.Children() {
+		deleter, _ := getDeletable(child)
+		if deleter != nil {
+			return deleter, nil
+		}
+	}
+	return nil, ErrDeleteFromNotSupported.New()
+}
+
+func getDeletableTable(t sql.Table) (sql.Deleter, error) {
+	switch t := t.(type) {
+	case sql.Deleter:
+		return t, nil
+	case sql.TableWrapper:
+		return getDeletableTable(t.Underlying())
+	default:
+		return nil, ErrDeleteFromNotSupported.New()
+	}
+}
+
+// Execute inserts the rows in the database.
+func (p *DeleteFrom) Execute(ctx *sql.Context) (int, error) {
+	deletable, err := getDeletable(p.Node)
+	if err != nil {
+		return 0, err
+	}
+
+	iter, err := p.Node.RowIter(ctx)
+	if err != nil {
+		return 0, err
+	}
+
+	i := 0
+	for {
+		row, err := iter.Next()
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			_ = iter.Close()
+			return i, err
+		}
+
+		if err := deletable.Delete(ctx, row); err != nil {
+			_ = iter.Close()
+			return i, err
+		}
+
+		i++
+	}
+
+	return i, nil
+}
+
+// RowIter implements the Node interface.
+func (p *DeleteFrom) RowIter(ctx *sql.Context) (sql.RowIter, error) {
+	n, err := p.Execute(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return sql.RowsToRowIter(sql.NewRow(int64(n))), nil
+}
+
+// WithChildren implements the Node interface.
+func (p *DeleteFrom) WithChildren(children ...sql.Node) (sql.Node, error) {
+	if len(children) != 1 {
+		return nil, sql.ErrInvalidChildrenNumber.New(p, len(children), 1)
+	}
+	return NewDeleteFrom(children[0]), nil
+}
+
+func (p DeleteFrom) String() string {
+	pr := sql.NewTreePrinter()
+	_ = pr.WriteNode("Delete")
+	_ = pr.WriteChildren(p.Node.String())
+	return pr.String()
+}
diff --git a/sql/plan/insert.go b/sql/plan/insert.go
index d795cf2..807191f 100644
--- a/sql/plan/insert.go
+++ b/sql/plan/insert.go
@@ -1,33 +1,39 @@
 package plan
 
 import (
-	"io"
-	"strings"
-
 	"github.com/src-d/go-mysql-server/sql"
 	"github.com/src-d/go-mysql-server/sql/expression"
 	"gopkg.in/src-d/go-errors.v1"
+	"io"
+	"strings"
 )
 
 // ErrInsertIntoNotSupported is thrown when a table doesn't support inserts
 var ErrInsertIntoNotSupported = errors.NewKind("table doesn't support INSERT INTO")
+var ErrReplaceIntoNotSupported = errors.NewKind("table doesn't support REPLACE INTO")
 var ErrInsertIntoMismatchValueCount =
 	errors.NewKind("number of values does not match number of columns provided")
 var ErrInsertIntoUnsupportedValues = errors.NewKind("%T is unsupported for inserts")
 var ErrInsertIntoDuplicateColumn = errors.NewKind("duplicate column name %v")
 var ErrInsertIntoNonexistentColumn = errors.NewKind("invalid column name %v")
+var ErrInsertIntoNonNullableDefaultNullColumn =
+	errors.NewKind("column name '%v' is non-nullable but attempted to set default value of null")
+var ErrInsertIntoNonNullableProvidedNull =
+	errors.NewKind("column name '%v' is non-nullable but attempted to set a value of null")
 
 // InsertInto is a node describing the insertion into some table.
 type InsertInto struct {
 	BinaryNode
-	Columns []string
+	Columns   []string
+	IsReplace bool
 }
 
 // NewInsertInto creates an InsertInto node.
-func NewInsertInto(dst, src sql.Node, cols []string) *InsertInto {
+func NewInsertInto(dst, src sql.Node, isReplace bool, cols []string) *InsertInto {
 	return &InsertInto{
 		BinaryNode: BinaryNode{Left: dst, Right: src},
 		Columns:    cols,
+		IsReplace:  isReplace,
 	}
 }
 
@@ -70,6 +76,15 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 		return 0, err
 	}
 
+	var replaceable sql.Replacer
+	if p.IsReplace {
+		var ok bool
+		replaceable, ok = insertable.(sql.Replacer)
+		if !ok {
+			return 0, ErrReplaceIntoNotSupported.New()
+		}
+	}
+
 	dstSchema := p.Left.Schema()
 	projExprs := make([]sql.Expression, len(dstSchema))
 
@@ -102,7 +117,10 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 		}
 
 		if !found {
-			projExprs[i] = expression.NewLiteral(nil, f.Type)
+			if !f.Nullable && f.Default == nil {
+				return 0, ErrInsertIntoNonNullableDefaultNullColumn.New(f.Name)
+			}
+			projExprs[i] = expression.NewLiteral(f.Default, f.Type)
 		}
 	}
 
@@ -119,17 +137,37 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 		if err == io.EOF {
 			break
 		}
-
 		if err != nil {
 			_ = iter.Close()
 			return i, err
 		}
 
-		if err := insertable.Insert(ctx, row); err != nil {
+		err = p.validateNullability(ctx, dstSchema, row)
+		if err != nil {
 			_ = iter.Close()
 			return i, err
 		}
 
+		if replaceable != nil {
+			if err = replaceable.Delete(ctx, row); err != nil {
+				if err != sql.ErrDeleteRowNotFound {
+					_ = iter.Close()
+					return i, err
+				}
+			} else {
+				i++
+			}
+
+			if err = replaceable.Insert(ctx, row); err != nil {
+				_ = iter.Close()
+				return i, err
+			}
+		} else {
+			if err := insertable.Insert(ctx, row); err != nil {
+				_ = iter.Close()
+				return i, err
+			}
+		}
 		i++
 	}
 
@@ -152,12 +190,16 @@ func (p *InsertInto) WithChildren(children ...sql.Node) (sql.Node, error) {
 		return nil, sql.ErrInvalidChildrenNumber.New(p, len(children), 2)
 	}
 
-	return NewInsertInto(children[0], children[1], p.Columns), nil
+	return NewInsertInto(children[0], children[1], p.IsReplace, p.Columns), nil
 }
 
 func (p InsertInto) String() string {
 	pr := sql.NewTreePrinter()
-	_ = pr.WriteNode("Insert(%s)", strings.Join(p.Columns, ", "))
+	if p.IsReplace {
+		_ = pr.WriteNode("Replace(%s)", strings.Join(p.Columns, ", "))
+	} else {
+		_ = pr.WriteNode("Insert(%s)", strings.Join(p.Columns, ", "))
+	}
 	_ = pr.WriteChildren(p.Left.String(), p.Right.String())
 	return pr.String()
 }
@@ -194,3 +236,12 @@ func (p *InsertInto) validateColumns(ctx *sql.Context, dstSchema sql.Schema) err
 	}
 	return nil
 }
+
+func (p *InsertInto) validateNullability(ctx *sql.Context, dstSchema sql.Schema, row sql.Row) error {
+	for i, col := range dstSchema {
+		if !col.Nullable && row[i] == nil {
+			return ErrInsertIntoNonNullableProvidedNull.New(col.Name)
+		}
+	}
+	return nil
+}


Test patch:
diff --git a/engine_test.go b/engine_test.go
index 0b447d5..5b5949f 100644
--- a/engine_test.go
+++ b/engine_test.go
@@ -1824,11 +1824,6 @@ func TestOrderByColumns(t *testing.T) {
 }
 
 func TestInsertInto(t *testing.T) {
-	timeParse := func(layout string, value string) time.Time {
-		t, _ := time.Parse(layout, value)
-		return t
-	}
-
 	var insertions = []struct {
 		insertQuery    string
 		expectedInsert []sql.Row
@@ -1841,12 +1836,24 @@ func TestInsertInto(t *testing.T) {
 			"SELECT i FROM mytable WHERE s = 'x';",
 			[]sql.Row{{int64(999)}},
 		},
+		{
+			"INSERT INTO mytable SET s = 'x', i = 999;",
+			[]sql.Row{{int64(1)}},
+			"SELECT i FROM mytable WHERE s = 'x';",
+			[]sql.Row{{int64(999)}},
+		},
 		{
 			"INSERT INTO mytable VALUES (999, 'x');",
 			[]sql.Row{{int64(1)}},
 			"SELECT i FROM mytable WHERE s = 'x';",
 			[]sql.Row{{int64(999)}},
 		},
+		{
+			"INSERT INTO mytable SET i = 999, s = 'x';",
+			[]sql.Row{{int64(1)}},
+			"SELECT i FROM mytable WHERE s = 'x';",
+			[]sql.Row{{int64(999)}},
+		},
 		{
 			`INSERT INTO typestable VALUES (
 			999, 127, 32767, 2147483647, 9223372036854775807,
@@ -1865,6 +1872,24 @@ func TestInsertInto(t *testing.T) {
 				"random text", true, `{"key":"value"}`, "blobdata",
 			}},
 		},
+		{
+			`INSERT INTO typestable SET
+			id = 999, i8 = 127, i16 = 32767, i32 = 2147483647, i64 = 9223372036854775807,
+			u8 = 255, u16 = 65535, u32 = 4294967295, u64 = 18446744073709551615,
+			f32 = 3.40282346638528859811704183484516925440e+38, f64 = 1.797693134862315708145274237317043567981e+308,
+			ti = '2132-04-05 12:51:36', da = '2231-11-07',
+			te = 'random text', bo = true, js = '{"key":"value"}', bl = 'blobdata'
+			;`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{
+				int64(999), int64(math.MaxInt8), int64(math.MaxInt16), int64(math.MaxInt32), int64(math.MaxInt64),
+				int64(math.MaxUint8), int64(math.MaxUint16), int64(math.MaxUint32), uint64(math.MaxUint64),
+				float64(math.MaxFloat32), float64(math.MaxFloat64),
+				timeParse(sql.TimestampLayout, "2132-04-05 12:51:36"), timeParse(sql.DateLayout, "2231-11-07"),
+				"random text", true, `{"key":"value"}`, "blobdata",
+			}},
+		},
 		{
 			`INSERT INTO typestable VALUES (
 			999, -128, -32768, -2147483648, -9223372036854775808,
@@ -1883,6 +1908,24 @@ func TestInsertInto(t *testing.T) {
 				"", false, ``, "",
 			}},
 		},
+		{
+			`INSERT INTO typestable SET
+			id = 999, i8 = -128, i16 = -32768, i32 = -2147483648, i64 = -9223372036854775808,
+			u8 = 0, u16 = 0, u32 = 0, u64 = 0,
+			f32 = 1.401298464324817070923729583289916131280e-45, f64 = 4.940656458412465441765687928682213723651e-324,
+			ti = '0010-04-05 12:51:36', da = '0101-11-07',
+			te = '', bo = false, js = '', bl = ''
+			;`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{
+				int64(999), int64(-math.MaxInt8-1), int64(-math.MaxInt16-1), int64(-math.MaxInt32-1), int64(-math.MaxInt64-1),
+				int64(0), int64(0), int64(0), int64(0),
+				float64(math.SmallestNonzeroFloat32), float64(math.SmallestNonzeroFloat64),
+				timeParse(sql.TimestampLayout, "0010-04-05 12:51:36"), timeParse(sql.DateLayout, "0101-11-07"),
+				"", false, ``, "",
+			}},
+		},
 		{
 			`INSERT INTO typestable VALUES (999, null, null, null, null, null, null, null, null,
 			null, null, null, null, null, null, null, null);`,
@@ -1890,6 +1933,13 @@ func TestInsertInto(t *testing.T) {
 			"SELECT * FROM typestable WHERE id = 999;",
 			[]sql.Row{{int64(999), nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil}},
 		},
+		{
+			`INSERT INTO typestable SET id=999, i8=null, i16=null, i32=null, i64=null, u8=null, u16=null, u32=null, u64=null,
+			f32=null, f64=null, ti=null, da=null, te=null, bo=null, js=null, bl=null;`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{int64(999), nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil}},
+		},
 	}
 
 	for _, insertion := range insertions {
@@ -1926,13 +1976,229 @@ func TestInsertIntoErrors(t *testing.T) {
 			"INSERT INTO mytable VALUES (999, 'x', 'y');",
 		},
 		{
-			"non-existent column",
+			"non-existent column values",
 			"INSERT INTO mytable (i, s, z) VALUES (999, 'x', 999);",
 		},
+		{
+			"non-existent column set",
+			"INSERT INTO mytable SET i = 999, s = 'x', z = 999;",
+		},
 		{
 			"duplicate column",
 			"INSERT INTO mytable (i, s, s) VALUES (999, 'x', 'x');",
 		},
+		{
+			"duplicate column set",
+			"INSERT INTO mytable SET i = 999, s = 'y', s = 'y';",
+		},
+		{
+			"null given to non-nullable",
+			"INSERT INTO mytable (i, s) VALUES (null, 'y');",
+		},
+	}
+
+	for _, expectedFailure := range expectedFailures {
+		t.Run(expectedFailure.name, func(t *testing.T) {
+			_, _, err := newEngine(t).Query(newCtx(), expectedFailure.query)
+			require.Error(t, err)
+		})
+	}
+}
+
+func TestReplaceInto(t *testing.T) {
+	var insertions = []struct {
+		replaceQuery    string
+		expectedReplace []sql.Row
+		selectQuery     string
+		expectedSelect  []sql.Row
+	}{
+		{
+			"REPLACE INTO mytable VALUES (1, 'first row');",
+			[]sql.Row{{int64(2)}},
+			"SELECT s FROM mytable WHERE i = 1;",
+			[]sql.Row{{"first row"}},
+		},
+		{
+			"REPLACE INTO mytable SET i = 1, s = 'first row';",
+			[]sql.Row{{int64(2)}},
+			"SELECT s FROM mytable WHERE i = 1;",
+			[]sql.Row{{"first row"}},
+		},
+		{
+			"REPLACE INTO mytable VALUES (1, 'new row same i');",
+			[]sql.Row{{int64(1)}},
+			"SELECT s FROM mytable WHERE i = 1;",
+			[]sql.Row{{"first row"}, {"new row same i"}},
+		},
+		{
+			"REPLACE INTO mytable (s, i) VALUES ('x', 999);",
+			[]sql.Row{{int64(1)}},
+			"SELECT i FROM mytable WHERE s = 'x';",
+			[]sql.Row{{int64(999)}},
+		},
+		{
+			"REPLACE INTO mytable SET s = 'x', i = 999;",
+			[]sql.Row{{int64(1)}},
+			"SELECT i FROM mytable WHERE s = 'x';",
+			[]sql.Row{{int64(999)}},
+		},
+		{
+			"REPLACE INTO mytable VALUES (999, 'x');",
+			[]sql.Row{{int64(1)}},
+			"SELECT i FROM mytable WHERE s = 'x';",
+			[]sql.Row{{int64(999)}},
+		},
+		{
+			"REPLACE INTO mytable SET i = 999, s = 'x';",
+			[]sql.Row{{int64(1)}},
+			"SELECT i FROM mytable WHERE s = 'x';",
+			[]sql.Row{{int64(999)}},
+		},
+		{
+			`REPLACE INTO typestable VALUES (
+			999, 127, 32767, 2147483647, 9223372036854775807,
+			255, 65535, 4294967295, 18446744073709551615,
+			3.40282346638528859811704183484516925440e+38, 1.797693134862315708145274237317043567981e+308,
+			'2132-04-05 12:51:36', '2231-11-07',
+			'random text', true, '{"key":"value"}', 'blobdata'
+			);`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{
+				int64(999), int64(math.MaxInt8), int64(math.MaxInt16), int64(math.MaxInt32), int64(math.MaxInt64),
+				int64(math.MaxUint8), int64(math.MaxUint16), int64(math.MaxUint32), uint64(math.MaxUint64),
+				float64(math.MaxFloat32), float64(math.MaxFloat64),
+				timeParse(sql.TimestampLayout, "2132-04-05 12:51:36"), timeParse(sql.DateLayout, "2231-11-07"),
+				"random text", true, `{"key":"value"}`, "blobdata",
+			}},
+		},
+		{
+			`REPLACE INTO typestable SET
+			id = 999, i8 = 127, i16 = 32767, i32 = 2147483647, i64 = 9223372036854775807,
+			u8 = 255, u16 = 65535, u32 = 4294967295, u64 = 18446744073709551615,
+			f32 = 3.40282346638528859811704183484516925440e+38, f64 = 1.797693134862315708145274237317043567981e+308,
+			ti = '2132-04-05 12:51:36', da = '2231-11-07',
+			te = 'random text', bo = true, js = '{"key":"value"}', bl = 'blobdata'
+			;`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{
+				int64(999), int64(math.MaxInt8), int64(math.MaxInt16), int64(math.MaxInt32), int64(math.MaxInt64),
+				int64(math.MaxUint8), int64(math.MaxUint16), int64(math.MaxUint32), uint64(math.MaxUint64),
+				float64(math.MaxFloat32), float64(math.MaxFloat64),
+				timeParse(sql.TimestampLayout, "2132-04-05 12:51:36"), timeParse(sql.DateLayout, "2231-11-07"),
+				"random text", true, `{"key":"value"}`, "blobdata",
+			}},
+		},
+		{
+			`REPLACE INTO typestable VALUES (
+			999, -128, -32768, -2147483648, -9223372036854775808,
+			0, 0, 0, 0,
+			1.401298464324817070923729583289916131280e-45, 4.940656458412465441765687928682213723651e-324,
+			'0010-04-05 12:51:36', '0101-11-07',
+			'', false, '', ''
+			);`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{
+				int64(999), int64(-math.MaxInt8-1), int64(-math.MaxInt16-1), int64(-math.MaxInt32-1), int64(-math.MaxInt64-1),
+				int64(0), int64(0), int64(0), int64(0),
+				float64(math.SmallestNonzeroFloat32), float64(math.SmallestNonzeroFloat64),
+				timeParse(sql.TimestampLayout, "0010-04-05 12:51:36"), timeParse(sql.DateLayout, "0101-11-07"),
+				"", false, ``, "",
+			}},
+		},
+		{
+			`REPLACE INTO typestable SET
+			id = 999, i8 = -128, i16 = -32768, i32 = -2147483648, i64 = -9223372036854775808,
+			u8 = 0, u16 = 0, u32 = 0, u64 = 0,
+			f32 = 1.401298464324817070923729583289916131280e-45, f64 = 4.940656458412465441765687928682213723651e-324,
+			ti = '0010-04-05 12:51:36', da = '0101-11-07',
+			te = '', bo = false, js = '', bl = ''
+			;`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{
+				int64(999), int64(-math.MaxInt8-1), int64(-math.MaxInt16-1), int64(-math.MaxInt32-1), int64(-math.MaxInt64-1),
+				int64(0), int64(0), int64(0), int64(0),
+				float64(math.SmallestNonzeroFloat32), float64(math.SmallestNonzeroFloat64),
+				timeParse(sql.TimestampLayout, "0010-04-05 12:51:36"), timeParse(sql.DateLayout, "0101-11-07"),
+				"", false, ``, "",
+			}},
+		},
+		{
+			`REPLACE INTO typestable VALUES (999, null, null, null, null, null, null, null, null,
+			null, null, null, null, null, null, null, null);`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{int64(999), nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil}},
+		},
+		{
+			`REPLACE INTO typestable SET id=999, i8=null, i16=null, i32=null, i64=null, u8=null, u16=null, u32=null, u64=null,
+			f32=null, f64=null, ti=null, da=null, te=null, bo=null, js=null, bl=null;`,
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM typestable WHERE id = 999;",
+			[]sql.Row{{int64(999), nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil}},
+		},
+	}
+
+	for _, insertion := range insertions {
+		e := newEngine(t)
+		ctx := newCtx()
+		testQueryWithContext(ctx, t, e, insertion.replaceQuery, insertion.expectedReplace)
+		testQueryWithContext(ctx, t, e, insertion.selectQuery, insertion.expectedSelect)
+	}
+}
+
+func TestReplaceIntoErrors(t *testing.T) {
+	var expectedFailures = []struct {
+		name  string
+		query string
+	}{
+		{
+			"too few values",
+			"REPLACE INTO mytable (s, i) VALUES ('x');",
+		},
+		{
+			"too many values one column",
+			"REPLACE INTO mytable (s) VALUES ('x', 999);",
+		},
+		{
+			"too many values two columns",
+			"REPLACE INTO mytable (i, s) VALUES (999, 'x', 'y');",
+		},
+		{
+			"too few values no columns specified",
+			"REPLACE INTO mytable VALUES (999);",
+		},
+		{
+			"too many values no columns specified",
+			"REPLACE INTO mytable VALUES (999, 'x', 'y');",
+		},
+		{
+			"non-existent column values",
+			"REPLACE INTO mytable (i, s, z) VALUES (999, 'x', 999);",
+		},
+		{
+			"non-existent column set",
+			"REPLACE INTO mytable SET i = 999, s = 'x', z = 999;",
+		},
+		{
+			"duplicate column values",
+			"REPLACE INTO mytable (i, s, s) VALUES (999, 'x', 'x');",
+		},
+		{
+			"duplicate column set",
+			"REPLACE INTO mytable SET i = 999, s = 'y', s = 'y';",
+		},
+		{
+			"null given to non-nullable values",
+			"INSERT INTO mytable (i, s) VALUES (null, 'y');",
+		},
+		{
+			"null given to non-nullable set",
+			"INSERT INTO mytable SET i = null, s = 'y';",
+		},
 	}
 
 	for _, expectedFailure := range expectedFailures {
@@ -2761,6 +3027,142 @@ func TestDescribeNoPruneColumns(t *testing.T) {
 	require.Len(p.Schema(), 3)
 }
 
+func TestDeleteFrom(t *testing.T) {
+	var deletions = []struct {
+		deleteQuery    string
+		expectedDelete []sql.Row
+		selectQuery    string
+		expectedSelect []sql.Row
+	}{
+		{
+			"DELETE FROM mytable;",
+			[]sql.Row{{int64(3)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{},
+		},
+		{
+			"DELETE FROM mytable WHERE i = 2;",
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(1), "first row"}, {int64(3), "third row"}},
+		},
+		{
+			"DELETE FROM mytable WHERE i < 3;",
+			[]sql.Row{{int64(2)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(3), "third row"}},
+		},
+		{
+			"DELETE FROM mytable WHERE i > 1;",
+			[]sql.Row{{int64(2)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(1), "first row"}},
+		},
+		{
+			"DELETE FROM mytable WHERE i <= 2;",
+			[]sql.Row{{int64(2)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(3), "third row"}},
+		},
+		{
+			"DELETE FROM mytable WHERE i >= 2;",
+			[]sql.Row{{int64(2)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(1), "first row"}},
+		},
+		{
+			"DELETE FROM mytable WHERE s = 'first row';",
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(2), "second row"}, {int64(3), "third row"}},
+		},
+		{
+			"DELETE FROM mytable WHERE s <> 'dne';",
+			[]sql.Row{{int64(3)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{},
+		},
+		{
+			"DELETE FROM mytable WHERE s LIKE '%row';",
+			[]sql.Row{{int64(3)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{},
+		},
+		{
+			"DELETE FROM mytable WHERE s = 'dne';",
+			[]sql.Row{{int64(0)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(1), "first row"}, {int64(2), "second row"}, {int64(3), "third row"}},
+		},
+		{
+			"DELETE FROM mytable WHERE i = 'invalid';",
+			[]sql.Row{{int64(0)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(1), "first row"}, {int64(2), "second row"}, {int64(3), "third row"}},
+		},
+		{
+			"DELETE FROM mytable ORDER BY i ASC LIMIT 2;",
+			[]sql.Row{{int64(2)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(3), "third row"}},
+		},
+		{
+			"DELETE FROM mytable ORDER BY i DESC LIMIT 1;",
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(1), "first row"}, {int64(2), "second row"}},
+		},
+		{
+			"DELETE FROM mytable ORDER BY i DESC LIMIT 1 OFFSET 1;",
+			[]sql.Row{{int64(1)}},
+			"SELECT * FROM mytable;",
+			[]sql.Row{{int64(1), "first row"}, {int64(3), "third row"}},
+		},
+	}
+
+	for _, deletion := range deletions {
+		e := newEngine(t)
+		ctx := newCtx()
+		testQueryWithContext(ctx, t, e, deletion.deleteQuery, deletion.expectedDelete)
+		testQueryWithContext(ctx, t, e, deletion.selectQuery, deletion.expectedSelect)
+	}
+}
+
+func TestDeleteFromErrors(t *testing.T) {
+	var expectedFailures = []struct {
+		name  string
+		query string
+	}{
+		{
+			"invalid table",
+			"DELETE FROM invalidtable WHERE x < 1;",
+		},
+		{
+			"invalid column",
+			"DELETE FROM mytable WHERE z = 'dne';",
+		},
+		{
+			"negative limit",
+			"DELETE FROM mytable LIMIT -1;",
+		},
+		{
+			"negative offset",
+			"DELETE FROM mytable LIMIT 1 OFFSET -1;",
+		},
+		{
+			"missing keyword from",
+			"DELETE mytable WHERE id = 1;",
+		},
+	}
+
+	for _, expectedFailure := range expectedFailures {
+		t.Run(expectedFailure.name, func(t *testing.T) {
+			_, _, err := newEngine(t).Query(newCtx(), expectedFailure.query)
+			require.Error(t, err)
+		})
+	}
+}
+
 var generatorQueries = []struct {
 	query    string
 	expected []sql.Row
@@ -2861,6 +3263,14 @@ func newLockableTable(t sql.Table) *lockableTable {
 	return &lockableTable{Table: t}
 }
 
+func timeParse(layout string, value string) time.Time {
+	t, err := time.Parse(layout, value)
+	if err != nil {
+		panic(err)
+	}
+	return t
+}
+
 var _ sql.Lockable = (*lockableTable)(nil)
 
 func (l *lockableTable) Lock(ctx *sql.Context, write bool) error {
diff --git a/sql/parse/parse_test.go b/sql/parse/parse_test.go
index 42f4123..01e2a4e 100644
--- a/sql/parse/parse_test.go
+++ b/sql/parse/parse_test.go
@@ -252,6 +252,16 @@ var fixtures = map[string]sql.Node{
 			expression.NewLiteral("a", sql.Text),
 			expression.NewLiteral(int64(1), sql.Int64),
 		}}),
+		false,
+		[]string{"col1", "col2"},
+	),
+	`REPLACE INTO t1 (col1, col2) VALUES ('a', 1)`: plan.NewInsertInto(
+		plan.NewUnresolvedTable("t1", ""),
+		plan.NewValues([][]sql.Expression{{
+			expression.NewLiteral("a", sql.Text),
+			expression.NewLiteral(int64(1), sql.Int64),
+		}}),
+		true,
 		[]string{"col1", "col2"},
 	),
 	`SHOW TABLES`:               plan.NewShowTables(sql.UnresolvedDatabase(""), false),