postgres: Move lock out of ensureVersionTable, for consistency with o…

…ther SQL operations (golang-migrate#173)

* Consistently lock in ensureVersionTable and do not call ensureVersionTable from Drop across all database implementations

* Add test for dropping postgres databases

* Fix failing database tests

* Fix CockroachDb test, lock table should be created before versionTable

* Add Initialize() to Driver interface, and add integration tests for Drop() between database implementations and migrate

* Remove Initialize, document breaking behaviour of Drop

* Revert introduction of Initialize method

* Removed Initialize in Stub as well

* Remove call to non-existent Initialize and make sure to close re-initialized database connections

* Revert changes to TestDrop in database/testing

* Split Test and TestMigrate into different test entrypoints

* Remove unused import in migrate_testing

* Remove erroneous code to fix tests

* Add stub source imports to database tests

* Add Stub source to migrate tests

* Use example migrations for tests

* Add file driver to database tests

* Align database directory layout

* Add file source driver to Cassandra

* Review changes

* Minor syntactic change for cleaner diff

database/cassandra/cassandra.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/gocql/gocql"
 	"github.com/golang-migrate/migrate/v4/database"
+	"github.com/hashicorp/go-multierror"
 )
 
 func init() {
@@ -240,13 +241,29 @@ func (c *Cassandra) Drop() error {
 			return err
 		}
 	}
-	// Re-create the version table
-	return c.ensureVersionTable()
+
+	return nil
 }
 
-// Ensure version table exists
-func (c *Cassandra) ensureVersionTable() error {
-	err := c.session.Query(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (version bigint, dirty boolean, PRIMARY KEY(version))", c.config.MigrationsTable)).Exec()
+// ensureVersionTable checks if versions table exists and, if not, creates it.
+// Note that this function locks the database, which deviates from the usual
+// convention of "caller locks" in the Cassandra type.
+func (c *Cassandra) ensureVersionTable() (err error) {
+	if err = c.Lock(); err != nil {
+		return err
+	}
+
+	defer func() {
+		if e := c.Unlock(); e != nil {
+			if err == nil {
+				err = e
+			} else {
+				err = multierror.Append(err, e)
+			}
+		}
+	}()
+
+	err = c.session.Query(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (version bigint, dirty boolean, PRIMARY KEY(version))", c.config.MigrationsTable)).Exec()
 	if err != nil {
 		return err
 	}

database/cassandra/cassandra_test.go

@@ -3,6 +3,7 @@ package cassandra
 import (
 	"context"
 	"fmt"
+	"github.com/golang-migrate/migrate/v4"
 	"strconv"
 	"testing"
 )
@@ -15,6 +16,7 @@ import (
 import (
 	dt "github.com/golang-migrate/migrate/v4/database/testing"
 	"github.com/golang-migrate/migrate/v4/dktesting"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
 )
 
 var (
@@ -72,3 +74,25 @@ func Test(t *testing.T) {
 		dt.Test(t, d, []byte("SELECT table_name from system_schema.tables"))
 	})
 }
+
+func TestMigrate(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.Port(9042)
+		if err != nil {
+			t.Fatal("Unable to get mapped port:", err)
+		}
+		addr := fmt.Sprintf("cassandra://%v:%v/testks", ip, port)
+		p := &Cassandra{}
+		d, err := p.Open(addr)
+		if err != nil {
+			t.Fatalf("%v", err)
+		}
+		defer d.Close()
+
+		m, err := migrate.NewWithDatabaseInstance("file://./examples/migrations", "testks", d)
+		if err != nil {
+			t.Fatalf("%v", err)
+		}
+		dt.TestMigrate(t, m, []byte("SELECT table_name from system_schema.tables"))
+	})
+}

database/cassandra/examples/migrations/1_simple_select.down.sql

@@ -0,0 +1 @@
+SELECT table_name from system_schema.tables

database/cassandra/examples/migrations/1_simple_select.up.sql

@@ -0,0 +1 @@
+SELECT table_name from system_schema.tables

database/clickhouse/clickhouse.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/golang-migrate/migrate/v4"
 	"github.com/golang-migrate/migrate/v4/database"
+	"github.com/hashicorp/go-multierror"
 )
 
 var DefaultMigrationsTable = "schema_migrations"
@@ -159,7 +160,25 @@ func (ch *ClickHouse) SetVersion(version int, dirty bool) error {
 	return tx.Commit()
 }
 
-func (ch *ClickHouse) ensureVersionTable() error {
+
+// ensureVersionTable checks if versions table exists and, if not, creates it.
+// Note that this function locks the database, which deviates from the usual
+// convention of "caller locks" in the ClickHouse type.
+func (ch *ClickHouse) ensureVersionTable() (err error) {
+	if err = ch.Lock(); err != nil {
+		return err
+	}
+
+	defer func() {
+		if e := ch.Unlock(); e != nil {
+			if err == nil {
+				err = e
+			} else {
+				err = multierror.Append(err, e)
+			}
+		}
+	}()
+
 	var (
 		table string
 		query = "SHOW TABLES FROM " + ch.config.DatabaseName + " LIKE '" + ch.config.MigrationsTable + "'"
@@ -207,7 +226,7 @@ func (ch *ClickHouse) Drop() error {
 			return &database.Error{OrigErr: err, Query: []byte(query)}
 		}
 	}
-	return ch.ensureVersionTable()
+	return nil
 }
 
 func (ch *ClickHouse) Lock() error   { return nil }

database/cockroachdb/cockroachdb.go

@@ -13,6 +13,7 @@ import (
 
 import (
 	"github.com/cockroachdb/cockroach-go/crdb"
+	"github.com/hashicorp/go-multierror"
 	"github.com/lib/pq"
 )
 
@@ -85,11 +86,12 @@ func WithInstance(instance *sql.DB, config *Config) (database.Driver, error) {
 		config: config,
 	}
 
-	if err := px.ensureVersionTable(); err != nil {
+	// ensureVersionTable is a locking operation, so we need to ensureLockTable before we ensureVersionTable.
+	if err := px.ensureLockTable(); err != nil {
 		return nil, err
 	}
 
-	if err := px.ensureLockTable(); err != nil {
+	if err := px.ensureVersionTable(); err != nil {
 		return nil, err
 	}
 
@@ -294,15 +296,29 @@ func (c *CockroachDb) Drop() error {
 				return &database.Error{OrigErr: err, Query: []byte(query)}
 			}
 		}
-		if err := c.ensureVersionTable(); err != nil {
-			return err
-		}
 	}
 
 	return nil
 }
 
-func (c *CockroachDb) ensureVersionTable() error {
+// ensureVersionTable checks if versions table exists and, if not, creates it.
+// Note that this function locks the database, which deviates from the usual
+// convention of "caller locks" in the CockroachDb type.
+func (c *CockroachDb) ensureVersionTable() (err error) {
+	if err = c.Lock(); err != nil {
+		return err
+	}
+
+	defer func() {
+		if e := c.Unlock(); e != nil {
+			if err == nil {
+				err = e
+			} else {
+				err = multierror.Append(err, e)
+			}
+		}
+	}()
+
 	// check if migration table exists
 	var count int
 	query := `SELECT COUNT(1) FROM information_schema.tables WHERE table_name = $1 AND table_schema = (SELECT current_schema()) LIMIT 1`

database/cockroachdb/cockroachdb_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"github.com/golang-migrate/migrate/v4"
 	"strings"
 	"testing"
 )
@@ -18,6 +19,7 @@ import (
 import (
 	dt "github.com/golang-migrate/migrate/v4/database/testing"
 	"github.com/golang-migrate/migrate/v4/dktesting"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
 )
 
 const defaultPort = 26257
@@ -92,6 +94,30 @@ func Test(t *testing.T) {
 	})
 }
 
+func TestMigrate(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, ci dktest.ContainerInfo) {
+		createDB(t, ci)
+
+		ip, port, err := ci.Port(26257)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addr := fmt.Sprintf("cockroach://root@%v:%v/migrate?sslmode=disable", ip, port)
+		c := &CockroachDb{}
+		d, err := c.Open(addr)
+		if err != nil {
+			t.Fatalf("%v", err)
+		}
+
+		m, err := migrate.NewWithDatabaseInstance("file://./examples/migrations", "migrate", d)
+		if err != nil {
+			t.Fatalf("%v", err)
+		}
+		dt.TestMigrate(t, m, []byte("SELECT 1"))
+	})
+}
+
 func TestMultiStatement(t *testing.T) {
 	dktesting.ParallelTest(t, specs, func(t *testing.T, ci dktest.ContainerInfo) {
 		createDB(t, ci)

database/driver.go

@@ -74,6 +74,8 @@ type Driver interface {
 	Version() (version int, dirty bool, err error)
 
 	// Drop deletes everything in the database.
+	// Note that this is a breaking action, a new call to Open() is necessary to
+	// ensure subsequent calls work as expected.
 	Drop() error
 }
 

database/mongodb/mongodb.go

@@ -59,6 +59,7 @@ func WithInstance(instance *mongo.Client, config *Config) (database.Driver, erro
 		db:     instance.Database(config.DatabaseName),
 		config: config,
 	}
+
 	return mc, nil
 }
 
@@ -77,9 +78,6 @@ func (m *Mongo) Open(dsn string) (database.Driver, error) {
 		return nil, err
 	}
 	migrationsCollection := purl.Query().Get("x-migrations-collection")
-	if len(migrationsCollection) == 0 {
-		migrationsCollection = DefaultMigrationsCollection
-	}
 
 	transactionMode, _ := strconv.ParseBool(purl.Query().Get("x-transaction-mode"))
 

database/mongodb/mongodb_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"github.com/golang-migrate/migrate/v4"
 	"io"
 	"os"
 	"strconv"
@@ -20,6 +21,7 @@ import (
 import (
 	dt "github.com/golang-migrate/migrate/v4/database/testing"
 	"github.com/golang-migrate/migrate/v4/dktesting"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
 )
 
 var (
@@ -83,6 +85,28 @@ func Test(t *testing.T) {
 	})
 }
 
+func TestMigrate(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.FirstPort()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addr := mongoConnectionString(ip, port)
+		p := &Mongo{}
+		d, err := p.Open(addr)
+		if err != nil {
+			t.Fatalf("%v", err)
+		}
+		defer d.Close()
+		m, err := migrate.NewWithDatabaseInstance("file://./examples/migrations", "", d)
+		if err != nil {
+			t.Fatalf("%v", err)
+		}
+		dt.TestMigrate(t, m, []byte(`[{"insert":"hello","documents":[{"wild":"world"}]}]`))
+	})
+}
+
 func TestWithAuth(t *testing.T) {
 	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
 		ip, port, err := c.FirstPort()

database/mysql/examples/migrations/1_init.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS test;

database/mysql/examples/migrations/1_init.up.sql

@@ -0,0 +1,3 @@
+CREATE TABLE IF NOT EXISTS test (
+  firstname VARCHAR(16)
+);

database/mysql/mysql.go

@@ -17,6 +17,7 @@ import (
 
 import (
 	"github.com/go-sql-driver/mysql"
+	"github.com/hashicorp/go-multierror"
 )
 
 import (
@@ -127,9 +128,6 @@ func (m *Mysql) Open(url string) (database.Driver, error) {
 	purl.RawQuery = q.Encode()
 
 	migrationsTable := purl.Query().Get("x-migrations-table")
-	if len(migrationsTable) == 0 {
-		migrationsTable = DefaultMigrationsTable
-	}
 
 	// use custom TLS?
 	ctls := purl.Query().Get("tls")
@@ -342,15 +340,29 @@ func (m *Mysql) Drop() error {
 				return &database.Error{OrigErr: err, Query: []byte(query)}
 			}
 		}
-		if err := m.ensureVersionTable(); err != nil {
-			return err
-		}
 	}
 
 	return nil
 }
 
-func (m *Mysql) ensureVersionTable() error {
+// ensureVersionTable checks if versions table exists and, if not, creates it.
+// Note that this function locks the database, which deviates from the usual
+// convention of "caller locks" in the Mysql type.
+func (m *Mysql) ensureVersionTable() (err error) {
+	if err = m.Lock(); err != nil {
+		return err
+	}
+
+	defer func() {
+		if e := m.Unlock(); e != nil {
+			if err == nil {
+				err = e
+			} else {
+				err = multierror.Append(err, e)
+			}
+		}
+	}()
+
 	// check if migration table exists
 	var result string
 	query := `SHOW TABLES LIKE "` + m.config.MigrationsTable + `"`