Adds vertica driver support (#393)

.github/workflows/e2e.yml

@@ -38,3 +38,16 @@ jobs:
           go-version: "1.18"
       - name: Run clickhouse test
         run: make test-clickhouse
+  test-vertica:
+    name: Run vertica tests
+    timeout-minutes: 10
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+      - name: Install Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: "1.18"
+      - name: Run vertica test
+        run: make test-vertica

Makefile

@@ -22,6 +22,9 @@ test-e2e-mysql:
 test-clickhouse:
 	go test -timeout=10m -count=1 -race -v ./tests/clickhouse -test.short
 
+test-vertica:
+	go test -count=1 -v ./tests/vertica
+
 docker-cleanup:
 	docker stop -t=0 $$(docker ps --filter="label=goose_test" -aq)
 

README.md

@@ -61,6 +61,7 @@ Drivers:
     redshift
     tidb
     clickhouse
+    vertica
 
 Examples:
     goose sqlite3 ./foo.db status
@@ -74,6 +75,8 @@ Examples:
     goose redshift "postgres://user:password@qwerty.us-east-1.redshift.amazonaws.com:5439/db" status
     goose tidb "user:password@/dbname?parseTime=true" status
     goose mssql "sqlserver://user:password@dbname:1433?database=master" status
+    goose clickhouse "tcp://127.0.0.1:9000" status
+    goose vertica "vertica://user:password@localhost:5433/dbname?connection_load_balance=1" status
 
 Options:
 

cmd/goose/driver_vertica.go

@@ -0,0 +1,8 @@
+//go:build !no_vertica
+// +build !no_vertica
+
+package main
+
+import (
+	_ "github.com/vertica/vertica-sql-go"
+)

cmd/goose/main.go

@@ -197,6 +197,7 @@ Drivers:
     redshift
     tidb
     clickhouse
+    vertica
 
 Examples:
     goose sqlite3 ./foo.db status
@@ -211,6 +212,7 @@ Examples:
     goose tidb "user:password@/dbname?parseTime=true" status
     goose mssql "sqlserver://user:password@dbname:1433?database=master" status
     goose clickhouse "tcp://127.0.0.1:9000" status
+    goose vertica "vertica://user:password@localhost:5433/dbname?connection_load_balance=1" status
 
     GOOSE_DRIVER=sqlite3 GOOSE_DBSTRING=./foo.db goose status
     GOOSE_DRIVER=sqlite3 GOOSE_DBSTRING=./foo.db goose create init sql

db.go

@@ -22,7 +22,7 @@ func OpenDBWithDriver(driver string, dbstring string) (*sql.DB, error) {
 	}
 
 	switch driver {
-	case "postgres", "pgx", "sqlite3", "sqlite", "mysql", "sqlserver", "clickhouse":
+	case "postgres", "pgx", "sqlite3", "sqlite", "mysql", "sqlserver", "clickhouse", "vertica":
 		return sql.Open(driver, dbstring)
 	default:
 		return nil, fmt.Errorf("unsupported driver %s", driver)

dialect.go

@@ -39,6 +39,8 @@ func SetDialect(d string) error {
 		dialect = &TiDBDialect{}
 	case "clickhouse":
 		dialect = &ClickHouseDialect{}
+	case "vertica":
+		dialect = &VerticaDialect{}
 	default:
 		return fmt.Errorf("%q: unknown dialect", d)
 	}
@@ -322,3 +324,41 @@ func (m ClickHouseDialect) migrationSQL() string {
 func (m ClickHouseDialect) deleteVersionSQL() string {
 	return fmt.Sprintf("ALTER TABLE %s DELETE WHERE version_id = $1", TableName())
 }
+
+////////////////////////////
+// Vertica
+////////////////////////////
+
+// VerticaDialect struct.
+type VerticaDialect struct{}
+
+func (v VerticaDialect) createVersionTableSQL() string {
+	return fmt.Sprintf(`CREATE TABLE %s (
+                id identity(1,1) NOT NULL,
+                version_id bigint NOT NULL,
+                is_applied boolean NOT NULL,
+                tstamp timestamp NULL default now(),
+                PRIMARY KEY(id)
+            );`, TableName())
+}
+
+func (v VerticaDialect) insertVersionSQL() string {
+	return fmt.Sprintf("INSERT INTO %s (version_id, is_applied) VALUES (?, ?);", TableName())
+}
+
+func (v VerticaDialect) dbVersionQuery(db *sql.DB) (*sql.Rows, error) {
+	rows, err := db.Query(fmt.Sprintf("SELECT version_id, is_applied from %s ORDER BY id DESC", TableName()))
+	if err != nil {
+		return nil, err
+	}
+
+	return rows, err
+}
+
+func (m VerticaDialect) migrationSQL() string {
+	return fmt.Sprintf("SELECT tstamp, is_applied FROM %s WHERE version_id=? ORDER BY tstamp DESC LIMIT 1", TableName())
+}
+
+func (v VerticaDialect) deleteVersionSQL() string {
+	return fmt.Sprintf("DELETE FROM %s WHERE version_id=?;", TableName())
+}

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/jackc/pgx/v4 v4.17.2
 	github.com/ory/dockertest/v3 v3.9.1
+	github.com/vertica/vertica-sql-go v1.2.2
 	github.com/ziutek/mymysql v1.5.4
 	modernc.org/sqlite v1.19.2
 )

go.sum

@@ -254,6 +254,8 @@ github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG
 github.com/tklauser/go-sysconf v0.3.10/go.mod h1:C8XykCvCb+Gn0oNCWPIlcb0RuglQTYaQ2hGm7jmxEFk=
 github.com/tklauser/numcpus v0.4.0/go.mod h1:1+UI3pD8NW14VMwdgJNJ1ESk2UnwhAnz5hMwiKKqXCQ=
 github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
+github.com/vertica/vertica-sql-go v1.2.2 h1:woI501lizEoqONmO5B7a5DNsLTQTsT0HnD1JM7SiNhk=
+github.com/vertica/vertica-sql-go v1.2.2/go.mod h1:fGr44VWdEvL+f+Qt5LkKLOT7GoxaWdoUCnPBU9h6t04=
 github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE=
 github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=

internal/testdb/testdb.go

@@ -16,3 +16,8 @@ func NewPostgres(options ...OptionsFunc) (db *sql.DB, cleanup func(), err error)
 func NewMariaDB(options ...OptionsFunc) (db *sql.DB, cleanup func(), err error) {
 	return newMariaDB(options...)
 }
+
+// NewVertica starts a Vertica docker container. Returns a db connection and a docker cleanup function.
+func NewVertica(options ...OptionsFunc) (db *sql.DB, cleanup func(), err error) {
+	return newVertica(options...)
+}

internal/testdb/vertica.go

@@ -0,0 +1,96 @@
+package testdb
+
+import (
+	"database/sql"
+	"fmt"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/ory/dockertest/v3"
+	"github.com/ory/dockertest/v3/docker"
+	_ "github.com/vertica/vertica-sql-go"
+)
+
+const (
+	// https://hub.docker.com/r/vertica/vertica-ce
+	VERTICA_IMAGE   = "vertica/vertica-ce"
+	VERTICA_VERSION = "12.0.0-0"
+	VERTICA_DB      = "testdb"
+)
+
+func newVertica(opts ...OptionsFunc) (*sql.DB, func(), error) {
+	option := &options{}
+	for _, f := range opts {
+		f(option)
+	}
+	// Uses a sensible default on windows (tcp/http) and linux/osx (socket).
+	pool, err := dockertest.NewPool("")
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to connect to docker: %v", err)
+	}
+	options := &dockertest.RunOptions{
+		Repository: VERTICA_IMAGE,
+		Tag:        VERTICA_VERSION,
+		Env: []string{
+			"VERTICA_DB_NAME=" + VERTICA_DB,
+			"VMART_ETL_SCRIPT=", // Don't install VMART data inside container.
+		},
+		Labels:       map[string]string{"goose_test": "1"},
+		PortBindings: make(map[docker.Port][]docker.PortBinding),
+		// Prevent package installation for faster container startup.
+		Mounts: []string{"/tmp/empty:/opt/vertica/packages"},
+	}
+	if option.bindPort > 0 {
+		options.PortBindings[docker.Port("5433/tcp")] = []docker.PortBinding{
+			{HostPort: strconv.Itoa(option.bindPort)},
+		}
+	}
+	container, err := pool.RunWithOptions(
+		options,
+		func(config *docker.HostConfig) {
+			// Set AutoRemove to true so that stopped container goes away by itself.
+			config.AutoRemove = true
+			config.RestartPolicy = docker.RestartPolicy{Name: "no"}
+		},
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create docker container: %v", err)
+	}
+	cleanup := func() {
+		if option.debug {
+			// User must manually delete the Docker container.
+			return
+		}
+		if err := pool.Purge(container); err != nil {
+			log.Printf("failed to purge resource: %v", err)
+		}
+	}
+	verticaInfo := fmt.Sprintf("vertica://%s:%s@%s:%s/%s",
+		"dbadmin",
+		"",
+		"localhost",
+		container.GetPort("5433/tcp"), // Fetch port dynamically assigned to container
+		VERTICA_DB,
+	)
+
+	var db *sql.DB
+	// Give vertica a head start since the container takes a little bit to start up.
+	time.Sleep(time.Second * 15)
+
+	// Exponential backoff-retry, because the application in the container
+	// might not be ready to accept connections yet.
+	if err := pool.Retry(
+		func() error {
+			var err error
+			db, err = sql.Open("vertica", verticaInfo)
+			if err != nil {
+				return err
+			}
+			return db.Ping()
+		},
+	); err != nil {
+		return nil, cleanup, fmt.Errorf("could not connect to docker database: %v", err)
+	}
+	return db, cleanup, nil
+}

tests/vertica/testdata/migrations/00001_a.sql

@@ -0,0 +1,9 @@
+-- +goose Up
+-- +goose StatementBegin
+CREATE SCHEMA IF NOT EXISTS testing;
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DROP SCHEMA IF EXISTS testing;
+-- +goose StatementEnd

tests/vertica/testdata/migrations/00002_b.sql

@@ -0,0 +1,44 @@
+-- +goose Up
+-- +goose StatementBegin
+CREATE TABLE IF NOT EXISTS testing.dim_test_scd
+(
+    test_key    BIGINT  NOT NULL,
+    test_id     UUID    NOT NULL,
+    valid_from  DATE    NOT NULL,
+    valid_to    DATE    NOT NULL,
+    is_current  BOOLEAN NOT NULL
+        DEFAULT (valid_to = '9999/12/31'),
+    external_id VARCHAR(100)
+) UNSEGMENTED ALL NODES;
+-- +goose StatementEnd
+-- +goose StatementBegin
+CREATE PROJECTION IF NOT EXISTS testing.dim_test_scd_proj_is_current AS
+    SELECT test_key,
+           test_id,
+           valid_from,
+           valid_to,
+           is_current,
+           external_id
+    FROM testing.dim_test_scd
+    ORDER BY is_current, test_id
+    SEGMENTED BY HASH(test_key) ALL NODES;
+-- +goose StatementEnd
+-- +goose StatementBegin
+CREATE OR REPLACE VIEW testing.Test AS
+SELECT test_key,
+       test_id,
+       external_id
+FROM testing.dim_test_scd
+WHERE is_current = true;
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DROP VIEW IF EXISTS testing.Test;
+-- +goose StatementEnd
+-- +goose StatementBegin
+DROP PROJECTION IF EXISTS testing.dim_test_scd_proj_is_current;
+-- +goose StatementEnd
+-- +goose StatementBegin
+DROP TABLE IF EXISTS testing.dim_test_scd;
+-- +goose StatementEnd

tests/vertica/testdata/migrations/00003_c.sql

@@ -0,0 +1,15 @@
+-- +goose Up
+-- +goose StatementBegin
+INSERT INTO testing.dim_test_scd VALUES (1, '575a0dd4-bd97-44ac-aae0-987090181da8', '2021-10-02', '2021-10-03', false, '123');
+-- +goose StatementEnd
+-- +goose StatementBegin
+INSERT INTO testing.dim_test_scd VALUES (2, '575a0dd4-bd97-44ac-aae0-987090181da8', '2021-10-03', '2021-10-04', false, '456');
+-- +goose StatementEnd
+-- +goose StatementBegin
+INSERT INTO testing.dim_test_scd VALUES (3, '575a0dd4-bd97-44ac-aae0-987090181da8', '2021-10-04', '9999-12-31', true, '789');
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DELETE FROM testing.dim_test_scd where test_id = '575a0dd4-bd97-44ac-aae0-987090181da8';
+-- +goose StatementEnd

tests/vertica/vertica_test.go

@@ -0,0 +1,115 @@
+package vertica_test
+
+import (
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/pressly/goose/v3"
+	"github.com/pressly/goose/v3/internal/check"
+	"github.com/pressly/goose/v3/internal/testdb"
+)
+
+/*
+This test applies all up migrations, asserts we have all the entries in
+the versions table, applies all down migration and asserts we have zero
+migrations applied.
+
+Limitations:
+1) Only one instance of Vertica can be running on a host at any time.
+*/
+func TestVerticaUpDownAll(t *testing.T) {
+	t.Parallel()
+
+	migrationDir := filepath.Join("testdata", "migrations")
+	db, cleanup, err := testdb.NewVertica()
+	check.NoError(t, err)
+	t.Cleanup(cleanup)
+
+	goose.SetDialect("vertica")
+
+	goose.SetTableName("goose_db_version")
+
+	migrations, err := goose.CollectMigrations(migrationDir, 0, goose.MaxVersion)
+	check.NoError(t, err)
+
+	currentVersion, err := goose.GetDBVersion(db)
+	check.NoError(t, err)
+	check.Number(t, currentVersion, 0)
+
+	err = goose.Up(db, migrationDir)
+	check.NoError(t, err)
+	currentVersion, err = goose.GetDBVersion(db)
+	check.NoError(t, err)
+	check.Number(t, currentVersion, len(migrations))
+
+	type result struct {
+		TestKey    int64     `db:"test_key"`
+		TestID     string    `db:"test_id"`
+		ValidFrom  time.Time `db:"valid_from"`
+		ValidTo    time.Time `db:"valid_to"`
+		IsCurrent  bool      `db:"is_current"`
+		ExternalID string    `db:"external_id"`
+	}
+	rows, err := db.Query(`SELECT * FROM testing.dim_test_scd ORDER BY test_key`)
+	check.NoError(t, err)
+	var results []result
+	for rows.Next() {
+		var r result
+		err = rows.Scan(&r.TestKey, &r.TestID, &r.ValidFrom, &r.ValidTo, &r.IsCurrent, &r.ExternalID)
+		check.NoError(t, err)
+		results = append(results, r)
+	}
+	check.Number(t, len(results), 3)
+	check.NoError(t, rows.Close())
+	check.NoError(t, rows.Err())
+
+	parseTime := func(t *testing.T, s string) time.Time {
+		t.Helper()
+		tm, err := time.Parse("2006-01-02", s)
+		check.NoError(t, err)
+		return tm
+	}
+	want := []result{
+		{
+			TestKey:    1,
+			TestID:     "575a0dd4-bd97-44ac-aae0-987090181da8",
+			ValidFrom:  parseTime(t, "2021-10-02"),
+			ValidTo:    parseTime(t, "2021-10-03"),
+			IsCurrent:  false,
+			ExternalID: "123",
+		},
+		{
+			TestKey:    2,
+			TestID:     "575a0dd4-bd97-44ac-aae0-987090181da8",
+			ValidFrom:  parseTime(t, "2021-10-03"),
+			ValidTo:    parseTime(t, "2021-10-04"),
+			IsCurrent:  false,
+			ExternalID: "456",
+		},
+		{
+			TestKey:    3,
+			TestID:     "575a0dd4-bd97-44ac-aae0-987090181da8",
+			ValidFrom:  parseTime(t, "2021-10-04"),
+			ValidTo:    parseTime(t, "9999-12-31"),
+			IsCurrent:  true,
+			ExternalID: "789",
+		},
+	}
+	for i, result := range results {
+		check.Equal(t, result.TestKey, want[i].TestKey)
+		check.Equal(t, result.TestID, want[i].TestID)
+		check.Equal(t, result.ValidFrom, want[i].ValidFrom)
+		check.Equal(t, result.ValidTo, want[i].ValidTo)
+		check.Equal(t, result.IsCurrent, want[i].IsCurrent)
+		check.Equal(t, result.ExternalID, want[i].ExternalID)
+	}
+
+	err = goose.DownTo(db, migrationDir, 0)
+	check.NoError(t, err)
+	check.NoError(t, err)
+
+	currentVersion, err = goose.GetDBVersion(db)
+	check.NoError(t, err)
+	check.Number(t, currentVersion, 0)
+}