Support clickhouse v2

database/cassandra/cassandra.go

@@ -199,14 +199,14 @@ func (c *Cassandra) Close() error {
 }
 
 func (c *Cassandra) Lock() error {
-	if !c.isLocked.CAS(false, true) {
+	if !c.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 
 func (c *Cassandra) Unlock() error {
-	if !c.isLocked.CAS(true, false) {
+	if !c.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/clickhouse/clickhouse.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2"
 	"go.uber.org/atomic"
 
 	"github.com/golang-migrate/migrate/v4"
@@ -73,11 +74,12 @@ func (ch *ClickHouse) Open(dsn string) (database.Driver, error) {
 		return nil, err
 	}
 	q := migrate.FilterCustomQuery(purl)
-	q.Scheme = "tcp"
-	conn, err := sql.Open("clickhouse", q.String())
+	q.Scheme = "clickhouse"
+	opt, err := clickhouse.ParseDSN(q.String())
 	if err != nil {
 		return nil, err
 	}
+	conn := clickhouse.OpenDB(opt)
 
 	multiStatementMaxSize := DefaultMultiStatementMaxSize
 	if s := purl.Query().Get("x-multi-statement-max-size"); len(s) > 0 {
@@ -167,7 +169,7 @@ func (ch *ClickHouse) Version() (int, bool, error) {
 	var (
 		version int
 		dirty   uint8
-		query   = "SELECT version, dirty FROM `" + ch.config.MigrationsTable + "` ORDER BY sequence DESC LIMIT 1"
+		query   = "SELECT version, dirty FROM " + quoteIdentifier(ch.config.DatabaseName) + "." + quoteIdentifier(ch.config.MigrationsTable) + " ORDER BY sequence DESC LIMIT 1"
 	)
 	if err := ch.conn.QueryRow(query).Scan(&version, &dirty); err != nil {
 		if err == sql.ErrNoRows {
@@ -192,7 +194,7 @@ func (ch *ClickHouse) SetVersion(version int, dirty bool) error {
 		return err
 	}
 
-	query := "INSERT INTO " + ch.config.MigrationsTable + " (version, dirty, sequence) VALUES (?, ?, ?)"
+	query := "INSERT INTO " + quoteIdentifier(ch.config.DatabaseName) + "." + quoteIdentifier(ch.config.MigrationsTable) + " (version, dirty, sequence) VALUES (?, ?, ?)"
 	if _, err := tx.Exec(query, version, bool(dirty), time.Now().UnixNano()); err != nil {
 		return &database.Error{OrigErr: err, Query: []byte(query)}
 	}
@@ -234,18 +236,18 @@ func (ch *ClickHouse) ensureVersionTable() (err error) {
 	// if not, create the empty migration table
 	if len(ch.config.ClusterName) > 0 {
 		query = fmt.Sprintf(`
-			CREATE TABLE %s ON CLUSTER %s (
-				version    Int64,
-				dirty      UInt8,
-				sequence   UInt64
-			) Engine=%s`, ch.config.MigrationsTable, ch.config.ClusterName, ch.config.MigrationsTableEngine)
+      CREATE TABLE %s.%s ON CLUSTER %s (
+        version    Int64,
+        dirty      UInt8,
+        sequence   UInt64
+      ) Engine=%s`, quoteIdentifier(ch.config.DatabaseName), quoteIdentifier(ch.config.MigrationsTable), ch.config.ClusterName, ch.config.MigrationsTableEngine)
 	} else {
 		query = fmt.Sprintf(`
-			CREATE TABLE %s (
-				version    Int64,
-				dirty      UInt8,
-				sequence   UInt64
-			) Engine=%s`, ch.config.MigrationsTable, ch.config.MigrationsTableEngine)
+      CREATE TABLE %s.%s (
+        version    Int64,
+        dirty      UInt8,
+        sequence   UInt64
+      ) Engine=%s`, quoteIdentifier(ch.config.DatabaseName), quoteIdentifier(ch.config.MigrationsTable), ch.config.MigrationsTableEngine)
 	}
 
 	if strings.HasSuffix(ch.config.MigrationsTableEngine, "Tree") {
@@ -291,14 +293,14 @@ func (ch *ClickHouse) Drop() (err error) {
 }
 
 func (ch *ClickHouse) Lock() error {
-	if !ch.isLocked.CAS(false, true) {
+	if !ch.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 
 	return nil
 }
 func (ch *ClickHouse) Unlock() error {
-	if !ch.isLocked.CAS(true, false) {
+	if !ch.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 

database/clickhouse/clickhouse_test.go

@@ -6,9 +6,10 @@ import (
 	sqldriver "database/sql/driver"
 	"fmt"
 	"log"
+	"net/url"
 	"testing"
 
-	_ "github.com/ClickHouse/clickhouse-go"
+	_ "github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/dhui/dktest"
 	"github.com/golang-migrate/migrate/v4"
 	"github.com/golang-migrate/migrate/v4/database/clickhouse"
@@ -33,12 +34,12 @@ var (
 func clickhouseConnectionString(host, port, engine string) string {
 	if engine != "" {
 		return fmt.Sprintf(
-			"clickhouse://%v:%v?username=user&password=password&database=db&x-multi-statement=true&x-migrations-table-engine=%v&debug=false",
+			"clickhouse://%v:%v/db?username=user&password=password&x-multi-statement=true&x-migrations-table-engine=%v&debug=false",
 			host, port, engine)
 	}
 
 	return fmt.Sprintf(
-		"clickhouse://%v:%v?username=user&password=password&database=db&x-multi-statement=true&debug=false",
+		"clickhouse://%v:%v/db?username=user&password=password&x-multi-statement=true&debug=false",
 		host, port)
 }
 
@@ -48,7 +49,9 @@ func isReady(ctx context.Context, c dktest.ContainerInfo) bool {
 		return false
 	}
 
-	db, err := sql.Open("clickhouse", clickhouseConnectionString(ip, port, ""))
+	addr := clickhouseConnectionString(ip, port, "")
+	purl, _ := url.Parse(addr)
+	db, err := sql.Open("clickhouse", migrate.FilterCustomQuery(purl).String())
 
 	if err != nil {
 		log.Println("open error", err)
@@ -114,7 +117,8 @@ func testSimpleWithInstanceDefaultConfigValues(t *testing.T) {
 		}
 
 		addr := clickhouseConnectionString(ip, port, "")
-		conn, err := sql.Open("clickhouse", addr)
+		purl, _ := url.Parse(addr)
+		conn, err := sql.Open("clickhouse", migrate.FilterCustomQuery(purl).String())
 		if err != nil {
 			t.Fatal(err)
 		}

database/firebird/firebird.go

@@ -107,14 +107,14 @@ func (f *Firebird) Close() error {
 }
 
 func (f *Firebird) Lock() error {
-	if !f.isLocked.CAS(false, true) {
+	if !f.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 
 func (f *Firebird) Unlock() error {
-	if !f.isLocked.CAS(true, false) {
+	if !f.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/ql/ql.go

@@ -166,13 +166,13 @@ func (m *Ql) Drop() (err error) {
 	return nil
 }
 func (m *Ql) Lock() error {
-	if !m.isLocked.CAS(false, true) {
+	if !m.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 func (m *Ql) Unlock() error {
-	if !m.isLocked.CAS(true, false) {
+	if !m.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/redshift/redshift.go

@@ -127,14 +127,14 @@ func (p *Redshift) Close() error {
 
 // Redshift does not support advisory lock functions: https://docs.aws.amazon.com/redshift/latest/dg/c_unsupported-postgresql-functions.html
 func (p *Redshift) Lock() error {
-	if !p.isLocked.CAS(false, true) {
+	if !p.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 
 func (p *Redshift) Unlock() error {
-	if !p.isLocked.CAS(true, false) {
+	if !p.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/rqlite/rqlite.go

@@ -143,7 +143,7 @@ func (r *Rqlite) Close() error {
 // If the implementation can't provide this functionality, return nil.
 // Return database.ErrLocked if database is already locked.
 func (r *Rqlite) Lock() error {
-	if !r.isLocked.CAS(false, true) {
+	if !r.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
@@ -152,7 +152,7 @@ func (r *Rqlite) Lock() error {
 // Unlock should release the lock. Migrate will call this function after
 // all migrations have been run.
 func (r *Rqlite) Unlock() error {
-	if !r.isLocked.CAS(true, false) {
+	if !r.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/snowflake/snowflake.go

@@ -159,14 +159,14 @@ func (p *Snowflake) Close() error {
 }
 
 func (p *Snowflake) Lock() error {
-	if !p.isLocked.CAS(false, true) {
+	if !p.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 
 func (p *Snowflake) Unlock() error {
-	if !p.isLocked.CAS(true, false) {
+	if !p.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/spanner/spanner.go

@@ -153,15 +153,15 @@ func (s *Spanner) Close() error {
 // Lock implements database.Driver but doesn't do anything because Spanner only
 // enqueues the UpdateDatabaseDdlRequest.
 func (s *Spanner) Lock() error {
-	if swapped := s.lock.CAS(unlockedVal, lockedVal); swapped {
+	if swapped := s.lock.CompareAndSwap(unlockedVal, lockedVal); swapped {
 		return nil
 	}
 	return ErrLockHeld
 }
 
 // Unlock implements database.Driver but no action required, see Lock.
 func (s *Spanner) Unlock() error {
-	if swapped := s.lock.CAS(lockedVal, unlockedVal); swapped {
+	if swapped := s.lock.CompareAndSwap(lockedVal, unlockedVal); swapped {
 		return nil
 	}
 	return ErrLockNotHeld

database/sqlcipher/sqlcipher.go

@@ -178,14 +178,14 @@ func (m *Sqlite) Drop() (err error) {
 }
 
 func (m *Sqlite) Lock() error {
-	if !m.isLocked.CAS(false, true) {
+	if !m.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 
 func (m *Sqlite) Unlock() error {
-	if !m.isLocked.CAS(true, false) {
+	if !m.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/sqlite/sqlite.go

@@ -178,14 +178,14 @@ func (m *Sqlite) Drop() (err error) {
 }
 
 func (m *Sqlite) Lock() error {
-	if !m.isLocked.CAS(false, true) {
+	if !m.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 
 func (m *Sqlite) Unlock() error {
-	if !m.isLocked.CAS(true, false) {
+	if !m.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/sqlite3/sqlite3.go

@@ -178,14 +178,14 @@ func (m *Sqlite) Drop() (err error) {
 }
 
 func (m *Sqlite) Lock() error {
-	if !m.isLocked.CAS(false, true) {
+	if !m.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 
 func (m *Sqlite) Unlock() error {
-	if !m.isLocked.CAS(true, false) {
+	if !m.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/stub/stub.go

@@ -50,14 +50,14 @@ func (s *Stub) Close() error {
 }
 
 func (s *Stub) Lock() error {
-	if !s.isLocked.CAS(false, true) {
+	if !s.isLocked.CompareAndSwap(false, true) {
 		return database.ErrLocked
 	}
 	return nil
 }
 
 func (s *Stub) Unlock() error {
-	if !s.isLocked.CAS(true, false) {
+	if !s.isLocked.CompareAndSwap(true, false) {
 		return database.ErrNotLocked
 	}
 	return nil

database/util.go

@@ -21,7 +21,7 @@ func GenerateAdvisoryLockId(databaseName string, additionalNames ...string) (str
 
 // CasRestoreOnErr CAS wrapper to automatically restore the lock state on error
 func CasRestoreOnErr(lock *atomic.Bool, o, n bool, casErr error, f func() error) error {
-	if !lock.CAS(o, n) {
+	if !lock.CompareAndSwap(o, n) {
 		return casErr
 	}
 	if err := f(); err != nil {