Merge pull request #2439 from dolthub/nicktobey/autoinc

For AutoIncrement lock modes other than "interleaved", hold the lock for the duration of the insert iter.

memory/table_editor.go

@@ -277,6 +277,11 @@ func (t *tableEditor) SetAutoIncrementValue(ctx *sql.Context, val uint64) error
 	return nil
 }
 
+func (t *tableEditor) AcquireAutoIncrementLock(ctx *sql.Context) (func(), error) {
+	// TODO: Add concurrency tests for AutoIncrement locking modes.
+	return func() {}, nil
+}
+
 func (t *tableEditor) PreciseMatch() bool {
 	return true
 }

sql/fulltext/multi_editor.go

@@ -140,6 +140,11 @@ func (editor MultiTableEditor) SetAutoIncrementValue(ctx *sql.Context, u uint64)
 	return editor.primary.(sql.AutoIncrementSetter).SetAutoIncrementValue(ctx, u)
 }
 
+func (editor MultiTableEditor) AcquireAutoIncrementLock(ctx *sql.Context) (func(), error) {
+	// TODO: Add concurrency tests for AutoIncrement locking modes.
+	return func() {}, nil
+}
+
 // Close implements the interface sql.TableEditor.
 func (editor MultiTableEditor) Close(ctx *sql.Context) error {
 	var err error

sql/plan/foreign_key_handler.go

@@ -30,6 +30,10 @@ type ForeignKeyHandler struct {
 	AllUpdaters  []sql.ForeignKeyEditor
 }
 
+func (n *ForeignKeyHandler) Underlying() sql.Table {
+	return n.Table
+}
+
 var _ sql.Node = (*ForeignKeyHandler)(nil)
 var _ sql.CollationCoercible = (*ForeignKeyHandler)(nil)
 var _ sql.Table = (*ForeignKeyHandler)(nil)
@@ -41,6 +45,7 @@ var _ sql.TableEditor = (*ForeignKeyHandler)(nil)
 var _ sql.RowInserter = (*ForeignKeyHandler)(nil)
 var _ sql.RowUpdater = (*ForeignKeyHandler)(nil)
 var _ sql.RowDeleter = (*ForeignKeyHandler)(nil)
+var _ sql.TableWrapper = (*ForeignKeyHandler)(nil)
 
 // Resolved implements the interface sql.Node.
 func (n *ForeignKeyHandler) Resolved() bool {

sql/rowexec/dml.go

@@ -15,6 +15,7 @@
 package rowexec
 
 import (
+	"errors"
 	"fmt"
 	"strings"
 	"sync"
@@ -55,7 +56,28 @@ func (b *BaseBuilder) buildInsertInto(ctx *sql.Context, ii *plan.InsertInto, row
 		return nil, err
 	}
 
+	var unlocker func()
 	insertExpressions := getInsertExpressions(ii.Source)
+	if ii.HasUnspecifiedAutoInc {
+		_, i, _ := sql.SystemVariables.GetGlobal("innodb_autoinc_lock_mode")
+		lockMode, ok := i.(int64)
+		if !ok {
+			return nil, errors.New(fmt.Sprintf("unexpected type for innodb_autoinc_lock_mode, expected int64, got %T", i))
+		}
+		// Lock modes "traditional" (0) and "consecutive" (1) require that a single lock is held for the entire iteration.
+		// Lock mode "interleaved" (2) will acquire the lock only when inserting into the table.
+		if lockMode != 2 {
+			autoIncrementable, ok := sql.GetUnderlyingTable(insertable).(sql.AutoIncrementTable)
+			if !ok {
+				return nil, errors.New("auto increment expression on non-AutoIncrement table. This should not be possible")
+			}
+
+			unlocker, err = autoIncrementable.AutoIncrementSetter(ctx).AcquireAutoIncrementLock(ctx)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
 	insertIter := &insertIter{
 		schema:              dstSchema,
 		tableNode:           ii.Destination,
@@ -64,6 +86,7 @@ func (b *BaseBuilder) buildInsertInto(ctx *sql.Context, ii *plan.InsertInto, row
 		updater:             updater,
 		rowSource:           rowIter,
 		hasAutoAutoIncValue: ii.HasUnspecifiedAutoInc,
+		unlocker:            unlocker,
 		updateExprs:         ii.OnDupExprs,
 		insertExprs:         insertExpressions,
 		checks:              ii.Checks(),

sql/rowexec/insert.go

@@ -37,6 +37,7 @@ type insertIter struct {
 	rowSource           sql.RowIter
 	lastInsertIdUpdated bool
 	hasAutoAutoIncValue bool
+	unlocker            func()
 	ctx                 *sql.Context
 	insertExprs         []sql.Expression
 	updateExprs         []sql.Expression
@@ -257,6 +258,9 @@ func (i *insertIter) resolveValues(ctx *sql.Context, insertRow sql.Row) error {
 func (i *insertIter) Close(ctx *sql.Context) error {
 	if !i.closed {
 		i.closed = true
+		if i.unlocker != nil {
+			i.unlocker()
+		}
 		var rsErr, iErr, rErr, uErr error
 		if i.rowSource != nil {
 			rsErr = i.rowSource.Close(ctx)

sql/session.go

@@ -73,6 +73,7 @@ type Session interface {
 	// SetUserVariable sets the given user variable to the value given for this session, or creates it for this session.
 	SetUserVariable(ctx *Context, varName string, value interface{}, typ Type) error
 	// GetSessionVariable returns this session's value of the system variable with the given name.
+	// To access global scope, use sql.SystemVariables.GetGlobal instead.
 	GetSessionVariable(ctx *Context, sysVarName string) (interface{}, error)
 	// GetUserVariable returns this session's value of the user variable with the given name, along with its most
 	// appropriate type.
@@ -83,7 +84,7 @@ type Session interface {
 	// To access global scope, use sql.StatusVariables instead.
 	GetStatusVariable(ctx *Context, statVarName string) (interface{}, error)
 	// SetStatusVariable sets the value of the status variable with session scope with the given name.
-	// To access global scope, use sql.StatusVariables instead.
+	// To access global scope, use sql.StatusVariables.GetGlobal instead.
 	SetStatusVariable(ctx *Context, statVarName string, val interface{}) error
 	// GetAllStatusVariables returns a map of all status variables with session scope and their values.
 	// To access global scope, use sql.StatusVariables instead.

sql/tables.go

@@ -67,6 +67,13 @@ type TableWrapper interface {
 	Underlying() Table
 }
 
+func GetUnderlyingTable(t Table) Table {
+	if tw, ok := t.(TableWrapper); ok {
+		return GetUnderlyingTable(tw.Underlying())
+	}
+	return t
+}
+
 // MutableTableWrapper is a TableWrapper that can change its underlying table.
 type MutableTableWrapper interface {
 	TableWrapper
@@ -332,6 +339,11 @@ type AutoIncrementTable interface {
 type AutoIncrementSetter interface {
 	// SetAutoIncrementValue sets a new AUTO_INCREMENT value.
 	SetAutoIncrementValue(*Context, uint64) error
+
+	// AcquireAutoIncrementLock acquires (if necessary) an exclusive lock on generating auto-increment values for the underlying table.
+	// This is called when @@innodb_autoinc_lock_mode is set to 0 (traditional) or 1 (consecutive), in order to guarentee that insert
+	// operations get a consecutive range of generated ids. The function returns a callback to release the lock.
+	AcquireAutoIncrementLock(ctx *Context) (func(), error)
 	// Closer finalizes the set operation, persisting the result.
 	Closer
 }

sql/variables/system_variables.go

@@ -1029,9 +1029,8 @@ var systemVars = map[string]sql.SystemVariable{
 		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
 		Dynamic:           false,
 		SetVarHintApplies: false,
-		// TODO: lower bound should be 0: https://github.com/dolthub/dolt/issues/7634
-		Type:    types.NewSystemIntType("innodb_autoinc_lock_mode", 2, 2, false),
-		Default: int64(2),
+		Type:              types.NewSystemIntType("innodb_autoinc_lock_mode", 0, 2, false),
+		Default:           int64(2),
 	},
 	// Row locking is currently not supported. This variable is provided for 3p tools, and we always return the
 	// Lowest value allowed by MySQL, which is 1. If you attempt to set this value to anything other than 1, errors ensue.

sql/variables/system_variables_test.go

@@ -61,17 +61,20 @@ func TestInitSystemVars(t *testing.T) {
 		{
 			varName: "innodb_autoinc_lock_mode",
 			varVal:  0,
-			err:     sql.ErrInvalidSystemVariableValue,
 		},
 		{
 			varName: "innodb_autoinc_lock_mode",
 			varVal:  1,
-			err:     sql.ErrInvalidSystemVariableValue,
 		},
 		{
 			varName: "innodb_autoinc_lock_mode",
 			varVal:  2,
 		},
+		{
+			varName: "innodb_autoinc_lock_mode",
+			varVal:  3,
+			err:     sql.ErrInvalidSystemVariableValue,
+		},
 	}
 
 	for _, test := range tests {