Support transaction isolation level in BeginTx

Author: Maciej Zimnoch

AUTHORS

@@ -61,6 +61,7 @@ Xiangyu Hu <xiangyu.hu at outlook.com>
 Xiaobing Jiang <s7v7nislands at gmail.com>
 Xiuming Chen <cc at cxm.cc>
 Zhenye Xie <xiezhenye at gmail.com>
+Maciej Zimnoch <[email protected]>
 
 # Organizations
 

connection_go18.go

@@ -41,15 +41,23 @@ func (mc *mysqlConn) Ping(ctx context.Context) error {
 
 // BeginTx implements driver.ConnBeginTx interface
 func (mc *mysqlConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
-	if sql.IsolationLevel(opts.Isolation) != sql.LevelDefault {
-		// TODO: support isolation levels
-		return nil, errors.New("mysql: isolation levels not supported")
-	}
 	if opts.ReadOnly {
 		// TODO: support read-only transactions
 		return nil, errors.New("mysql: read-only transactions not supported")
 	}
 
+	if sql.IsolationLevel(opts.Isolation) != sql.LevelDefault {
+		level, err := mapIsolationLevel(opts.Isolation)
+		if err != nil {
+			return nil, err
+		}
+		err = mc.exec("SET TRANSACTION ISOLATION LEVEL " + level)
+		if err != nil {
+			return nil, err
+		}
+		mc.finish()
+	}
+
 	if err := mc.watchCancel(ctx); err != nil {
 		return nil, err
 	}

driver_go18_test.go

@@ -16,6 +16,7 @@ import (
 	"database/sql/driver"
 	"fmt"
 	"reflect"
+	"sync"
 	"testing"
 	"time"
 )
@@ -468,3 +469,100 @@ func TestContextCancelBegin(t *testing.T) {
 		}
 	})
 }
+
+func TestContextBeginIsolationLevel(t *testing.T) {
+	runTests(t, dsn, func(dbt *DBTest) {
+		dbt.mustExec("CREATE TABLE test (v INTEGER)")
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		// Waitgroup syncing BeginTx
+		beginWg := sync.WaitGroup{}
+		beginWg.Add(2)
+
+		// Waitgroup syncing insert in writer transaction
+		insertWg := sync.WaitGroup{}
+		insertWg.Add(1)
+
+		// Waitgroup syncing writer transaction commit before reader reads
+		readWg := sync.WaitGroup{}
+		readWg.Add(1)
+
+		// Waitgroup syncing commit in writer transaction
+		commitWg := sync.WaitGroup{}
+		commitWg.Add(1)
+
+		// Waitgroup syncing end of both goroutines
+		testDoneWg := sync.WaitGroup{}
+		testDoneWg.Add(2)
+
+		repeatableReadGoroutine := func() {
+			tx, err := dbt.db.BeginTx(ctx, &sql.TxOptions{
+				Isolation: sql.LevelRepeatableRead,
+			})
+			if err != nil {
+				dbt.Fatal(err)
+			}
+			beginWg.Done()
+			// Wait until other session will begin it's transaction
+			beginWg.Wait()
+
+			_, err = tx.ExecContext(ctx, "INSERT INTO test VALUES (1)")
+			if err != nil {
+				dbt.Fatal(err)
+			}
+			insertWg.Done()
+
+			// Wait until reader transaction finish reading
+			readWg.Wait()
+
+			err = tx.Commit()
+			if err != nil {
+				dbt.Fatal(err)
+			}
+			commitWg.Done()
+
+			testDoneWg.Done()
+		}
+
+		readCommitedGoroutine := func() {
+			tx, err := dbt.db.BeginTx(ctx, &sql.TxOptions{
+				Isolation: sql.LevelReadCommitted,
+			})
+			if err != nil {
+				dbt.Fatal(err)
+			}
+			beginWg.Done()
+			// Wait until writer transaction will begin
+			beginWg.Wait()
+			// Wait until writer transaction will insert value
+			insertWg.Wait()
+			var v int
+			row := tx.QueryRowContext(ctx, "SELECT COUNT(*) FROM test")
+			if err := row.Scan(&v); err != nil {
+				dbt.Fatal(err)
+			}
+			// Because writer transaction wasn't commited yet, it should be available
+			if v != 0 {
+				dbt.Errorf("expected val to be 0, got %d", v)
+			}
+			readWg.Done()
+			// Wait until writer transaction will commit
+			commitWg.Wait()
+			row = tx.QueryRowContext(ctx, "SELECT COUNT(*) FROM test")
+			if err := row.Scan(&v); err != nil {
+				dbt.Fatal(err)
+			}
+			// Data written by writer transaction is already commited, it should be selectable
+			if v != 1 {
+				dbt.Errorf("expected val to be 1, got %d", v)
+			}
+			tx.Commit()
+			testDoneWg.Done()
+		}
+
+		go repeatableReadGoroutine()
+		go readCommitedGoroutine()
+		testDoneWg.Wait()
+	})
+}

utils_go18.go

@@ -12,6 +12,7 @@ package mysql
 
 import (
 	"crypto/tls"
+	"database/sql"
 	"database/sql/driver"
 	"errors"
 )
@@ -31,3 +32,18 @@ func namedValueToValue(named []driver.NamedValue) ([]driver.Value, error) {
 	}
 	return dargs, nil
 }
+
+func mapIsolationLevel(level driver.IsolationLevel) (string, error) {
+	switch sql.IsolationLevel(level) {
+	case sql.LevelRepeatableRead:
+		return "REPEATABLE READ", nil
+	case sql.LevelReadCommitted:
+		return "READ COMMITTED", nil
+	case sql.LevelReadUncommitted:
+		return "READ UNCOMMITTED", nil
+	case sql.LevelSerializable:
+		return "SERIALIZABLE", nil
+	default:
+		return "", errors.New("mysql: unsupported isolation level: " + string(level))
+	}
+}

utils_go18_test.go

@@ -0,0 +1,54 @@
+// Go MySQL Driver - A MySQL-Driver for Go's database/sql package
+//
+// Copyright 2017 The Go-MySQL-Driver Authors. All rights reserved.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at http://mozilla.org/MPL/2.0/.
+
+// +build go1.8
+
+package mysql
+
+import (
+	"database/sql"
+	"database/sql/driver"
+	"testing"
+)
+
+func TestIsolationLevelMapping(t *testing.T) {
+
+	data := []struct {
+		level    driver.IsolationLevel
+		expected string
+	}{
+		{
+			level:    driver.IsolationLevel(sql.LevelReadCommitted),
+			expected: "READ COMMITTED",
+		},
+		{
+			level:    driver.IsolationLevel(sql.LevelRepeatableRead),
+			expected: "REPEATABLE READ",
+		},
+		{
+			level:    driver.IsolationLevel(sql.LevelReadUncommitted),
+			expected: "READ UNCOMMITTED",
+		},
+		{
+			level:    driver.IsolationLevel(sql.LevelSerializable),
+			expected: "SERIALIZABLE",
+		},
+	}
+
+	for i, td := range data {
+		if actual, err := mapIsolationLevel(td.level); actual != td.expected || err != nil {
+			t.Fatal(i, td.expected, actual, err)
+		}
+	}
+
+	// check unsupported mapping
+	if actual, err := mapIsolationLevel(driver.IsolationLevel(sql.LevelLinearizable)); actual != "" || err == nil {
+		t.Fatal("Expected error on unsupported isolation level")
+	}
+
+}