Merge pull request #10 from dinedal/feature/add_cassandra_driver

Allows for migrations to take place against cassandra

.travis.yml

@@ -8,7 +8,12 @@ go:
 addons:
   postgresql: "9.3"
 
+services:
+  - cassandra
+
 before_script:
+  - >
+      /usr/local/cassandra/bin/cqlsh -e "CREATE KEYSPACE migratetest WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor' : 1};"
   - psql -c 'create database migratetest;' -U postgres
 
 script: go test -p 1 ./...

driver/cassandra/cassandra.go

@@ -0,0 +1,103 @@
+// Package cassandra implements the Driver interface.
+package cassandra
+
+import (
+	"net/url"
+	"time"
+
+	"github.com/gocql/gocql"
+	"github.com/mattes/migrate/file"
+	"github.com/mattes/migrate/migrate/direction"
+)
+
+type Driver struct {
+	session *gocql.Session
+}
+
+const tableName = "schema_migrations"
+const versionRow = 1
+
+// Cassandra Driver URL format:
+// cassandra://host:port/keyspace
+//
+// Example:
+// cassandra://localhost/SpaceOfKeys
+func (driver *Driver) Initialize(rawurl string) error {
+	u, err := url.Parse(rawurl)
+
+	cluster := gocql.NewCluster(u.Host)
+	cluster.Keyspace = u.Path[1:len(u.Path)]
+	cluster.Consistency = gocql.All
+	cluster.Timeout = 1 * time.Minute
+
+	driver.session, err = cluster.CreateSession()
+
+	if err != nil {
+		return err
+	}
+
+	if err := driver.ensureVersionTableExists(); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (driver *Driver) Close() error {
+	driver.session.Close()
+	return nil
+}
+
+func (driver *Driver) ensureVersionTableExists() error {
+	err := driver.session.Query("CREATE TABLE IF NOT EXISTS " + tableName + " (version counter, versionRow bigint primary key);").Exec()
+	if err != nil {
+		return err
+	}
+
+	_, err = driver.Version()
+	if err != nil {
+		driver.session.Query("UPDATE "+tableName+" SET version = version + 1 where versionRow = ?", versionRow).Exec()
+	}
+
+	return nil
+}
+
+func (driver *Driver) FilenameExtension() string {
+	return "cql"
+}
+
+func (driver *Driver) Migrate(f file.File, pipe chan interface{}) {
+	defer close(pipe)
+	pipe <- f
+
+	if f.Direction == direction.Up {
+		err := driver.session.Query("UPDATE "+tableName+" SET version = version + 1 where versionRow = ?", versionRow).Exec()
+		if err != nil {
+			pipe <- err
+			return
+		}
+	} else if f.Direction == direction.Down {
+		err := driver.session.Query("UPDATE "+tableName+" SET version = version - 1 where versionRow = ?", versionRow).Exec()
+		if err != nil {
+			pipe <- err
+			return
+		}
+	}
+
+	if err := f.ReadContent(); err != nil {
+		pipe <- err
+		return
+	}
+
+	err := driver.session.Query(string(f.Content)).Exec()
+
+	if err != nil {
+		pipe <- err
+		return
+	}
+}
+
+func (driver *Driver) Version() (uint64, error) {
+	var version int64
+	err := driver.session.Query("SELECT version FROM "+tableName+" WHERE versionRow = ?", versionRow).Scan(&version)
+	return uint64(version) - 1, err
+}

driver/cassandra/cassandra_test.go

@@ -0,0 +1,109 @@
+package cassandra
+
+import (
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/gocql/gocql"
+	"github.com/mattes/migrate/file"
+	"github.com/mattes/migrate/migrate/direction"
+	pipep "github.com/mattes/migrate/pipe"
+)
+
+func TestMigrate(t *testing.T) {
+	var session *gocql.Session
+	driverUrl := "cassandra://localhost/migratetest"
+
+	// prepare a clean test database
+	u, err := url.Parse(driverUrl)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cluster := gocql.NewCluster(u.Host)
+	cluster.Keyspace = u.Path[1:len(u.Path)]
+	cluster.Consistency = gocql.All
+	cluster.Timeout = 1 * time.Minute
+
+	session, err = cluster.CreateSession()
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := session.Query(`DROP TABLE IF EXISTS yolo`).Exec(); err != nil {
+		t.Fatal(err)
+	}
+	if err := session.Query(`DROP TABLE IF EXISTS ` + tableName).Exec(); err != nil {
+		t.Fatal(err)
+	}
+
+	d := &Driver{}
+	if err := d.Initialize(driverUrl); err != nil {
+		t.Fatal(err)
+	}
+
+	files := []file.File{
+		{
+			Path:      "/foobar",
+			FileName:  "001_foobar.up.sql",
+			Version:   1,
+			Name:      "foobar",
+			Direction: direction.Up,
+			Content: []byte(`
+                CREATE TABLE yolo (
+                    id varint primary key
+                );
+            `),
+		},
+		{
+			Path:      "/foobar",
+			FileName:  "002_foobar.down.sql",
+			Version:   1,
+			Name:      "foobar",
+			Direction: direction.Down,
+			Content: []byte(`
+                DROP TABLE yolo;
+            `),
+		},
+		{
+			Path:      "/foobar",
+			FileName:  "002_foobar.up.sql",
+			Version:   2,
+			Name:      "foobar",
+			Direction: direction.Up,
+			Content: []byte(`
+                CREATE TABLE error (
+                    id THIS WILL CAUSE AN ERROR
+                )
+            `),
+		},
+	}
+
+	pipe := pipep.New()
+	go d.Migrate(files[0], pipe)
+	errs := pipep.ReadErrors(pipe)
+	if len(errs) > 0 {
+		t.Fatal(errs)
+	}
+
+	pipe = pipep.New()
+	go d.Migrate(files[1], pipe)
+	errs = pipep.ReadErrors(pipe)
+	if len(errs) > 0 {
+		t.Fatal(errs)
+	}
+
+	pipe = pipep.New()
+	go d.Migrate(files[2], pipe)
+	errs = pipep.ReadErrors(pipe)
+	if len(errs) == 0 {
+		t.Error("Expected test case to fail")
+	}
+
+	if err := d.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+}

driver/driver.go

@@ -4,10 +4,12 @@ package driver
 import (
 	"errors"
 	"fmt"
+	neturl "net/url" // alias to allow `url string` func signature in New
+
 	"github.com/mattes/migrate/driver/bash"
+	"github.com/mattes/migrate/driver/cassandra"
 	"github.com/mattes/migrate/driver/postgres"
 	"github.com/mattes/migrate/file"
-	neturl "net/url" // alias to allow `url string` func signature in New
 )
 
 // Driver is the interface type that needs to implemented by all drivers.
@@ -58,6 +60,13 @@ func New(url string) (Driver, error) {
 			return nil, err
 		}
 		return d, nil
+	case "cassandra":
+		d := &cassandra.Driver{}
+		verifyFilenameExtension("cassanda", d)
+		if err := d.Initialize(url); err != nil {
+			return nil, err
+		}
+		return d, nil
 	default:
 		return nil, errors.New(fmt.Sprintf("Driver '%s' not found.", u.Scheme))
 	}

migrate/migrate.go

@@ -4,16 +4,17 @@ package migrate
 
 import (
 	"fmt"
-	"github.com/mattes/migrate/driver"
-	"github.com/mattes/migrate/file"
-	"github.com/mattes/migrate/migrate/direction"
-	pipep "github.com/mattes/migrate/pipe"
 	"io/ioutil"
 	"os"
 	"os/signal"
 	"path"
 	"strconv"
 	"strings"
+
+	"github.com/mattes/migrate/driver"
+	"github.com/mattes/migrate/file"
+	"github.com/mattes/migrate/migrate/direction"
+	pipep "github.com/mattes/migrate/pipe"
 )
 
 // Up applies all available migrations