Add DuckDB support with official driver and modern stdlib

Makefile

@@ -1,6 +1,6 @@
 SOURCE ?= file go_bindata github github_ee bitbucket aws_s3 google_cloud_storage godoc_vfs gitlab
 DATABASE ?= postgres mysql redshift cassandra spanner cockroachdb yugabytedb clickhouse mongodb sqlserver firebird neo4j pgx pgx5 rqlite
-DATABASE_TEST ?= $(DATABASE) sqlite sqlite3 sqlcipher
+DATABASE_TEST ?= $(DATABASE) sqlite sqlite3 sqlcipher duckdb
 VERSION ?= $(shell git describe --tags 2>/dev/null | cut -c 2-)
 TEST_FLAGS ?=
 REPO_OWNER ?= $(shell cd .. && basename "$$(pwd)")

README.md

@@ -41,6 +41,7 @@ Database drivers run migrations. [Add a new database?](database/driver.go)
 * [CockroachDB](database/cockroachdb)
 * [YugabyteDB](database/yugabytedb)
 * [ClickHouse](database/clickhouse)
+* [DuckDB](database/duckdb)
 * [Firebird](database/firebird)
 * [MS SQL Server](database/sqlserver)
 * [rqlite](database/rqlite)

database/duckdb/README.md

@@ -0,0 +1,14 @@
+# DuckDB
+
+`duckdb://path/to/database.db`
+
+| URL Query  | Description |
+|------------|-------------|
+| `x-migrations-table` | Name of the migrations table (default: `schema_migrations`) |
+| `x-no-tx-wrap` | Disable automatic transaction wrapping for migrations (default: `false`) |
+
+## Notes
+
+* DuckDB is an in-process SQL OLAP database management system.
+* Uses the official DuckDB Go driver: [github.com/duckdb/duckdb-go/v2](https://github.com/duckdb/duckdb-go)
+* Supports in-memory databases using `:memory:` as the path.

database/duckdb/duckdb.go

@@ -0,0 +1,260 @@
+package duckdb
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"io"
+	nurl "net/url"
+	"strconv"
+	"strings"
+	"sync/atomic"
+
+	"github.com/golang-migrate/migrate/v4"
+	"github.com/golang-migrate/migrate/v4/database"
+
+	_ "github.com/duckdb/duckdb-go/v2"
+)
+
+func init() {
+	database.Register("duckdb", &DuckDB{})
+}
+
+var DefaultMigrationsTable = "schema_migrations"
+
+var (
+	ErrNilConfig = errors.New("no config")
+)
+
+type Config struct {
+	MigrationsTable string
+	NoTxWrap        bool
+}
+
+type DuckDB struct {
+	db       *sql.DB
+	isLocked atomic.Bool
+	config   *Config
+}
+
+func (d *DuckDB) Open(url string) (database.Driver, error) {
+	purl, err := nurl.Parse(url)
+	if err != nil {
+		return nil, fmt.Errorf("parsing url: %w", err)
+	}
+	dbfile := strings.Replace(migrate.FilterCustomQuery(purl).String(), "duckdb://", "", 1)
+	db, err := sql.Open("duckdb", dbfile)
+	if err != nil {
+		return nil, fmt.Errorf("opening '%s': %w", dbfile, err)
+	}
+
+	qv := purl.Query()
+	migrationsTable := qv.Get("x-migrations-table")
+	if len(migrationsTable) == 0 {
+		migrationsTable = DefaultMigrationsTable
+	}
+
+	noTxWrap := false
+	if v := qv.Get("x-no-tx-wrap"); v != "" {
+		noTxWrap, err = strconv.ParseBool(v)
+		if err != nil {
+			return nil, fmt.Errorf("x-no-tx-wrap: %s", err)
+		}
+	}
+
+	if err := db.Ping(); err != nil {
+		return nil, fmt.Errorf("pinging: %w", err)
+	}
+	cfg := &Config{
+		MigrationsTable: migrationsTable,
+		NoTxWrap:        noTxWrap,
+	}
+	return WithInstance(db, cfg)
+}
+
+func (d *DuckDB) Close() error {
+	return d.db.Close()
+}
+
+func (d *DuckDB) Lock() error {
+	if !d.isLocked.CompareAndSwap(false, true) {
+		return database.ErrLocked
+	}
+	return nil
+}
+
+func (d *DuckDB) Unlock() error {
+	if !d.isLocked.CompareAndSwap(true, false) {
+		return database.ErrNotLocked
+	}
+	return nil
+}
+
+func (d *DuckDB) Drop() error {
+	tablesQuery := `SELECT schema_name, table_name FROM duckdb_tables()`
+	tables, err := d.db.Query(tablesQuery)
+	if err != nil {
+		return &database.Error{OrigErr: err, Query: []byte(tablesQuery)}
+	}
+	defer func() {
+		if errClose := tables.Close(); errClose != nil {
+			err = errors.Join(err, errClose)
+		}
+	}()
+
+	tableNames := []string{}
+	for tables.Next() {
+		var (
+			schemaName string
+			tableName  string
+		)
+
+		if err := tables.Scan(&schemaName, &tableName); err != nil {
+			return &database.Error{OrigErr: err, Err: "scanning schema and table name"}
+		}
+
+		if len(schemaName) > 0 {
+			tableNames = append(tableNames, fmt.Sprintf("%s.%s", schemaName, tableName))
+		} else {
+			tableNames = append(tableNames, tableName)
+		}
+	}
+	if err := tables.Err(); err != nil {
+		return &database.Error{OrigErr: err, Query: []byte(tablesQuery), Err: "err in rows after scanning"}
+	}
+
+	for _, t := range tableNames {
+		dropQuery := fmt.Sprintf("DROP TABLE %s", t)
+		if _, err := d.db.Exec(dropQuery); err != nil {
+			return &database.Error{OrigErr: err, Query: []byte(dropQuery)}
+		}
+	}
+
+	return nil
+
+}
+
+func (d *DuckDB) SetVersion(version int, dirty bool) error {
+	tx, err := d.db.Begin()
+	if err != nil {
+		return &database.Error{OrigErr: err, Err: "transaction start failed"}
+	}
+
+	query := "DELETE FROM " + d.config.MigrationsTable
+	if _, err := tx.Exec(query); err != nil {
+		return &database.Error{OrigErr: err, Query: []byte(query)}
+	}
+
+	// Also re-write the schema version for nil dirty versions to prevent
+	// empty schema version for failed down migration on the first migration
+	// See: https://github.com/golang-migrate/migrate/issues/330
+	//
+	// NOTE: Copied from sqlite implementation, unsure if this is necessary for
+	// duckdb
+	if version >= 0 || (version == database.NilVersion && dirty) {
+		query := fmt.Sprintf(`INSERT INTO %s (version, dirty) VALUES (?, ?)`, d.config.MigrationsTable)
+		if _, err := tx.Exec(query, version, dirty); err != nil {
+			if errRollback := tx.Rollback(); errRollback != nil {
+				err = errors.Join(err, errRollback)
+			}
+			return &database.Error{OrigErr: err, Query: []byte(query)}
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		return &database.Error{OrigErr: err, Err: "transaction commit failed"}
+	}
+
+	return nil
+}
+
+func (m *DuckDB) Version() (version int, dirty bool, err error) {
+	query := "SELECT version, dirty FROM " + m.config.MigrationsTable + " LIMIT 1"
+	err = m.db.QueryRow(query).Scan(&version, &dirty)
+	if err != nil {
+		return database.NilVersion, false, nil
+	}
+	return version, dirty, nil
+}
+
+func (d *DuckDB) Run(migration io.Reader) error {
+	migr, err := io.ReadAll(migration)
+	if err != nil {
+		return fmt.Errorf("reading migration: %w", err)
+	}
+	query := string(migr[:])
+
+	if d.config.NoTxWrap {
+		if _, err := d.db.Exec(query); err != nil {
+			return &database.Error{OrigErr: err, Query: []byte(query)}
+		}
+		return nil
+	}
+
+	tx, err := d.db.Begin()
+	if err != nil {
+		return &database.Error{OrigErr: err, Err: "transaction start failed"}
+	}
+	if _, err := tx.Exec(query); err != nil {
+		if errRollback := tx.Rollback(); errRollback != nil {
+			err = errors.Join(err, errRollback)
+		}
+		return &database.Error{OrigErr: err, Query: []byte(query)}
+	}
+	if err := tx.Commit(); err != nil {
+		return &database.Error{OrigErr: err, Err: "transaction commit failed"}
+	}
+	return nil
+}
+
+// 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 Sqlite type.
+func (d *DuckDB) ensureVersionTable() (err error) {
+	if err = d.Lock(); err != nil {
+		return err
+	}
+
+	defer func() {
+		if e := d.Unlock(); e != nil {
+			if err == nil {
+				err = e
+			} else {
+				err = errors.Join(err, e)
+			}
+		}
+	}()
+
+	query := fmt.Sprintf(`
+	CREATE TABLE IF NOT EXISTS %s (version BIGINT, dirty BOOLEAN);
+	CREATE UNIQUE INDEX IF NOT EXISTS version_unique ON %s (version);
+`, d.config.MigrationsTable, d.config.MigrationsTable)
+
+	if _, err := d.db.Exec(query); err != nil {
+		return fmt.Errorf("creating version table via '%s': %w", query, err)
+	}
+	return nil
+}
+
+func WithInstance(instance *sql.DB, config *Config) (database.Driver, error) {
+	if config == nil {
+		return nil, ErrNilConfig
+	}
+
+	if err := instance.Ping(); err != nil {
+		return nil, err
+	}
+
+	if len(config.MigrationsTable) == 0 {
+		config.MigrationsTable = DefaultMigrationsTable
+	}
+
+	mx := &DuckDB{
+		db:     instance,
+		config: config,
+	}
+	if err := mx.ensureVersionTable(); err != nil {
+		return nil, err
+	}
+	return mx, nil
+}