fix(iceberg): resilient namespace existence check (GET over HEAD) + CREATE NAMESPACE IF NOT EXISTS (#65)

* fix(iceberg): use GET instead of HEAD for namespace existence

Some Iceberg REST servers (older apache/iceberg-rest builds on JdbcCatalog)
do not implement HEAD /v1/namespaces/{ns} and reject it with 400 for any
namespace, which made `up` die before the first migration. Probe existence
via GET (LoadNamespaceProperties) instead: 404 -> not exists, success -> exists.

* feat(iceberg): support CREATE NAMESPACE IF NOT EXISTS

Parse the optional IF NOT EXISTS clause for CREATE NAMESPACE and record it in
the IR. On dispatch, skip creation when the namespace already exists (checked
via the GET-based NamespaceExists), making namespace creation idempotent.

* chore: track CHANGELOG.md and add v1.8.2 notes

Author: raoptimus

.gitignore

@@ -5,6 +5,7 @@
 !.goreleaser.yaml
 *.md
 !README.md
+!CHANGELOG.md
 build.sh
 docs
 dist

CHANGELOG.md

@@ -0,0 +1,83 @@
+# Changelog
+
+## v1.8.2
+
+### Fixes
+- **iceberg**: use GET instead of HEAD for namespace existence (some REST servers reject HEAD with 400).
+
+### Improvements
+- **iceberg**: support `CREATE NAMESPACE IF NOT EXISTS` (idempotent namespace creation).
+
+## v1.6.0
+
+### New Commands
+
+#### `release` - Atomic Batch Apply
+Applies ALL pending migrations atomically within a single database transaction.
+
+- All migrations in a release share the same `apply_time` value, enabling batch identification for later rollback
+- If any migration fails, the entire batch is rolled back automatically
+- Individual `.safe` migrations skip their inner transaction when inside a release (outer transaction provides atomicity)
+
+```bash
+DSN="postgres://user:pass@localhost:5432/db" db-migrator release
+```
+
+#### `rollback` - Atomic Batch Revert
+Reverts all migrations from the latest release batch, identified by `MAX(apply_time)`.
+
+- Pre-checks that all `.down.sql` files exist before starting the rollback
+- Wraps all reverts in a single transaction for atomicity
+- If no release batch is found, shows an informational message
+
+```bash
+DSN="postgres://user:pass@localhost:5432/db" db-migrator rollback
+```
+
+### New Repository Methods
+- `InsertMigrationWithApplyTime` - insert migration record with explicit apply time (used by `release` to assign shared batch timestamp)
+- `MigrationsByMaxApplyTime` - query migrations belonging to the latest release batch
+
+Implemented for all 4 database drivers: PostgreSQL, MySQL, ClickHouse, Tarantool.
+
+### New Domain Service Methods
+- `ApplyFileWithApplyTime` - apply migration file with explicit apply time (reuses `applyFileCore` extracted from `ApplyFile`)
+- `LatestReleaseMigrations` - retrieve and map latest release batch migrations
+- `ExecInTransaction` - execute a function within a database transaction
+- `FileExists` - check whether a migration file exists
+
+### Internal Improvements
+- Refactored `ApplyFile` into `applyFileCore` + wrapper for code reuse between `ApplyFile` and `ApplyFileWithApplyTime`
+- Added 23 new unit tests covering release handler, rollback handler, and new domain service methods
+- Updated documentation in CLAUDE.md and README.md
+
+## v1.5.0
+
+- Implementation of `to` command - bidirectional migration to specific version
+- Supports 4 version formats: timestamp, full name, datetime string, UNIX timestamp
+
+## v1.4.0
+
+- Implementation of dry run mode (`DRY_RUN=true`)
+
+## v1.3.0
+
+- Refactor to Clean Architecture
+- Credential masking in log output
+- SQL identifier validation
+- Comprehensive unit test coverage
+
+## v1.2.0
+
+- Tarantool database driver support
+
+## v1.1.0
+
+- ClickHouse cluster and replication support
+- MySQL driver support
+
+## v1.0.0
+
+- Initial release
+- PostgreSQL and ClickHouse support
+- Migration file management (up, down, redo, create, history, new)

internal/infrastructure/dal/repository/iceberg.go

@@ -199,6 +199,15 @@ func (i *Iceberg) ExecQuery(ctx context.Context, query string, _ ...any) error {
 
 	switch op.Kind {
 	case ddl.CreateNamespace:
+		if op.IfNotExists {
+			exists, err := i.cat.NamespaceExists(ctx, op.Table.Namespace)
+			if err != nil {
+				return err
+			}
+			if exists {
+				return nil
+			}
+		}
 		return i.cat.CreateNamespace(ctx, op.Table.Namespace, op.Props)
 	case ddl.DropNamespace:
 		return i.cat.DropNamespace(ctx, op.Table.Namespace)

internal/infrastructure/dal/repository/iceberg_test.go

@@ -764,6 +764,51 @@ func TestIceberg_ExecQuery_Dispatch(t *testing.T) {
 	}
 }
 
+// TestIceberg_ExecQuery_CreateNamespaceIfNotExists verifies idempotent namespace creation:
+// CREATE NAMESPACE IF NOT EXISTS skips CreateNamespace when the namespace already exists (checked
+// via the GET-based NamespaceExists), and calls it exactly once when it does not.
+func TestIceberg_ExecQuery_CreateNamespaceIfNotExists(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("namespace exists → CreateNamespace skipped", func(t *testing.T) {
+		repo, cat := newIcebergRepo(t)
+		cat.EXPECT().Warehouse().Return("").Once()
+		cat.EXPECT().
+			NamespaceExists(ctx, []string{"analytics"}).
+			Return(true, nil).Once()
+		// No CreateNamespace expectation: the mock fails if it is called.
+
+		err := repo.ExecQuery(ctx, "CREATE NAMESPACE IF NOT EXISTS analytics")
+		require.NoError(t, err)
+	})
+
+	t.Run("namespace missing → CreateNamespace called once", func(t *testing.T) {
+		repo, cat := newIcebergRepo(t)
+		cat.EXPECT().Warehouse().Return("").Once()
+		cat.EXPECT().
+			NamespaceExists(ctx, []string{"analytics"}).
+			Return(false, nil).Once()
+		cat.EXPECT().
+			CreateNamespace(ctx, []string{"analytics"}, (map[string]string)(nil)).
+			Return(nil).Once()
+
+		err := repo.ExecQuery(ctx, "CREATE NAMESPACE IF NOT EXISTS analytics")
+		require.NoError(t, err)
+	})
+
+	t.Run("NamespaceExists error is propagated", func(t *testing.T) {
+		repo, cat := newIcebergRepo(t)
+		cat.EXPECT().Warehouse().Return("").Once()
+		cat.EXPECT().
+			NamespaceExists(ctx, []string{"analytics"}).
+			Return(false, errors.New("catalog unreachable")).Once()
+
+		err := repo.ExecQuery(ctx, "CREATE NAMESPACE IF NOT EXISTS analytics")
+		require.Error(t, err)
+		assert.ErrorContains(t, err, "catalog unreachable")
+	})
+}
+
 // TestIceberg_ExecQuery_ParseError verifies that a parse error is propagated directly.
 func TestIceberg_ExecQuery_ParseError(t *testing.T) {
 	ctx := context.Background()

internal/infrastructure/iceberg/catalog/catalog.go

@@ -25,6 +25,7 @@ import (
 	"strings"
 
 	iceberg "github.com/apache/iceberg-go"
+	icebergcatalog "github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/catalog/rest"
 	_ "github.com/apache/iceberg-go/io/gocloud" // register s3/gcs/azure schemes
 	"github.com/pkg/errors"
@@ -129,11 +130,18 @@ func (c *Client) DropNamespace(ctx context.Context, ns []string) error {
 
 // NamespaceExists checks whether the given namespace exists.
 func (c *Client) NamespaceExists(ctx context.Context, ns []string) (bool, error) {
-	exists, err := c.cat.CheckNamespaceExists(ctx, ns)
+	// Some Iceberg REST servers (older apache/iceberg-rest builds on JdbcCatalog) do not
+	// implement HEAD /v1/namespaces/{ns} and reject it with 400 for any namespace. Probe via
+	// GET (LoadNamespaceProperties) instead: 404 -> not found, success -> exists. GET is
+	// supported by all spec-compliant REST catalogs.
+	_, err := c.cat.LoadNamespaceProperties(ctx, ns)
 	if err != nil {
+		if errors.Is(err, icebergcatalog.ErrNoSuchNamespace) {
+			return false, nil
+		}
 		return false, errors.WithMessage(err, "check namespace exists")
 	}
-	return exists, nil
+	return true, nil
 }
 
 // LoadNamespaceProperties returns the properties of the given namespace.

internal/infrastructure/iceberg/catalog/catalog_test.go

@@ -9,6 +9,10 @@
 package catalog_test
 
 import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/raoptimus/db-migrator.go/internal/helper/dsn"
@@ -88,6 +92,72 @@ func TestNew_DSNParsing(t *testing.T) {
 	}
 }
 
+// brokenHeadCatalog emulates an Iceberg REST server (older apache/iceberg-rest builds on
+// JdbcCatalog) that does NOT implement HEAD /v1/namespaces/{ns} and rejects it with 400 for
+// any namespace, while GET works correctly. NamespaceExists must rely on GET only, so that a
+// missing namespace yields (false, nil) — not a bad-request error — even though HEAD returns 400.
+func brokenHeadCatalog(t *testing.T) *httptest.Server {
+	t.Helper()
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+
+		// HEAD on any namespace is unsupported and rejected with 400.
+		if r.Method == http.MethodHead && strings.HasPrefix(r.URL.Path, "/v1/namespaces/") {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		switch r.URL.Path {
+		case "/v1/config":
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte(`{"defaults":{},"overrides":{}}`))
+		case "/v1/namespaces/exists":
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte(`{"namespace":["exists"],"properties":{}}`))
+		case "/v1/namespaces/missing":
+			w.WriteHeader(http.StatusNotFound)
+			_, _ = w.Write([]byte(`{"error":{"message":"namespace does not exist",` +
+				`"type":"NoSuchNamespaceException","code":404}}`))
+		case "/v1/namespaces/boom":
+			w.WriteHeader(http.StatusInternalServerError)
+			_, _ = w.Write([]byte(`{"error":{"message":"boom","type":"ServerError","code":500}}`))
+		default:
+			w.WriteHeader(http.StatusNotFound)
+			_, _ = w.Write([]byte(`{"error":{"message":"not found","type":"NotFound","code":404}}`))
+		}
+	}))
+}
+
+// TestNamespaceExists_HeadUnsupported is a regression test for the dev bug where db-migrator up
+// died with "check namespace exists: REST error: bad request" against a REST server that rejects
+// HEAD /v1/namespaces/{ns} with 400. NamespaceExists must probe via GET, so it works despite HEAD
+// returning 400 — the passing test proves HEAD is no longer used.
+func TestNamespaceExists_HeadUnsupported(t *testing.T) {
+	ts := brokenHeadCatalog(t)
+	defer ts.Close()
+
+	parsed, err := dsn.Parse("iceberg://" + strings.TrimPrefix(ts.URL, "http://") + "/warehouse")
+	require.NoError(t, err)
+
+	client, err := catalog.New(parsed)
+	require.NoError(t, err)
+	require.NotNil(t, client)
+
+	ctx := context.Background()
+
+	exists, err := client.NamespaceExists(ctx, []string{"exists"})
+	require.NoError(t, err)
+	assert.True(t, exists, "existing namespace must be reported as existing")
+
+	exists, err = client.NamespaceExists(ctx, []string{"missing"})
+	require.NoError(t, err, "missing namespace must not surface HEAD 400 as an error")
+	assert.False(t, exists, "missing namespace must be reported as not existing")
+
+	_, err = client.NamespaceExists(ctx, []string{"boom"})
+	require.Error(t, err, "a non-404 error on GET must be propagated")
+	assert.Contains(t, err.Error(), "check namespace exists")
+}
+
 // TestNew_InvalidOAuth2ServerURI verifies that a malformed oauth2_server_uri returns an error.
 func TestNew_InvalidOAuth2ServerURI(t *testing.T) {
 	parsed, err := dsn.Parse("iceberg://localhost:8181/warehouse?credential=c:s&oauth2_server_uri=://bad")

internal/infrastructure/iceberg/ddl/ir.go

@@ -27,14 +27,15 @@ type Ident struct {
 
 // Operation is the IR produced by the parser for a single Spark-SQL DDL statement.
 type Operation struct {
-	Kind      OpKind
-	Table     Ident
-	RenameTo  *Ident            // RenameTable: destination identifier
-	Column    *Field            // AddColumn / DropColumn / AlterColumnType / RenameColumn (source column)
-	NewName   string            // RenameColumn: new column name
-	Partition *PartitionField   // AddPartitionField / DropPartitionField
-	Create    *CreateTableSpec  // CreateTable: full table specification
-	Props     map[string]string // CreateNamespace: optional properties
+	Kind        OpKind
+	Table       Ident
+	RenameTo    *Ident            // RenameTable: destination identifier
+	Column      *Field            // AddColumn / DropColumn / AlterColumnType / RenameColumn (source column)
+	NewName     string            // RenameColumn: new column name
+	Partition   *PartitionField   // AddPartitionField / DropPartitionField
+	Create      *CreateTableSpec  // CreateTable: full table specification
+	Props       map[string]string // CreateNamespace: optional properties
+	IfNotExists bool              // CreateNamespace: skip if the namespace already exists
 }
 
 // CreateTableSpec holds the full specification of a CREATE TABLE statement.

internal/infrastructure/iceberg/ddl/parse.go

@@ -161,31 +161,44 @@ func (p *parser) parseCreate() (Operation, error) {
 	}
 }
 
+// consumeIfNotExists consumes an optional "IF NOT EXISTS" clause and reports whether it was present.
+func (p *parser) consumeIfNotExists() (bool, error) {
+	if !p.peekUpperIs("IF") {
+		return false, nil
+	}
+	p.consume()
+	if err := p.expectConsume("NOT"); err != nil {
+		return false, err
+	}
+	if err := p.expectConsume("EXISTS"); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
 func (p *parser) parseCreateNamespace() (Operation, error) {
 	p.mustConsume(kwNAMESPACE)
+	ifNotExists, err := p.consumeIfNotExists()
+	if err != nil {
+		return Operation{}, err
+	}
 	ns, err := p.parseNamespaceIdent()
 	if err != nil {
 		return Operation{}, err
 	}
 	op := Operation{
-		Kind:  CreateNamespace,
-		Table: Ident{Namespace: ns},
+		Kind:        CreateNamespace,
+		Table:       Ident{Namespace: ns},
+		IfNotExists: ifNotExists,
 	}
 	// Remaining tokens may be PROPERTIES (ignore for now, not in subset v1).
 	return op, nil
 }
 
 func (p *parser) parseCreateTable() (Operation, error) {
 	p.mustConsume(kwTABLE)
-	// Optional IF NOT EXISTS
-	if p.peekUpperIs("IF") {
-		p.consume()
-		if err := p.expectConsume("NOT"); err != nil {
-			return Operation{}, err
-		}
-		if err := p.expectConsume("EXISTS"); err != nil {
-			return Operation{}, err
-		}
+	if _, err := p.consumeIfNotExists(); err != nil {
+		return Operation{}, err
 	}
 	id, err := p.parseIdent()
 	if err != nil {

internal/infrastructure/iceberg/ddl/parse_test.go

@@ -554,6 +554,30 @@ func TestParse_CreateNamespace_WithCatalog(t *testing.T) {
 	})
 }
 
+// TestParse_CreateNamespace_IfNotExists verifies that the optional IF NOT EXISTS clause is
+// parsed and recorded in the IR, enabling idempotent namespace creation.
+func TestParse_CreateNamespace_IfNotExists(t *testing.T) {
+	t.Parallel()
+
+	t.Run("with IF NOT EXISTS", func(t *testing.T) {
+		t.Parallel()
+		op, err := Parse("iceberg", "CREATE NAMESPACE IF NOT EXISTS iceberg.raw")
+		require.NoError(t, err)
+		assert.Equal(t, CreateNamespace, op.Kind)
+		assert.Equal(t, []string{"raw"}, op.Table.Namespace)
+		assert.True(t, op.IfNotExists)
+	})
+
+	t.Run("without IF NOT EXISTS", func(t *testing.T) {
+		t.Parallel()
+		op, err := Parse("", "CREATE NAMESPACE analytics")
+		require.NoError(t, err)
+		assert.Equal(t, CreateNamespace, op.Kind)
+		assert.Equal(t, []string{"analytics"}, op.Table.Namespace)
+		assert.False(t, op.IfNotExists)
+	})
+}
+
 // TestParse_NotNull_OutsideSubset verifies that NOT NULL (outside subset v1) returns ErrParse
 // and does not panic. Field.Required is not supported in subset v1.
 func TestParse_NotNull_OutsideSubset(t *testing.T) {