feat(iceberg): honor IF [NOT] EXISTS and support WRITE ORDERED BY sort order (#66)

Make the whole IF [NOT] EXISTS family idempotent and add table write sort order
DDL to the Iceberg driver.

IF [NOT] EXISTS (fix):
- The parser captured but discarded the clause for CREATE/DROP TABLE, and did
  not understand it at all for DROP NAMESPACE — so re-running release/rollback
  failed with AlreadyExistsException / NoSuchTable. The parser now records both
  IfNotExists and IfExists, and ExecQuery probes existence before create/drop.
- New catalog TableExists: HEAD-first CheckTableExists with a GET-based
  ListTables fallback (SetPageSize(0)) for older REST catalogs that reject HEAD,
  caching the capability so a failed HEAD happens at most once per run.

WRITE ORDERED BY (feature):
- ALTER TABLE <id> WRITE ORDERED BY <col> [ASC|DESC] [NULLS FIRST|LAST], ...
  and WRITE UNORDERED set/clear the table write sort order.
- Applied via catalog CommitTable (AddSortOrder + SetDefaultSortOrder, guarded
  by AssertTableUUID); iceberg-go has no sort-order transaction builder.
- Sort columns reuse the partition transform vocabulary (bucket/truncate/
  days/...); direction defaults to ASC, null ordering to Iceberg defaults.

Covered by parser + repository unit tests and integration tests against the
iceberg-rest + MinIO stack; README DDL subset updated.

Author: raoptimus

README.md

@@ -532,9 +532,16 @@ The supported subset covers the most common schema-evolution DDL (v1):
 | `ALTER TABLE <id> ALTER COLUMN <name> TYPE <type>` | Change a column type (widening only) |
 | `ALTER TABLE <id> ADD PARTITION FIELD <transform>(<col>)` | Add a partition field |
 | `ALTER TABLE <id> DROP PARTITION FIELD <transform>(<col>)` | Drop a partition field |
+| `ALTER TABLE <id> WRITE ORDERED BY <col> [ASC\|DESC] [NULLS FIRST\|LAST], …` | Set the table write sort order |
+| `ALTER TABLE <id> WRITE UNORDERED` | Clear the table write sort order |
 
 SQL comments (`--` and `/* */`) are supported inside migration files.
 
+`WRITE ORDERED BY` accepts plain columns or partition-style transforms (e.g. `bucket(8, id)`,
+`days(ts)`). Direction defaults to `ASC`; null ordering defaults to `NULLS FIRST` for `ASC` and
+`NULLS LAST` for `DESC` (Iceberg convention). Setting a sort order is not automatically
+reversible — a `.down.sql` should restore the previous order explicitly or use `WRITE UNORDERED`.
+
 **Supported column types:**
 
 | Spark SQL type | Iceberg type | Notes |

internal/application/handler/integration_iceberg_test.go

@@ -1040,6 +1040,146 @@ func TestIntegration_Iceberg_MultiLevelNamespace(t *testing.T) {
 	})
 }
 
+// TestIntegration_Iceberg_IfNotExistsIdempotent reproduces the reported bug where
+// `release` with a duplicate `CREATE TABLE IF NOT EXISTS` failed with AlreadyExistsException,
+// and verifies the whole IF [NOT] EXISTS family is now idempotent end-to-end against the catalog:
+//   - CREATE TABLE IF NOT EXISTS on an existing table is skipped (not an error);
+//   - DROP TABLE IF EXISTS on an already-dropped table is skipped;
+//   - DROP NAMESPACE IF EXISTS drops the namespace.
+func TestIntegration_Iceberg_IfNotExistsIdempotent(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	loadIcebergEnv(t)
+
+	tmpDir := t.TempDir()
+	write := func(name, content string) {
+		require.NoError(t, os.WriteFile(filepath.Join(tmpDir, name), []byte(content), 0o600))
+	}
+
+	// Namespace (bare name, like the reference fixtures).
+	write("260301_100000_ns.up.sql", "CREATE NAMESPACE IF NOT EXISTS idemp;\n")
+	write("260301_100000_ns.down.sql", "DROP NAMESPACE IF EXISTS idemp;\n")
+	// First table creation. DSN warehouse = iceberg, so the leading segment is stripped:
+	// iceberg.idemp.widgets => namespace=[idemp], table=widgets.
+	write("260301_100100_tbl.up.sql", "CREATE TABLE IF NOT EXISTS iceberg.idemp.widgets (id long);\n")
+	write("260301_100100_tbl.down.sql", "DROP TABLE IF EXISTS iceberg.idemp.widgets;\n")
+	// Duplicate creation of the SAME table: must be skipped idempotently, not fail.
+	write("260301_100200_tbl_dup.up.sql", "CREATE TABLE IF NOT EXISTS iceberg.idemp.widgets (id long);\n")
+	// Its down drops the table; the previous migration's down then hits DROP TABLE IF EXISTS
+	// on an already-dropped table, exercising the idempotent-drop path.
+	write("260301_100200_tbl_dup.down.sql", "DROP TABLE IF EXISTS iceberg.idemp.widgets;\n")
+
+	opts := &Options{
+		DSN:         icebergDSN(),
+		Directory:   tmpDir,
+		TableName:   "mig_ifnotexists",
+		Compact:     true,
+		Interactive: false,
+	}
+	handlers := NewHandlers(opts, &infralog.NopLogger{})
+
+	createCommand := func(arg string) *Command {
+		args := NewMockArgs(t)
+		args.EXPECT().First().Return(arg).Maybe()
+		args.EXPECT().Present().Return(true).Maybe()
+		return &Command{Args: args}
+	}
+
+	conn, err := connection.Try(opts.DSN, 1)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	repo, err := repository.New(conn, &repository.Options{TableName: opts.TableName})
+	require.NoError(t, err)
+
+	ctx := context.Background()
+
+	t.Run("release_with_duplicate_create_table_is_idempotent", func(t *testing.T) {
+		cleanupIceberg(handlers, createCommand)
+		defer func() { cleanupIceberg(handlers, createCommand) }()
+
+		// release applies all three migrations in one batch. The duplicate
+		// CREATE TABLE IF NOT EXISTS must be skipped, not raise AlreadyExistsException.
+		err = handlers.Release.Handle(createCommand(""))
+		require.NoError(t, err, "duplicate CREATE TABLE IF NOT EXISTS must be skipped, not fail")
+		assertIcebergMigrationsCount(t, ctx, repo, 4) // base + 3
+
+		// down all reverts in reverse order; the second DROP TABLE IF EXISTS lands on an
+		// already-dropped table (idempotent skip), then DROP NAMESPACE IF EXISTS removes it.
+		err = handlers.Downgrade.Handle(createCommand("all"))
+		require.NoError(t, err)
+		assertIcebergMigrationsCount(t, ctx, repo, 1) // base only
+	})
+}
+
+// TestIntegration_Iceberg_WriteOrderedBy verifies the ALTER TABLE … WRITE ORDERED BY / WRITE
+// UNORDERED sort-order operations end-to-end against the real catalog: a sort order with a plain
+// column and a transform is committed via CommitTable, and the down migration resets it with
+// WRITE UNORDERED.
+func TestIntegration_Iceberg_WriteOrderedBy(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	loadIcebergEnv(t)
+
+	tmpDir := t.TempDir()
+	write := func(name, content string) {
+		require.NoError(t, os.WriteFile(filepath.Join(tmpDir, name), []byte(content), 0o600))
+	}
+
+	write("260401_100000_ns.up.sql", "CREATE NAMESPACE IF NOT EXISTS sortns;\n")
+	write("260401_100000_ns.down.sql", "DROP NAMESPACE IF EXISTS sortns;\n")
+	write("260401_100100_tbl.up.sql",
+		"CREATE TABLE iceberg.sortns.orders (id long, amount long, created_at timestamp);\n")
+	write("260401_100100_tbl.down.sql", "DROP TABLE IF EXISTS iceberg.sortns.orders;\n")
+	// Sort order over a plain column (with explicit null ordering) and a bucket transform.
+	write("260401_100200_sort.up.sql",
+		"ALTER TABLE iceberg.sortns.orders WRITE ORDERED BY created_at DESC NULLS LAST, bucket(8, id);\n")
+	// Reverting a sort order is not automatic — reset to unsorted.
+	write("260401_100200_sort.down.sql", "ALTER TABLE iceberg.sortns.orders WRITE UNORDERED;\n")
+
+	opts := &Options{
+		DSN:         icebergDSN(),
+		Directory:   tmpDir,
+		TableName:   "mig_sortorder",
+		Compact:     true,
+		Interactive: false,
+	}
+	handlers := NewHandlers(opts, &infralog.NopLogger{})
+
+	createCommand := func(arg string) *Command {
+		args := NewMockArgs(t)
+		args.EXPECT().First().Return(arg).Maybe()
+		args.EXPECT().Present().Return(true).Maybe()
+		return &Command{Args: args}
+	}
+
+	conn, err := connection.Try(opts.DSN, 1)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	repo, err := repository.New(conn, &repository.Options{TableName: opts.TableName})
+	require.NoError(t, err)
+
+	ctx := context.Background()
+
+	t.Run("apply_sort_order_and_reset", func(t *testing.T) {
+		cleanupIceberg(handlers, createCommand)
+		defer func() { cleanupIceberg(handlers, createCommand) }()
+
+		// up applies namespace + table + WRITE ORDERED BY (commits sort order via CommitTable).
+		err = handlers.Upgrade.Handle(createCommand(""))
+		require.NoError(t, err, "WRITE ORDERED BY must commit the sort order successfully")
+		assertIcebergMigrationsCount(t, ctx, repo, 4) // base + 3
+
+		// down all reverts: WRITE UNORDERED (reset), DROP TABLE, DROP NAMESPACE.
+		err = handlers.Downgrade.Handle(createCommand("all"))
+		require.NoError(t, err)
+		assertIcebergMigrationsCount(t, ctx, repo, 1) // base only
+	})
+}
+
 // Compile-time verification that *testCapturingLogger satisfies Logger.
 // Uses the Logger type alias defined in dependency.go (= log.Logger interface).
 var _ Logger = (*testCapturingLogger)(nil)

internal/infrastructure/dal/repository/dependency.go

@@ -33,6 +33,8 @@ type IcebergCatalog interface {
 
 	// CreateTable creates an Iceberg table from the given IR specification.
 	CreateTable(ctx context.Context, ident ddl.Ident, spec ddl.CreateTableSpec) error
+	// TableExists checks whether the given table exists in the catalog.
+	TableExists(ctx context.Context, ident ddl.Ident) (bool, error)
 	// DropTable drops an Iceberg table identified by ident.
 	DropTable(ctx context.Context, ident ddl.Ident) error
 	// RenameTable renames an Iceberg table from from to to.
@@ -43,6 +45,9 @@ type IcebergCatalog interface {
 	// ApplySpecChange applies a partition-spec DDL operation (AddPartitionField,
 	// DropPartitionField) via an Iceberg spec update transaction.
 	ApplySpecChange(ctx context.Context, op ddl.Operation) error
+	// ApplySortOrderChange sets or clears the table write sort order (WRITE ORDERED BY /
+	// WRITE UNORDERED) via a catalog CommitTable call.
+	ApplySortOrderChange(ctx context.Context, op ddl.Operation) error
 }
 
 //go:generate mockery

internal/infrastructure/dal/repository/iceberg.go

@@ -210,13 +210,40 @@ func (i *Iceberg) ExecQuery(ctx context.Context, query string, _ ...any) error {
 		}
 		return i.cat.CreateNamespace(ctx, op.Table.Namespace, op.Props)
 	case ddl.DropNamespace:
+		if op.IfExists {
+			exists, err := i.cat.NamespaceExists(ctx, op.Table.Namespace)
+			if err != nil {
+				return err
+			}
+			if !exists {
+				return nil
+			}
+		}
 		return i.cat.DropNamespace(ctx, op.Table.Namespace)
 	case ddl.CreateTable:
 		if op.Create == nil {
 			return errors.New("iceberg: CreateTable IR has nil Create spec")
 		}
+		if op.IfNotExists {
+			exists, err := i.cat.TableExists(ctx, op.Table)
+			if err != nil {
+				return err
+			}
+			if exists {
+				return nil
+			}
+		}
 		return i.cat.CreateTable(ctx, op.Table, *op.Create)
 	case ddl.DropTable:
+		if op.IfExists {
+			exists, err := i.cat.TableExists(ctx, op.Table)
+			if err != nil {
+				return err
+			}
+			if !exists {
+				return nil
+			}
+		}
 		return i.cat.DropTable(ctx, op.Table)
 	case ddl.RenameTable:
 		if op.RenameTo == nil {
@@ -227,6 +254,8 @@ func (i *Iceberg) ExecQuery(ctx context.Context, query string, _ ...any) error {
 		return i.cat.ApplySchemaChange(ctx, op)
 	case ddl.AddPartitionField, ddl.DropPartitionField:
 		return i.cat.ApplySpecChange(ctx, op)
+	case ddl.SetSortOrder:
+		return i.cat.ApplySortOrderChange(ctx, op)
 	default:
 		return errors.WithStack(ddl.ErrUnsupportedDDL)
 	}