Merge pull request #28 from liquidata-inc/zachmu/insert-update-delete-perf

New interfaces for insert, update, delete, replace. The new interfaces allow batching of multiple rows for efficiency and allow implementors to implement atomic operations.

Author: zachmu

engine_test.go

@@ -3812,12 +3812,15 @@ func TestGenerators(t *testing.T) {
 	}
 }
 
-func insertRows(t *testing.T, table sql.Inserter, rows ...sql.Row) {
+func insertRows(t *testing.T, table sql.InsertableTable, rows ...sql.Row) {
 	t.Helper()
 
+	ctx := newCtx()
+	inserter := table.Inserter(ctx)
 	for _, r := range rows {
-		require.NoError(t, table.Insert(newCtx(), r))
+		require.NoError(t, inserter.Insert(ctx, r))
 	}
+	require.NoError(t, inserter.Close(ctx))
 }
 
 var pid uint64

memory/table.go

@@ -28,7 +28,10 @@ type Table struct {
 }
 
 var _ sql.Table = (*Table)(nil)
-var _ sql.Inserter = (*Table)(nil)
+var _ sql.InsertableTable = (*Table)(nil)
+var _ sql.UpdatableTable = (*Table)(nil)
+var _ sql.DeletableTable = (*Table)(nil)
+var _ sql.ReplaceableTable = (*Table)(nil)
 var _ sql.FilteredTable = (*Table)(nil)
 var _ sql.ProjectedTable = (*Table)(nil)
 var _ sql.IndexableTable = (*Table)(nil)
@@ -241,30 +244,70 @@ func encodeIndexValue(value *indexValue) ([]byte, error) {
 	return buf.Bytes(), nil
 }
 
-// Insert a new row into the table.
+type tableEditor struct {
+	table *Table
+}
+
+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 {
+	// 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 *Table) Inserter(*sql.Context) sql.RowInserter {
+	return &tableEditor{t}
+}
+
+func (t *Table) Updater(*sql.Context) sql.RowUpdater {
+	return &tableEditor{t}
+}
+
+func (t *Table) Replacer(*sql.Context) sql.RowReplacer {
+	return &tableEditor{t}
+}
+
+func (t *Table) Deleter(*sql.Context) sql.RowDeleter {
+	return &tableEditor{t}
+}
+
+// Convenience method to avoid having to create an inserter in test setup
 func (t *Table) Insert(ctx *sql.Context, row sql.Row) error {
-	if err := checkRow(t.schema, row); err != nil {
+	inserter := t.Inserter(ctx)
+	if err := inserter.Insert(ctx, row); err != nil {
+		return err
+	}
+	return inserter.Close(ctx)
+}
+
+// Insert a new row into the table.
+func (t *tableEditor) Insert(ctx *sql.Context, row sql.Row) error {
+	if err := checkRow(t.table.schema, row); err != nil {
 		return err
 	}
 
-	key := string(t.keys[t.insert])
-	t.insert++
-	if t.insert == len(t.keys) {
-		t.insert = 0
+	key := string(t.table.keys[t.table.insert])
+	t.table.insert++
+	if t.table.insert == len(t.table.keys) {
+		t.table.insert = 0
 	}
 
-	t.partitions[key] = append(t.partitions[key], row)
+	t.table.partitions[key] = append(t.table.partitions[key], row)
 	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 {
+func (t *tableEditor) Delete(ctx *sql.Context, row sql.Row) error {
+	if err := checkRow(t.table.schema, row); err != nil {
 		return err
 	}
 
 	matches := false
-	for partitionIndex, partition := range t.partitions {
+	for partitionIndex, partition := range t.table.partitions {
 		for partitionRowIndex, partitionRow := range partition {
 			matches = true
 			for rIndex, val := range row {
@@ -274,7 +317,7 @@ func (t *Table) Delete(ctx *sql.Context, row sql.Row) error {
 				}
 			}
 			if matches {
-				t.partitions[partitionIndex] = append(partition[:partitionRowIndex], partition[partitionRowIndex+1:]...)
+				t.table.partitions[partitionIndex] = append(partition[:partitionRowIndex], partition[partitionRowIndex+1:]...)
 				break
 			}
 		}
@@ -290,16 +333,16 @@ func (t *Table) Delete(ctx *sql.Context, row sql.Row) error {
 	return nil
 }
 
-func (t *Table) Update(ctx *sql.Context, oldRow sql.Row, newRow sql.Row) error {
-	if err := checkRow(t.schema, oldRow); err != nil {
+func (t *tableEditor) Update(ctx *sql.Context, oldRow sql.Row, newRow sql.Row) error {
+	if err := checkRow(t.table.schema, oldRow); err != nil {
 		return err
 	}
-	if err := checkRow(t.schema, newRow); err != nil {
+	if err := checkRow(t.table.schema, newRow); err != nil {
 		return err
 	}
 
 	matches := false
-	for partitionIndex, partition := range t.partitions {
+	for partitionIndex, partition := range t.table.partitions {
 		for partitionRowIndex, partitionRow := range partition {
 			matches = true
 			for rIndex, val := range oldRow {
@@ -309,7 +352,7 @@ func (t *Table) Update(ctx *sql.Context, oldRow sql.Row, newRow sql.Row) error {
 				}
 			}
 			if matches {
-				t.partitions[partitionIndex][partitionRowIndex] = newRow
+				t.table.partitions[partitionIndex][partitionRowIndex] = newRow
 				break
 			}
 		}

sql/core.go

@@ -205,28 +205,79 @@ type IndexableTable interface {
 	IndexKeyValues(*Context, []string) (PartitionIndexKeyValueIter, error)
 }
 
-// Inserter allow rows to be inserted in them.
-type Inserter interface {
-	// Insert the given row.
+// InsertableTable is a table that can process insertion of new rows.
+type InsertableTable interface {
+	// Inserter returns an Inserter for this table. The Inserter will get one call to Insert() for each row to be
+	// inserted, and will end with a call to Close() to finalize the insert operation.
+	Inserter(*Context) RowInserter
+}
+
+// RowInserter is an insert cursor that can insert one or more values to a table.
+type RowInserter interface {
+	// 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.
 	Insert(*Context, Row) error
+	// Close finalizes the insert operation, persisting its result.
+	Closer
+}
+
+// DeleteableTable is a table that can process the deletion of rows
+type DeletableTable interface {
+	// Deleter returns a RowDeleter for this table. The RowDeleter will get one call to Delete for each row to be deleted,
+	// and will end with a call to Close() to finalize the delete operation.
+	Deleter(*Context) RowDeleter
+}
+
+// RowDeleter is a delete cursor that can delete one or more rows from a table.
+type RowDeleter interface {
+	// 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.
+	Delete(*Context, Row) error
+	// Close finalizes the delete operation, persisting the result.
+	Closer
+}
+
+type Closer interface {
+	Close(*Context) error
 }
 
-// Deleter allow rows to be deleted from tables.
-type Deleter interface {
-	// Delete the given row. Returns ErrDeleteRowNotFound if the row was not found.
+// RowReplacer is a combination of RowDeleter and RowInserter. We can't embed those interfaces because go doesn't allow
+// for overlapping interfaces (they both declare Close)
+type RowReplacer interface {
+	// 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.
+	Insert(*Context, Row) error
+	// 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.
 	Delete(*Context, Row) error
+	// Close finalizes the replace operation, persisting the result.
+	Closer
 }
 
 // Replacer allows rows to be replaced through a Delete (if applicable) then Insert.
-type Replacer interface {
-	Deleter
-	Inserter
+type ReplaceableTable interface {
+	// Replacer returns a RowReplacer for this table. The RowReplacer will have Insert and optionally Delete called once
+	// for each row, followed by a call to Close() when all rows have been processed.
+	Replacer(ctx *Context) RowReplacer
+}
+
+// UpdateableTable is a table that can process updates of existing rows via update statements.
+type UpdatableTable interface {
+	// Updater returns a RowUpdater for this table. The RowUpdater will have Update called once for each row to be
+	// updated, followed by a call to Close() when all rows have been processed.
+	Updater(ctx *Context) RowUpdater
 }
 
-// Updater allows rows to be updated.
-type Updater interface {
+// RowUpdater is an update cursor that can update one or more rows in a table.
+type RowUpdater interface {
 	// 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.
+	Closer
 }
 
 // Database represents the database.

sql/plan/delete.go

@@ -37,9 +37,9 @@ func (p *DeleteFrom) Children() []sql.Node {
 	return []sql.Node{p.Node}
 }
 
-func getDeletable(node sql.Node) (sql.Deleter, error) {
+func getDeletable(node sql.Node) (sql.DeletableTable, error) {
 	switch node := node.(type) {
-	case sql.Deleter:
+	case sql.DeletableTable:
 		return node, nil
 	case *ResolvedTable:
 		return getDeletableTable(node.Table)
@@ -53,9 +53,9 @@ func getDeletable(node sql.Node) (sql.Deleter, error) {
 	return nil, ErrDeleteFromNotSupported.New()
 }
 
-func getDeletableTable(t sql.Table) (sql.Deleter, error) {
+func getDeletableTable(t sql.Table) (sql.DeletableTable, error) {
 	switch t := t.(type) {
-	case sql.Deleter:
+	case sql.DeletableTable:
 		return t, nil
 	case sql.TableWrapper:
 		return getDeletableTable(t.Underlying())
@@ -76,6 +76,8 @@ func (p *DeleteFrom) Execute(ctx *sql.Context) (int, error) {
 		return 0, err
 	}
 
+	deleter := deletable.Deleter(ctx)
+
 	i := 0
 	for {
 		row, err := iter.Next()
@@ -88,14 +90,18 @@ func (p *DeleteFrom) Execute(ctx *sql.Context) (int, error) {
 			return i, err
 		}
 
-		if err := deletable.Delete(ctx, row); err != nil {
+		if err := deleter.Delete(ctx, row); err != nil {
 			_ = iter.Close()
 			return i, err
 		}
 
 		i++
 	}
 
+	if err := deleter.Close(ctx); err != nil {
+		return 0, err
+	}
+
 	return i, nil
 }
 

sql/plan/insert.go

@@ -45,9 +45,9 @@ func (p *InsertInto) Schema() sql.Schema {
 	}}
 }
 
-func getInsertable(node sql.Node) (sql.Inserter, error) {
+func getInsertable(node sql.Node) (sql.InsertableTable, error) {
 	switch node := node.(type) {
-	case sql.Inserter:
+	case sql.InsertableTable:
 		return node, nil
 	case *ResolvedTable:
 		return getInsertableTable(node.Table)
@@ -56,9 +56,9 @@ func getInsertable(node sql.Node) (sql.Inserter, error) {
 	}
 }
 
-func getInsertableTable(t sql.Table) (sql.Inserter, error) {
+func getInsertableTable(t sql.Table) (sql.InsertableTable, error) {
 	switch t := t.(type) {
-	case sql.Inserter:
+	case sql.InsertableTable:
 		return t, nil
 	case sql.TableWrapper:
 		return getInsertableTable(t.Underlying())
@@ -74,10 +74,10 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 		return 0, err
 	}
 
-	var replaceable sql.Replacer
+	var replaceable sql.ReplaceableTable
 	if p.IsReplace {
 		var ok bool
-		replaceable, ok = insertable.(sql.Replacer)
+		replaceable, ok = insertable.(sql.ReplaceableTable)
 		if !ok {
 			return 0, ErrReplaceIntoNotSupported.New()
 		}
@@ -132,6 +132,14 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 		return 0, err
 	}
 
+	var inserter sql.RowInserter
+	var replacer sql.RowReplacer
+	if replaceable != nil {
+		replacer = replaceable.Replacer(ctx)
+	} else {
+		inserter = insertable.Inserter(ctx)
+	}
+
 	i := 0
 	for {
 		row, err := iter.Next()
@@ -163,8 +171,8 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 			}
 		}
 
-		if replaceable != nil {
-			if err = replaceable.Delete(ctx, row); err != nil {
+		if replacer != nil {
+			if err = replacer.Delete(ctx, row); err != nil {
 				if err != sql.ErrDeleteRowNotFound {
 					_ = iter.Close()
 					return i, err
@@ -173,19 +181,29 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 				i++
 			}
 
-			if err = replaceable.Insert(ctx, row); err != nil {
+			if err = replacer.Insert(ctx, row); err != nil {
 				_ = iter.Close()
 				return i, err
 			}
 		} else {
-			if err := insertable.Insert(ctx, row); err != nil {
+			if err := inserter.Insert(ctx, row); err != nil {
 				_ = iter.Close()
 				return i, err
 			}
 		}
 		i++
 	}
 
+	if replacer != nil {
+		if err := replacer.Close(ctx); err != nil {
+			return 0, err
+		}
+	} else {
+		if err := inserter.Close(ctx); err != nil {
+			return 0, err
+		}
+	}
+
 	return i, nil
 }