This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new e0c37e689 refactor(catalog)!: require Close and centralize catalog
cleanup (#2292)
e0c37e689 is described below
commit e0c37e68929117b4ead9815be5d24bd7f2a6f37c
Author: Minh Vu <[email protected]>
AuthorDate: Sun Oct 25 10:09:31 2026 +0100
refactor(catalog)!: require Close and centralize catalog cleanup (#2292)
* refactor(catalog)!: require Close in the public catalog interface
Make resource ownership explicit, centralize conformance cleanup, and
release catalogs in successful CLI flows.
Signed-off-by: Minh Vu <[email protected]>
* refactor(catalog)!: use required Close across built-in catalogs
Drop redundant Closer assertions and let the conformance suite own teardown.
Signed-off-by: Minh Vu <[email protected]>
* test(catalog): update test doubles for required Close
Signed-off-by: Minh Vu <[email protected]>
* test(catalog): keep package overview in the package file
Signed-off-by: Minh Vu <[email protected]>
* style(catalog): format required close changes
Signed-off-by: Minh Vu <[email protected]>
* docs(releases): describe breaking API migration
Signed-off-by: Minh Vu <[email protected]>
* fix(cli): close catalogs before exiting
Signed-off-by: Minh Vu <[email protected]>
* fix(cli): remove unused compact import
Signed-off-by: Minh Vu <[email protected]>
* fix(catalog): retain conformance cleanup contract and clarify Close docs
* fix(catalog): retain conformance cleanup contract and clarify Close docs
* test(catalog): remove obsolete suite-owned cleanup assertion
* test(catalog): restore factory-owned conformance cleanup
* test(catalog): restore factory-owned conformance cleanup
* test(catalog): restore factory-owned conformance cleanup
* test(catalog): restore factory-owned conformance cleanup
* test(catalog): restore factory-owned conformance cleanup
* test(cli): ensure command errors close catalogs before exiting
* style(catalogtest): remove extra blank line flagged by gofmt
---------
Signed-off-by: Minh Vu <[email protected]>
---
catalog/catalog.go | 36 +++----
catalog/example_multi_table_transaction_test.go | 2 +
catalog/glue/glue.go | 5 +-
catalog/hadoop/hadoop.go | 5 +-
catalog/hive/hive.go | 5 +-
catalog/multi_table_transaction_test.go | 2 +
catalog/registry.go | 3 +
catalog/rest/conformance_test.go | 6 +-
catalog/rest/external_conformance_test.go | 6 +-
catalog/rest/rest.go | 5 +-
catalog/sql/conformance_test.go | 2 +-
catalog/sql/sql.go | 5 +-
cmd/iceberg/clean_orphan_files.go | 8 +-
cmd/iceberg/compact.go | 13 ++-
cmd/iceberg/drop_test.go | 2 +
cmd/iceberg/exit.go | 43 +++++++-
cmd/iceberg/exit_test.go | 129 ++++++++++++++++++++++++
cmd/iceberg/expire_snapshots.go | 12 +--
cmd/iceberg/main.go | 92 +++++++++--------
cmd/iceberg/output.go | 11 +-
cmd/iceberg/properties_test.go | 2 +
cmd/iceberg/upgrade_rollback_test.go | 2 +
website/src/api.md | 5 +
website/src/configuration.md | 2 +-
website/src/releases.md | 1 +
25 files changed, 281 insertions(+), 123 deletions(-)
diff --git a/catalog/catalog.go b/catalog/catalog.go
index e7ea6791b..d5bda95dc 100644
--- a/catalog/catalog.go
+++ b/catalog/catalog.go
@@ -31,6 +31,7 @@ import (
"context"
"errors"
"fmt"
+ "io"
"iter"
"strings"
"unicode"
@@ -112,8 +113,14 @@ func NewCreateViewCfg() CreateViewCfg {
}
}
-// Catalog for iceberg table operations like create, drop, load, list and
others.
+// Catalog provides table and namespace operations plus an explicit resource
+// lifecycle. Implementations must release catalog-owned resources in Close.
+// Ownership of constructor-supplied dependencies depends on the
implementation.
+// Using a catalog after Close is unsupported; repeated Close calls have
+// implementation-specific behavior.
type Catalog interface {
+ io.Closer
+
// CatalogType returns the type of the catalog.
CatalogType() Type
@@ -152,6 +159,11 @@ type Catalog interface {
removals []string, updates iceberg.Properties)
(PropertiesUpdateSummary, error)
}
+// Closer is retained as an alias for io.Closer.
+//
+// Deprecated: Use [io.Closer] instead; Catalog embeds it directly.
+type Closer = io.Closer
+
// TransactionalCatalog is an optional interface implemented by catalogs
// that support atomic multi-table commits. Callers should check for this
// capability via a type assertion:
@@ -191,28 +203,6 @@ type PurgeableTable interface {
PurgeTable(ctx context.Context, identifier table.Identifier) error
}
-// Closer is an optional interface implemented by catalogs that hold releasable
-// resources — currently a metrics reporter and stateful resources (e.g.
-// HTTP-backed clients). It is not part of [Catalog] because adding a method
to that
-// widely-implemented interface would break every external implementation; a
-// follow-up may promote it. Callers holding a [Catalog] from [Load] should
-// release it via a type assertion:
-//
-// cat, err := catalog.Load(ctx, name, props)
-// if err != nil {
-// return err
-// }
-// if closer, ok := cat.(catalog.Closer); ok {
-// defer closer.Close()
-// }
-//
-// All built-in catalogs (REST, SQL, Glue, Hive, Hadoop) implement Closer.
Close
-// releases the catalog's own resources; a catalog built on a caller-owned
handle
-// (e.g. the SQL catalog's *sql.DB) does not close that handle.
-type Closer interface {
- Close() error
-}
-
// RefreshableCredentialCatalog is an optional interface implemented by
catalogs that
// support temporarily-vended credentials. Temporarily-vended credentials
scope object
// storage access to just files allowed by the user making a related request
to the
diff --git a/catalog/example_multi_table_transaction_test.go
b/catalog/example_multi_table_transaction_test.go
index 8deff9161..5a8614b58 100644
--- a/catalog/example_multi_table_transaction_test.go
+++ b/catalog/example_multi_table_transaction_test.go
@@ -34,6 +34,8 @@ type exampleCatalog struct {
tables map[string]*table.Table
}
+func (c *exampleCatalog) Close() error { return nil }
+
func (c *exampleCatalog) CatalogType() catalog.Type { return catalog.REST }
func (c *exampleCatalog) CommitTransaction(_ context.Context, _
[]table.TableCommit) error {
diff --git a/catalog/glue/glue.go b/catalog/glue/glue.go
index 9e554c9c8..c86660edc 100644
--- a/catalog/glue/glue.go
+++ b/catalog/glue/glue.go
@@ -273,12 +273,9 @@ func (c *Catalog) CatalogType() catalog.Type {
// Close releases the catalog's metrics reporter. The Glue catalog does not own
// the lifetime of the AWS clients it was configured with, so only the reporter
-// is released. Callers holding a [catalog.Catalog] can reach this via a
-// [catalog.Closer] type assertion.
+// is released. Close is required by [catalog.Catalog].
func (c *Catalog) Close() error { return c.reporter.Close() }
-var _ catalog.Closer = (*Catalog)(nil)
-
// CreateTable creates a new Iceberg table in the Glue catalog.
// This function will create the metadata file in S3 using the catalog and
table properties,
// to determine the bucket and key for the metadata location.
diff --git a/catalog/hadoop/hadoop.go b/catalog/hadoop/hadoop.go
index b59ecd2e4..fc277687a 100644
--- a/catalog/hadoop/hadoop.go
+++ b/catalog/hadoop/hadoop.go
@@ -277,12 +277,9 @@ func (c *Catalog) CatalogType() catalog.Type {
return catalog.Hadoop
}
-// Close releases the catalog's metrics reporter. Callers holding a
-// [catalog.Catalog] can reach this via a [catalog.Closer] type assertion.
+// Close releases the catalog's metrics reporter. Close is required by
[catalog.Catalog].
func (c *Catalog) Close() error { return c.reporter.Close() }
-var _ catalog.Closer = (*Catalog)(nil)
-
// joinPath is a helper that allows paths to be joined as both local filesystem
// paths or as remote URIs needed for filesystems like blob stores.
func joinPath(isLocal bool, base string, parts ...string) string {
diff --git a/catalog/hive/hive.go b/catalog/hive/hive.go
index 0cc298968..d460c2191 100644
--- a/catalog/hive/hive.go
+++ b/catalog/hive/hive.go
@@ -103,8 +103,7 @@ func (c *Catalog) CatalogType() catalog.Type {
// resource — not just when a close returns an error (errors.Join already keeps
// both), but when it panics: a bare errors.Join(c.client.Close(),
c.reporter.Close())
// would let a panic in the first close abort the second. Recovered panics are
-// converted to errors and joined with any close errors. Callers holding a
-// [catalog.Catalog] can reach this via a [catalog.Closer] type assertion.
+// converted to errors and joined with any close errors. Close is required by
[catalog.Catalog].
func (c *Catalog) Close() error {
return errors.Join(
closeIsolated(c.client.Close),
@@ -124,8 +123,6 @@ func closeIsolated(closeFn func() error) (err error) {
return closeFn()
}
-var _ catalog.Closer = (*Catalog)(nil)
-
// ListTables returns a list of table identifiers in the given namespace.
func (c *Catalog) ListTables(ctx context.Context, namespace table.Identifier)
iter.Seq2[table.Identifier, error] {
return func(yield func(table.Identifier, error) bool) {
diff --git a/catalog/multi_table_transaction_test.go
b/catalog/multi_table_transaction_test.go
index 87ee69e5a..88ddc42ec 100644
--- a/catalog/multi_table_transaction_test.go
+++ b/catalog/multi_table_transaction_test.go
@@ -111,6 +111,8 @@ func (s *fullStubCatalog)
UpdateNamespaceProperties(context.Context, table.Ident
return PropertiesUpdateSummary{}, nil
}
+func (s *fullStubCatalog) Close() error { return nil }
+
func (s *fullStubCatalog) CatalogType() Type { return REST }
func mtxTestTable(t *testing.T, ns, name string) *table.Table {
diff --git a/catalog/registry.go b/catalog/registry.go
index 1500e6516..e17ec2668 100644
--- a/catalog/registry.go
+++ b/catalog/registry.go
@@ -94,6 +94,9 @@ func GetRegisteredCatalogs() []string {
return defaultRegistry.getKeys()
}
+// Load returns a catalog that must be closed by the caller to release its
+// reporter, transports, and other catalog-owned resources.
+//
// Load allows loading a specific catalog by URI and properties.
//
// This is utilized alongside RegisterCatalog/UnregisterCatalog to not only
allow
diff --git a/catalog/rest/conformance_test.go b/catalog/rest/conformance_test.go
index 15f4e17f5..8e2c6192c 100644
--- a/catalog/rest/conformance_test.go
+++ b/catalog/rest/conformance_test.go
@@ -43,11 +43,7 @@ func TestRestCatalogConformance(t *testing.T) {
io.S3SecretAccessKey: "password",
})
require.NoError(t, err)
- t.Cleanup(func() {
- if closer, ok := cat.(catalog.Closer); ok {
- require.NoError(t, closer.Close())
- }
- })
+ t.Cleanup(func() { require.NoError(t, cat.Close()) })
return cat
},
diff --git a/catalog/rest/external_conformance_test.go
b/catalog/rest/external_conformance_test.go
index d76e82714..8e9746cfd 100644
--- a/catalog/rest/external_conformance_test.go
+++ b/catalog/rest/external_conformance_test.go
@@ -77,11 +77,7 @@ func TestExternalRestCatalogConformance(t *testing.T) {
NewCatalog: func(t *testing.T) catalog.Catalog {
cat, err := catalog.Load(context.Background(),
"external", props)
require.NoError(t, err)
- t.Cleanup(func() {
- if closer, ok := cat.(catalog.Closer); ok {
- require.NoError(t, closer.Close())
- }
- })
+ t.Cleanup(func() { require.NoError(t, cat.Close()) })
return cat
},
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index 17c4b2ae4..7e2eb7602 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -1280,8 +1280,7 @@ func (r *Catalog) CatalogType() catalog.Type { return
catalog.REST }
// releases its metrics reporter. Closing the metrics dispatcher cancels any
// in-flight reports and drains its workers so outbound telemetry stops.
// Caller-provided transports remain caller-owned. Close is safe to call more
-// than once. Callers holding a [catalog.Catalog] can reach this via a
-// [catalog.Closer] type assertion.
+// than once. Close is required by [catalog.Catalog].
func (r *Catalog) Close() error {
r.closeSessionOnce.Do(func() {
if r.closeSession != nil {
@@ -1297,8 +1296,6 @@ func (r *Catalog) Close() error {
return r.closeErr
}
-var _ catalog.Closer = (*Catalog)(nil)
-
func (r *Catalog) checkValidNamespace(ident table.Identifier) error {
if err := catalog.ValidateNamespaceIdentifier(ident); err != nil {
return err
diff --git a/catalog/sql/conformance_test.go b/catalog/sql/conformance_test.go
index 2c4129041..94068a846 100644
--- a/catalog/sql/conformance_test.go
+++ b/catalog/sql/conformance_test.go
@@ -45,7 +45,7 @@ func TestSqlCatalogConformance(t *testing.T) {
"warehouse": "file://" + warehouse,
})
require.NoError(t, err)
- t.Cleanup(func() { require.NoError(t,
cat.(*sqlcat.Catalog).Close()) })
+ t.Cleanup(func() { require.NoError(t, cat.Close()) })
return cat
},
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index a6c0da0a8..5a52b9300 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -462,8 +462,7 @@ func (c *Catalog) CatalogType() catalog.Type {
// Close releases the catalog's metrics reporter. When the catalog opened its
own
// database handle (the registry [catalog.Load] path), Close also closes that
// handle's connection pool; a handle supplied to [NewCatalog] stays owned by
the
-// caller and is left open. Callers holding a [catalog.Catalog] can reach this
via
-// a [catalog.Closer] type assertion.
+// caller and is left open. Close is required by [catalog.Catalog].
func (c *Catalog) Close() error {
err := c.reporter.Close()
if c.ownsDB {
@@ -479,8 +478,6 @@ func (c *Catalog) Close() error {
return err
}
-var _ catalog.Closer = (*Catalog)(nil)
-
func (c *Catalog) CreateSQLTables(ctx context.Context) error {
_, err := c.db.NewCreateTable().Model((*sqlIcebergTable)(nil)).
IfNotExists().Exec(ctx)
diff --git a/cmd/iceberg/clean_orphan_files.go
b/cmd/iceberg/clean_orphan_files.go
index b2b28a270..b0fb5f281 100644
--- a/cmd/iceberg/clean_orphan_files.go
+++ b/cmd/iceberg/clean_orphan_files.go
@@ -34,7 +34,7 @@ func runCleanOrphanFiles(ctx context.Context, output Output,
cat catalog.Catalog
olderThan, err := parseDuration(cmd.OlderThan)
if err != nil {
output.Error(fmt.Errorf("invalid --older-than: %w", err))
- os.Exit(1)
+ osExit(1)
}
tbl := loadTable(ctx, output, cat, cmd.TableID)
@@ -50,7 +50,7 @@ func runCleanOrphanFiles(ctx context.Context, output Output,
cat catalog.Catalog
plan, err := maintenance.PlanOrphanFiles(ctx, tbl, opts...)
if err != nil {
output.Error(fmt.Errorf("orphan file scan failed: %w", err))
- os.Exit(1)
+ osExit(1)
}
planFiles := plan.Files()
@@ -83,13 +83,13 @@ func runCleanOrphanFiles(ctx context.Context, output
Output, cat catalog.Catalog
len(planFiles), formatBytes(plan.TotalSizeBytes()),
tableIDString(tbl))
if err := confirmAction(prompt, cmd.Yes); err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
deleteResult, err := maintenance.ExecuteOrphanCleanup(ctx, tbl, plan)
if err != nil {
output.Error(fmt.Errorf("orphan file deletion failed: %w", err))
- os.Exit(1)
+ osExit(1)
}
cliResult = buildCleanOrphanFilesResult(tbl, deleteResult, false)
diff --git a/cmd/iceberg/compact.go b/cmd/iceberg/compact.go
index 9c0e776f8..3b689acbe 100644
--- a/cmd/iceberg/compact.go
+++ b/cmd/iceberg/compact.go
@@ -20,7 +20,6 @@ package main
import (
"context"
"fmt"
- "os"
"strconv"
"github.com/apache/iceberg-go/catalog"
@@ -51,7 +50,7 @@ func compact(ctx context.Context, output Output, cat
catalog.Catalog, cfg compac
plan, err := compaction.Analyze(ctx, tbl, compCfg)
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
if len(plan.Groups) == 0 {
@@ -138,7 +137,7 @@ func compactRun(ctx context.Context, output Output, tbl
*table.Table, plan compa
fs, err := tbl.FS(ctx)
if err != nil {
output.Error(fmt.Errorf("open fs for eq-delete
cleanup: %w", err))
- os.Exit(1)
+ osExit(1)
}
rewrittenSet := make(map[string]struct{})
for _, g := range plan.Groups {
@@ -150,7 +149,7 @@ func compactRun(ctx context.Context, output Output, tbl
*table.Table, plan compa
ctx, fs, tbl.Metadata(), snap, rewrittenSet)
if err != nil {
output.Error(fmt.Errorf("collect dead equality
deletes: %w", err))
- os.Exit(1)
+ osExit(1)
}
rewriteOpts.ExtraDeleteFilesToRemove = deadEqDeletes
}
@@ -162,20 +161,20 @@ func compactRun(ctx context.Context, output Output, tbl
*table.Table, plan compa
output.Text(fmt.Sprintf(" (partial: %d groups
committed before failure)", result.RewrittenGroups))
}
output.Error(fmt.Errorf("compaction failed: %w", err))
- os.Exit(1)
+ osExit(1)
}
if !cfg.partialProgress {
if _, err := tx.Commit(ctx); err != nil {
output.Error(fmt.Errorf("commit failed: %w", err))
- os.Exit(1)
+ osExit(1)
}
}
if len(result.FailedGroups) > 0 {
failure := result.FailedGroups[0]
output.Error(fmt.Errorf("compaction failed for %d group(s);
first group %q: %w",
len(result.FailedGroups), failure.PartitionKey,
failure.Err))
- os.Exit(1)
+ osExit(1)
}
totalRemovedDeletes := result.RemovedPositionDeleteFiles +
result.RemovedEqualityDeleteFiles
diff --git a/cmd/iceberg/drop_test.go b/cmd/iceberg/drop_test.go
index 74bc48b7f..ca4723c1b 100644
--- a/cmd/iceberg/drop_test.go
+++ b/cmd/iceberg/drop_test.go
@@ -41,6 +41,8 @@ type mockCatalogForDrop struct {
checkExistsErr error
}
+func (m *mockCatalogForDrop) Close() error { return nil }
+
func (m *mockCatalogForDrop) CatalogType() catalog.Type {
return m.catalogType
}
diff --git a/cmd/iceberg/exit.go b/cmd/iceberg/exit.go
index 6834c2094..3b26d588a 100644
--- a/cmd/iceberg/exit.go
+++ b/cmd/iceberg/exit.go
@@ -17,8 +17,49 @@
package main
-import "os"
+import (
+ "log"
+ "os"
+ "sync"
+
+ "github.com/apache/iceberg-go/catalog"
+)
// osExit is an indirection over os.Exit so command handlers can be exercised
in
// tests by swapping it for a func that records the exit code and unwinds.
var osExit = os.Exit
+
+// closeCatalogBeforeExit routes command exits through catalog cleanup and also
+// closes the catalog when the command returns normally. The returned function
+// restores the previous exit hook.
+func closeCatalogBeforeExit(cat catalog.Catalog, reportCloseError func(error))
func() {
+ previousExit := osExit
+ var closeOnce sync.Once
+ closeCatalog := func() {
+ closeOnce.Do(func() {
+ if err := cat.Close(); err != nil && reportCloseError
!= nil {
+ reportCloseError(err)
+ }
+ })
+ }
+
+ osExit = func(code int) {
+ closeCatalog()
+ previousExit(code)
+ }
+
+ return func() {
+ osExit = previousExit
+ closeCatalog()
+ }
+}
+
+func logFatal(v ...any) {
+ log.Print(v...)
+ osExit(1)
+}
+
+func logFatalf(format string, v ...any) {
+ log.Printf(format, v...)
+ osExit(1)
+}
diff --git a/cmd/iceberg/exit_test.go b/cmd/iceberg/exit_test.go
new file mode 100644
index 000000000..835ebf093
--- /dev/null
+++ b/cmd/iceberg/exit_test.go
@@ -0,0 +1,129 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package main
+
+import (
+ "errors"
+ "testing"
+
+ "github.com/apache/iceberg-go/catalog"
+)
+
+type closeCountingCatalog struct {
+ catalog.Catalog
+ closeCalls int
+ closeErr error
+}
+
+func (c *closeCountingCatalog) Close() error {
+ c.closeCalls++
+
+ return c.closeErr
+}
+
+func TestCloseCatalogBeforeExit(t *testing.T) {
+ closeErr := errors.New("close failed")
+ tests := []struct {
+ name string
+ exit bool
+ exitCode int
+ closeErr error
+ wantCloseErr error
+ }{
+ {name: "normal return"},
+ {name: "exit closes first", exit: true, exitCode: 7},
+ {name: "close error does not block exit", exit: true, exitCode:
9, closeErr: closeErr, wantCloseErr: closeErr},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ previousExit := osExit
+ t.Cleanup(func() { osExit = previousExit })
+
+ cat := &closeCountingCatalog{closeErr: tt.closeErr}
+ exitCalled := false
+ codeAtExit := 0
+ closesAtExit := 0
+ var reportedCloseErr error
+ osExit = func(code int) {
+ exitCalled = true
+ codeAtExit = code
+ closesAtExit = cat.closeCalls
+ }
+
+ restore := closeCatalogBeforeExit(cat, func(err error)
{ reportedCloseErr = err })
+ if tt.exit {
+ osExit(tt.exitCode)
+ }
+ restore()
+
+ if cat.closeCalls != 1 {
+ t.Fatalf("Close called %d times, want 1",
cat.closeCalls)
+ }
+ if exitCalled != tt.exit {
+ t.Fatalf("exit called = %t, want %t",
exitCalled, tt.exit)
+ }
+ if tt.exit && closesAtExit != 1 {
+ t.Fatalf("Close calls at exit = %d, want 1",
closesAtExit)
+ }
+ if tt.exit && codeAtExit != tt.exitCode {
+ t.Fatalf("exit code = %d, want %d", codeAtExit,
tt.exitCode)
+ }
+ if !errors.Is(reportedCloseErr, tt.wantCloseErr) {
+ t.Fatalf("reported close error = %v, want %v",
reportedCloseErr, tt.wantCloseErr)
+ }
+ })
+ }
+}
+
+func TestOutputErrorClosesCatalogBeforeExit(t *testing.T) {
+ for _, tt := range []struct {
+ name string
+ output Output
+ }{
+ {name: "text", output: textOutput{}},
+ {name: "json", output: jsonOutput{}},
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ previousExit := osExit
+ t.Cleanup(func() { osExit = previousExit })
+
+ cat := &closeCountingCatalog{}
+ exitCode := 0
+ closesAtExit := 0
+ osExit = func(code int) {
+ exitCode = code
+ closesAtExit = cat.closeCalls
+ }
+
+ restore := closeCatalogBeforeExit(cat, nil)
+ tt.output.Error(errors.New("command failed"))
+ restore()
+
+ if exitCode != 1 {
+ t.Fatalf("exit code = %d, want 1", exitCode)
+ }
+ if closesAtExit != 1 {
+ t.Fatalf("Close calls before exit = %d, want
1", closesAtExit)
+ }
+ if cat.closeCalls != 1 {
+ t.Fatalf("Close calls = %d, want 1",
cat.closeCalls)
+ }
+ })
+ }
+}
diff --git a/cmd/iceberg/expire_snapshots.go b/cmd/iceberg/expire_snapshots.go
index 852b5336a..197aa49cd 100644
--- a/cmd/iceberg/expire_snapshots.go
+++ b/cmd/iceberg/expire_snapshots.go
@@ -39,7 +39,7 @@ func runExpireSnapshots(ctx context.Context, output Output,
cat catalog.Catalog,
d, err := parseDuration(cmd.OlderThan)
if err != nil {
output.Error(fmt.Errorf("invalid --older-than: %w",
err))
- os.Exit(1)
+ osExit(1)
}
opts = append(opts, table.WithOlderThan(d))
@@ -54,13 +54,13 @@ func runExpireSnapshots(ctx context.Context, output Output,
cat catalog.Catalog,
tx := tbl.NewTransaction()
if err := tx.ExpireSnapshots(opts...); err != nil {
output.Error(fmt.Errorf("expire snapshots failed: %w", err))
- os.Exit(1)
+ osExit(1)
}
staged, err := tx.StagedTable()
if err != nil {
output.Error(fmt.Errorf("staging table failed: %w", err))
- os.Exit(1)
+ osExit(1)
}
expired := diffSnapshots(tbl.Metadata().Snapshots(),
staged.Metadata().Snapshots())
@@ -87,7 +87,7 @@ func runExpireSnapshots(ctx context.Context, output Output,
cat catalog.Catalog,
prompt := fmt.Sprintf("Expire %d snapshot(s) from %s?", len(expired),
tableIDString(tbl))
if err := confirmAction(prompt, cmd.Yes); err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
realOpts := make([]table.ExpireSnapshotsOpt, 0, len(opts)-1)
@@ -102,12 +102,12 @@ func runExpireSnapshots(ctx context.Context, output
Output, cat catalog.Catalog,
realTx := tbl.NewTransaction()
if err := realTx.ExpireSnapshots(realOpts...); err != nil {
output.Error(fmt.Errorf("expire snapshots failed: %w", err))
- os.Exit(1)
+ osExit(1)
}
if _, err := realTx.Commit(ctx); err != nil {
output.Error(fmt.Errorf("commit failed: %w", err))
- os.Exit(1)
+ osExit(1)
}
result.DryRun = false
diff --git a/cmd/iceberg/main.go b/cmd/iceberg/main.go
index ae6349f9d..f894614d0 100644
--- a/cmd/iceberg/main.go
+++ b/cmd/iceberg/main.go
@@ -242,7 +242,7 @@ func main() {
if parser.Subcommand() == nil {
parser.WriteHelp(os.Stderr)
- os.Exit(1)
+ osExit(1)
}
// Determine which flags were explicitly supplied on the command line so
@@ -258,29 +258,29 @@ func main() {
if err := applyConfigFile(&args, explicitFlags); err != nil {
log.Printf("configuration error: %v", err)
- os.Exit(1)
+ osExit(1)
}
// Validate nested subcommands before catalog init.
switch {
case args.Create != nil && args.Create.Namespace == nil &&
args.Create.Table == nil:
_ = parser.WriteHelpForSubcommand(os.Stderr, "create")
- os.Exit(1)
+ osExit(1)
case args.Drop != nil && args.Drop.Namespace == nil && args.Drop.Table
== nil:
_ = parser.WriteHelpForSubcommand(os.Stderr, "drop")
- os.Exit(1)
+ osExit(1)
case args.Properties != nil && args.Properties.Get == nil &&
args.Properties.Set == nil && args.Properties.Remove == nil:
_ = parser.WriteHelpForSubcommand(os.Stderr, "properties")
- os.Exit(1)
+ osExit(1)
case args.Compact != nil && args.Compact.Analyze == nil &&
args.Compact.Run == nil:
_ = parser.WriteHelpForSubcommand(os.Stderr, "compact")
- os.Exit(1)
+ osExit(1)
case args.Branch != nil && args.Branch.Create == nil &&
args.Branch.Delete == nil:
_ = parser.WriteHelpForSubcommand(os.Stderr, "branch")
- os.Exit(1)
+ osExit(1)
case args.Tag != nil && args.Tag.Create == nil && args.Tag.Delete ==
nil:
_ = parser.WriteHelpForSubcommand(os.Stderr, "tag")
- os.Exit(1)
+ osExit(1)
}
var output Output
@@ -290,7 +290,7 @@ func main() {
case "json":
output = jsonOutput{}
default:
- log.Fatal("unimplemented output type")
+ logFatal("unimplemented output type")
}
// Validate the rollback selector before catalog init so invalid flags
fail
@@ -298,7 +298,7 @@ func main() {
if args.Rollback != nil {
if err := validateRollbackSelector(args.Rollback); err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
}
@@ -306,10 +306,14 @@ func main() {
// as other user-facing errors (respects --output).
if catalog.Type(strings.ToLower(args.Catalog)) == catalog.DynamoDB {
output.Error(errors.New("dynamodb catalog is not implemented"))
- os.Exit(1)
+ osExit(1)
}
cat := initCatalog(ctx, args)
+ // Route command exits through a closer so os.Exit cannot skip cleanup.
+ defer closeCatalogBeforeExit(cat, func(err error) {
+ log.Printf("failed to close catalog: %v", err)
+ })()
switch {
case args.List != nil:
@@ -404,7 +408,7 @@ func initCatalog(ctx context.Context, args Args)
catalog.Catalog {
}
if cat, err = rest.NewCatalog(ctx, "rest", args.URI, opts...);
err != nil {
- log.Fatal(err)
+ logFatal(err)
}
case catalog.Glue:
var awsLoadOpts []func(*awsconfig.LoadOptions) error
@@ -414,14 +418,14 @@ func initCatalog(ctx context.Context, args Args)
catalog.Catalog {
awscfg, err := awsconfig.LoadDefaultConfig(ctx, awsLoadOpts...)
if err != nil {
- log.Fatal(err)
+ logFatal(err)
}
opts := []glue.Option{
glue.WithAwsConfig(awscfg),
}
if cat, err = glue.NewCatalog(opts...); err != nil {
- log.Fatal(err)
+ logFatal(err)
}
case catalog.Hive:
props := iceberg.Properties{
@@ -432,13 +436,13 @@ func initCatalog(ctx context.Context, args Args)
catalog.Catalog {
}
if cat, err = hive.NewCatalog(props); err != nil {
- log.Fatal(err)
+ logFatal(err)
}
case catalog.Hadoop:
if cat, err = hadoop.NewCatalog("hadoop", args.Warehouse,
iceberg.Properties{
"warehouse": args.Warehouse,
}); err != nil {
- log.Fatal(err)
+ logFatal(err)
}
case catalog.SQL:
// Always set uri/warehouse/credential keys (even when empty) so
@@ -459,10 +463,10 @@ func initCatalog(ctx context.Context, args Args)
catalog.Catalog {
}
if cat, err = catalog.Load(ctx, args.CatalogName, props); err
!= nil {
- log.Fatal(err)
+ logFatal(err)
}
default:
- log.Fatal("unrecognized catalog type")
+ logFatal("unrecognized catalog type")
}
return cat
@@ -481,14 +485,14 @@ func runDescribe(ctx context.Context, output Output, cat
catalog.Catalog, cmd *D
ident = cmd.Target
if ident == "" {
- log.Fatal("missing IDENTIFIER for describe namespace")
+ logFatal("missing IDENTIFIER for describe namespace")
}
case "table":
entityType = "tbl"
ident = cmd.Target
if ident == "" {
- log.Fatal("missing IDENTIFIER for describe table")
+ logFatal("missing IDENTIFIER for describe table")
}
}
@@ -511,7 +515,7 @@ func runCreate(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Cre
err := cat.CreateNamespace(ctx,
catalog.ToIdentifier(ns.Identifier), props)
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
output.Text("Namespace " + ns.Identifier + " created
successfully")
@@ -519,7 +523,7 @@ func runCreate(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Cre
tbl := cmd.Table
if tbl.Schema != "" && tbl.InferSchema != "" {
output.Error(errors.New("--schema and --infer-schema
are mutually exclusive"))
- os.Exit(1)
+ osExit(1)
}
var schema *iceberg.Schema
@@ -531,7 +535,7 @@ func runCreate(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Cre
schema, err = iceberg.NewSchemaFromJsonFields(0,
tbl.Schema)
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
case tbl.InferSchema != "":
var err error
@@ -539,14 +543,14 @@ func runCreate(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Cre
schema, err = schemaFromFile(tbl.InferSchema)
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
output.Text("Inferred schema from " + tbl.InferSchema +
":")
output.Schema(schema)
default:
output.Error(errors.New("missing --schema or
--infer-schema for table creation"))
- os.Exit(1)
+ osExit(1)
}
var opts []catalog.CreateTableOpt
@@ -558,7 +562,7 @@ func runCreate(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Cre
props, err := parseProperties(tbl.Properties)
if err != nil {
output.Error(fmt.Errorf("failed to parse
properties: %w", err))
- os.Exit(1)
+ osExit(1)
}
opts = append(opts, catalog.WithProperties(props))
@@ -568,7 +572,7 @@ func runCreate(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Cre
spec, err := parsePartitionSpec(tbl.PartitionSpec,
schema)
if err != nil {
output.Error(fmt.Errorf("failed to parse
partition spec: %w", err))
- os.Exit(1)
+ osExit(1)
}
opts = append(opts, catalog.WithPartitionSpec(spec))
@@ -578,7 +582,7 @@ func runCreate(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Cre
sortOrder, err := parseSortOrder(tbl.SortOrder, schema)
if err != nil {
output.Error(fmt.Errorf("failed to parse sort
order: %w", err))
- os.Exit(1)
+ osExit(1)
}
opts = append(opts, catalog.WithSortOrder(sortOrder))
@@ -588,7 +592,7 @@ func runCreate(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Cre
_, err := cat.CreateTable(ctx, ident, schema, opts...)
if err != nil {
output.Error(fmt.Errorf("failed to create table: %w",
err))
- os.Exit(1)
+ osExit(1)
}
output.Text("Table " + tbl.Identifier + " created successfully")
@@ -647,7 +651,7 @@ func runRename(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Ren
catalog.ToIdentifier(cmd.From), catalog.ToIdentifier(cmd.To))
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
output.Text("Renamed table from " + cmd.From + " to " + cmd.To)
@@ -655,7 +659,7 @@ func runRename(ctx context.Context, output Output, cat
catalog.Catalog, cmd *Ren
func validateEntityType(t string) {
if t != "namespace" && t != "table" {
- log.Fatalf("expected 'namespace' or 'table', got %q", t)
+ logFatalf("expected 'namespace' or 'table', got %q", t)
}
}
@@ -675,7 +679,7 @@ func runProperties(ctx context.Context, output Output, cat
catalog.Catalog, cmd
props, err = cat.LoadNamespaceProperties(ctx, ident)
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
} else {
tbl := loadTable(ctx, output, cat, get.Identifier)
@@ -707,7 +711,7 @@ func runProperties(ctx context.Context, output Output, cat
catalog.Catalog, cmd
nil, iceberg.Properties{set.PropName:
set.Value})
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
} else {
tbl := loadTable(ctx, output, cat, set.Identifier)
@@ -715,7 +719,7 @@ func runProperties(ctx context.Context, output Output, cat
catalog.Catalog, cmd
[]table.Update{table.NewSetPropertiesUpdate(iceberg.Properties{set.PropName:
set.Value})})
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
}
@@ -733,7 +737,7 @@ func runProperties(ctx context.Context, output Output, cat
catalog.Catalog, cmd
[]string{rm.PropName}, nil)
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
} else {
tbl := loadTable(ctx, output, cat, rm.Identifier)
@@ -741,7 +745,7 @@ func runProperties(ctx context.Context, output Output, cat
catalog.Catalog, cmd
[]table.Update{table.NewRemovePropertiesUpdate([]string{rm.PropName})})
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
}
@@ -772,7 +776,7 @@ func list(ctx context.Context, output Output, cat
catalog.Catalog, parent string
if pageSize < 0 {
output.Error(fmt.Errorf("--page-size must be non-negative, got
%d", pageSize))
- os.Exit(1)
+ osExit(1)
}
if pageSize > 0 {
@@ -788,7 +792,7 @@ func list(ctx context.Context, output Output, cat
catalog.Catalog, parent string
for id, err := range iter {
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
ids = append(ids, id)
@@ -799,7 +803,7 @@ func list(ctx context.Context, output Output, cat
catalog.Catalog, parent string
ns, err := cat.ListNamespaces(ctx, prnt)
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
ids = ns
@@ -818,11 +822,11 @@ func describe(ctx context.Context, output Output, cat
catalog.Catalog, id string
if errors.Is(err, catalog.ErrNoSuchNamespace) {
if entityType != "any" || len(ident) == 1 {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
} else {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
} else {
isNS = true
@@ -835,7 +839,7 @@ func describe(ctx context.Context, output Output, cat
catalog.Catalog, id string
if err != nil {
if !errors.Is(err, catalog.ErrNoSuchTable) ||
entityType != "any" {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
} else {
isTbl = true
@@ -846,7 +850,7 @@ func describe(ctx context.Context, output Output, cat
catalog.Catalog, id string
if !isNS && !isTbl {
output.Error(fmt.Errorf("%w: table or namespace does not exist:
%s",
catalog.ErrNoSuchNamespace, ident))
- os.Exit(1)
+ osExit(1)
}
}
@@ -854,7 +858,7 @@ func loadTable(ctx context.Context, output Output, cat
catalog.Catalog, id strin
tbl, err := cat.LoadTable(ctx, catalog.ToIdentifier(id))
if err != nil {
output.Error(err)
- os.Exit(1)
+ osExit(1)
}
return tbl
diff --git a/cmd/iceberg/output.go b/cmd/iceberg/output.go
index 2b19987e7..c69239b49 100644
--- a/cmd/iceberg/output.go
+++ b/cmd/iceberg/output.go
@@ -21,7 +21,6 @@ import (
"context"
"encoding/json"
"fmt"
- "log"
"os"
"strconv"
"strings"
@@ -158,12 +157,12 @@ func (t textOutput) Files(tbl *table.Table, history bool)
{
afs, err := tbl.FS(context.TODO())
if err != nil {
t.Error(err)
- os.Exit(1)
+ osExit(1)
}
manifestList, err := snap.Manifests(afs)
if err != nil {
t.Error(err)
- os.Exit(1)
+ osExit(1)
}
for _, m := range manifestList {
@@ -183,7 +182,7 @@ func (t textOutput) Files(tbl *table.Table, history bool) {
}
if iterErr != nil {
t.Error(iterErr)
- os.Exit(1)
+ osExit(1)
}
}
}
@@ -273,7 +272,7 @@ func (textOutput) Uuid(u uuid.UUID) {
}
func (textOutput) Error(err error) {
- log.Fatal(err)
+ logFatal(err)
}
type jsonOutput struct{}
@@ -398,5 +397,5 @@ func (j jsonOutput) Uuid(u uuid.UUID) {
}
func (j jsonOutput) Error(err error) {
- log.Fatal(err)
+ logFatal(err)
}
diff --git a/cmd/iceberg/properties_test.go b/cmd/iceberg/properties_test.go
index 303513dd4..ebaec58f2 100644
--- a/cmd/iceberg/properties_test.go
+++ b/cmd/iceberg/properties_test.go
@@ -43,6 +43,8 @@ func (m *mockCatalogForProperties) LoadTable(ctx
context.Context, ident table.Id
return m.tbl, nil
}
+func (m *mockCatalogForProperties) Close() error { return nil }
+
func (m *mockCatalogForProperties) CatalogType() catalog.Type {
panic("CatalogType must not be called")
}
diff --git a/cmd/iceberg/upgrade_rollback_test.go
b/cmd/iceberg/upgrade_rollback_test.go
index 70d5a3279..0774c7f28 100644
--- a/cmd/iceberg/upgrade_rollback_test.go
+++ b/cmd/iceberg/upgrade_rollback_test.go
@@ -408,6 +408,8 @@ type panicCatalog struct {
tbl *table.Table
}
+func (p *panicCatalog) Close() error { return nil }
+
func (p *panicCatalog) CatalogType() catalog.Type { return "panic" }
func (p *panicCatalog) CreateTable(context.Context, table.Identifier,
*iceberg.Schema, ...catalog.CreateTableOpt) (*table.Table, error) {
panic("CreateTable must not be called")
diff --git a/website/src/api.md b/website/src/api.md
index 225c06b01..0f00ddd37 100644
--- a/website/src/api.md
+++ b/website/src/api.md
@@ -112,8 +112,13 @@ cat, err := catalog.Load(ctx, "default",
iceberg.Properties{
"uri": "http://localhost:8181";,
"warehouse": "s3://my-bucket/warehouse",
})
+if err != nil { panic(err) }
+defer cat.Close()
```
+Every catalog implements `Close() error`. Close it when done to release
catalog-owned
+resources. A caller-supplied database connection remains caller-owned.
+
### Namespaces
```go
diff --git a/website/src/configuration.md b/website/src/configuration.md
index 0ec57818c..8f61d8a93 100644
--- a/website/src/configuration.md
+++ b/website/src/configuration.md
@@ -338,7 +338,7 @@ func init() {
}
```
-After registration, `catalog.Load(ctx, "default", iceberg.Properties{"type":
"mycatalog", ...})` will route to the factory. Built-in types: `rest`, `hive`,
`glue`, `sql`, `hadoop`.
+After registration, `catalog.Load(ctx, "default", iceberg.Properties{"type":
"mycatalog", ...})` will route to the factory. Custom catalogs must implement
`Close() error` to satisfy `catalog.Catalog`, and callers must close them when
finished. Built-in types: `rest`, `hive`, `glue`, `sql`, `hadoop`.
`catalog.GetRegisteredCatalogs()` returns the current list;
`catalog.Unregister(catalogType)` removes one.
diff --git a/website/src/releases.md b/website/src/releases.md
index 8dc9053cb..ba5859370 100644
--- a/website/src/releases.md
+++ b/website/src/releases.md
@@ -48,6 +48,7 @@ Per-release notes (highlights, breaking changes,
contributors) are published on
- **Substrait expression conversion no longer takes case sensitivity.**
`substrait.ConvertExpr(schema, expression, caseSensitive)` becomes
`substrait.ConvertExpr(schema, expression)`. Case sensitivity remains part of
expression binding.
- **Transform formatting requires the result type.**
`Transform.ToHumanStr(value)` is removed. Use
`Transform.ToHumanStrType(resultType, value)` to format partition values,
including timestamps with time zones.
+- **Catalogs now require cleanup.** The `catalog.Catalog` interface includes
`Close() error`. Custom catalog implementations must provide `Close`, and
callers should close catalog instances after use. `catalog.Closer` remains
available as an alias for `io.Closer`.
- **Puffin writer property setter renamed.** `puffin.Writer.AddProperties` is
now `puffin.Writer.SetProperties`. Repeated calls still merge keys; use
`ClearProperties` to clear custom properties before finishing.
