This is an automated email from the ASF dual-hosted git repository.
zeroshade 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 5f26d3d89 refactor(catalog): support isolated catalog registries
(#2335)
5f26d3d89 is described below
commit 5f26d3d894aa7cafb1db7d951aab391d1ef64f91
Author: Minh Vu <[email protected]>
AuthorDate: Wed Oct 28 18:06:46 2026 +0100
refactor(catalog): support isolated catalog registries (#2335)
* refactor(catalog)!: support isolated catalog registries
Signed-off-by: Minh Vu <[email protected]>
* docs(release): note catalog registration errors
Signed-off-by: Minh Vu <[email protected]>
* test(catalog): cover isolated factories and fix lint
Add coverage for registries sharing a type and zero-value registration.
Clarify that named catalog configuration remains process-wide.
Signed-off-by: Minh Vu <[email protected]>
* refactor(catalog): align registry API and preserve global Register
* test(catalog): cover default snapshots and replacement
* test(catalog): restore legacy Register behavior
* docs(release): remove obsolete catalog breaking-change note
---------
Signed-off-by: Minh Vu <[email protected]>
---
catalog/glue/glue.go | 2 +-
catalog/hadoop/hadoop.go | 2 +-
catalog/hive/hive.go | 2 +-
catalog/registry.go | 125 ++++++++++++++++------
catalog/registry_isolation_test.go | 208 +++++++++++++++++++++++++++++++++++++
catalog/registry_test.go | 10 +-
catalog/rest/rest.go | 6 +-
catalog/sql/sql.go | 2 +-
8 files changed, 319 insertions(+), 38 deletions(-)
diff --git a/catalog/glue/glue.go b/catalog/glue/glue.go
index d41c965da..f1273893e 100644
--- a/catalog/glue/glue.go
+++ b/catalog/glue/glue.go
@@ -98,7 +98,7 @@ var (
)
func init() {
- catalog.Register("glue", catalog.RegistrarFunc(func(ctx
context.Context, _ string, props iceberg.Properties) (catalog.Catalog, error) {
+ catalog.MustRegister("glue", catalog.RegistrarFunc(func(ctx
context.Context, _ string, props iceberg.Properties) (catalog.Catalog, error) {
awsConfig, err := toAwsConfig(ctx, props)
if err != nil {
return nil, err
diff --git a/catalog/hadoop/hadoop.go b/catalog/hadoop/hadoop.go
index d2cb89cd5..5963554ac 100644
--- a/catalog/hadoop/hadoop.go
+++ b/catalog/hadoop/hadoop.go
@@ -46,7 +46,7 @@ import (
)
func init() {
- catalog.Register(string(catalog.Hadoop), catalog.RegistrarFunc(
+ catalog.MustRegister(string(catalog.Hadoop), catalog.RegistrarFunc(
func(ctx context.Context, name string, props
iceberg.Properties) (catalog.Catalog, error) {
warehouse := props.Get("warehouse", "")
diff --git a/catalog/hive/hive.go b/catalog/hive/hive.go
index 83de89d18..ffb4b8dd6 100644
--- a/catalog/hive/hive.go
+++ b/catalog/hive/hive.go
@@ -49,7 +49,7 @@ var (
)
func init() {
- catalog.Register("hive", catalog.RegistrarFunc(func(ctx
context.Context, _ string, props iceberg.Properties) (catalog.Catalog, error) {
+ catalog.MustRegister("hive", catalog.RegistrarFunc(func(ctx
context.Context, _ string, props iceberg.Properties) (catalog.Catalog, error) {
return NewCatalog(props)
}))
}
diff --git a/catalog/registry.go b/catalog/registry.go
index e17ec2668..8dd45ab68 100644
--- a/catalog/registry.go
+++ b/catalog/registry.go
@@ -30,39 +30,52 @@ import (
"github.com/apache/iceberg-go/config"
)
-type registry map[string]Registrar
-
-func (r registry) getKeys() []string {
- regMutex.RLock()
- defer regMutex.RUnlock()
+// Registry holds catalog factories independently of the package-level default.
+// The zero value is ready to use, and its methods are safe for concurrent use.
+type Registry struct {
+ mu sync.RWMutex
+ registrars map[string]Registrar
+}
- return slices.Sorted(maps.Keys(r))
+// NewRegistry returns an independent snapshot of the default registry,
+// including backends already registered through side-effect imports. Future
+// registration changes to either registry do not affect the other.
+func NewRegistry() *Registry {
+ return defaultRegistry.Clone()
}
-func (r registry) set(catalogType string, reg Registrar) {
- regMutex.Lock()
- defer regMutex.Unlock()
- r[catalogType] = reg
+// Clone returns a snapshot with independent registrations and synchronization.
+// Existing Registrar implementations are shared; callers should make sure
+// their implementations are safe to use concurrently.
+func (r *Registry) Clone() *Registry {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ return &Registry{registrars: maps.Clone(r.registrars)}
}
-func (r registry) get(catalogType string) (Registrar, bool) {
- regMutex.RLock()
- defer regMutex.RUnlock()
- reg, ok := r[catalogType]
+// set registers a factory, replacing any existing factory of the same type.
+// Unlike Register, it also allows an empty type for legacy package-level
calls.
+func (r *Registry) set(catalogType string, reg Registrar) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
- return reg, ok
+ if r.registrars == nil {
+ r.registrars = make(map[string]Registrar)
+ }
+ r.registrars[strings.ToLower(catalogType)] = reg
}
-func (r registry) remove(catalogType string) {
- regMutex.Lock()
- defer regMutex.Unlock()
- delete(r, catalogType)
+func (r *Registry) get(catalogType string) (Registrar, bool) {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ registrar, ok := r.registrars[strings.ToLower(catalogType)]
+
+ return registrar, ok
}
-var (
- regMutex sync.RWMutex
- defaultRegistry = registry{}
-)
+var defaultRegistry = &Registry{registrars: make(map[string]Registrar)}
// Registrar is a factory for creating Catalog instances, used for registering
to use
// with LoadCatalog.
@@ -76,22 +89,67 @@ func (f RegistrarFunc) GetCatalog(ctx context.Context,
catalogName string, props
return f(ctx, catalogName, props)
}
-// Register adds the new catalog type to the registry. If the catalog type is
already registered, it will be replaced.
+// Register adds a catalog factory to this registry. An existing catalog type
+// is replaced. An empty name or nil registrar returns ErrInvalidArgument.
+func (r *Registry) Register(catalogType string, reg Registrar) error {
+ if catalogType == "" {
+ return fmt.Errorf("%w: catalog registry type must not be
empty", iceberg.ErrInvalidArgument)
+ }
+ if reg == nil {
+ return fmt.Errorf("%w: catalog registrar must not be nil",
iceberg.ErrInvalidArgument)
+ }
+
+ r.set(catalogType, reg)
+
+ return nil
+}
+
+// MustRegister registers a catalog factory or panics. Intended for package
init.
+func (r *Registry) MustRegister(catalogType string, reg Registrar) {
+ if err := r.Register(catalogType, reg); err != nil {
+ panic(err)
+ }
+}
+
+// Unregister removes a catalog factory from this registry.
+func (r *Registry) Unregister(catalogType string) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ delete(r.registrars, strings.ToLower(catalogType))
+}
+
+// Names lists catalog types in sorted order.
+func (r *Registry) Names() []string {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ return slices.Sorted(maps.Keys(r.registrars))
+}
+
+// Register adds a factory to the process-wide default registry, replacing any
+// existing factory of the same type. For compatibility, it panics on a nil
+// registrar and permits an empty catalog type.
func Register(catalogType string, reg Registrar) {
if reg == nil {
panic("catalog: RegisterCatalog catalog factory is nil")
}
- defaultRegistry.set(strings.ToLower(catalogType), reg)
+ defaultRegistry.set(catalogType, reg)
}
-// Unregister removes the requested catalog factory from the registry.
+// MustRegister registers a catalog factory in the default registry or panics.
+func MustRegister(catalogType string, reg Registrar) {
+ defaultRegistry.MustRegister(catalogType, reg)
+}
+
+// Unregister removes a catalog factory from the default registry.
func Unregister(catalogType string) {
- defaultRegistry.remove(strings.ToLower(catalogType))
+ defaultRegistry.Unregister(catalogType)
}
// GetRegisteredCatalogs returns the registered catalog types in sorted order.
func GetRegisteredCatalogs() []string {
- return defaultRegistry.getKeys()
+ return defaultRegistry.Names()
}
// Load returns a catalog that must be closed by the caller to release its
@@ -99,7 +157,7 @@ func GetRegisteredCatalogs() []string {
//
// Load allows loading a specific catalog by URI and properties.
//
-// This is utilized alongside RegisterCatalog/UnregisterCatalog to not only
allow
+// This is utilized alongside Register/Unregister to not only allow
// easier catalog loading but also to allow for custom catalog implementations
to
// be registered and loaded external to this module.
//
@@ -131,6 +189,13 @@ func GetRegisteredCatalogs() []string {
// registers "http" and "https" so that Load with a http/s URI will
automatically
// load the REST Catalog.
func Load(ctx context.Context, name string, props iceberg.Properties)
(Catalog, error) {
+ return defaultRegistry.Load(ctx, name, props)
+}
+
+// Load resolves a catalog using only the factories registered in this
registry.
+// Configuration resolution matches the package-level Load function and uses
+// the shared config.EnvConfig for named catalog settings.
+func (r *Registry) Load(ctx context.Context, name string, props
iceberg.Properties) (Catalog, error) {
if name == "" {
name = config.EnvConfig.DefaultCatalog
}
@@ -160,7 +225,7 @@ func Load(ctx context.Context, name string, props
iceberg.Properties) (Catalog,
}
}
- cat, ok := defaultRegistry.get(strings.ToLower(catalogType))
+ cat, ok := r.get(catalogType)
if !ok {
return nil, fmt.Errorf("%w: %s", ErrCatalogNotFound,
catalogType)
}
diff --git a/catalog/registry_isolation_test.go
b/catalog/registry_isolation_test.go
new file mode 100644
index 000000000..3cb3ac7f0
--- /dev/null
+++ b/catalog/registry_isolation_test.go
@@ -0,0 +1,208 @@
+// 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 catalog_test
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "sync"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func stubRegistrar() catalog.Registrar {
+ return catalog.RegistrarFunc(func(context.Context, string,
iceberg.Properties) (catalog.Catalog, error) {
+ return nil, nil
+ })
+}
+
+func TestCatalogRegistryIsolation(t *testing.T) {
+ first := catalog.NewRegistry()
+ second := catalog.NewRegistry()
+
+ require.NoError(t, first.Register("MyType", stubRegistrar()))
+ assert.Contains(t, first.Names(), "mytype")
+ assert.NotContains(t, second.Names(), "mytype")
+ assert.Contains(t, first.Names(), "rest")
+ assert.Contains(t, second.Names(), "rest")
+
+ _, err := first.Load(context.Background(), "test",
iceberg.Properties{"type": "MYTYPE"})
+ require.NoError(t, err)
+
+ _, err = second.Load(context.Background(), "test",
iceberg.Properties{"type": "mytype"})
+ require.ErrorIs(t, err, catalog.ErrCatalogNotFound)
+
+ _, err = catalog.Load(context.Background(), "test",
iceberg.Properties{"type": "mytype"})
+ require.ErrorIs(t, err, catalog.ErrCatalogNotFound)
+}
+
+func TestCatalogRegistriesWithSameTypeAreIsolated(t *testing.T) {
+ first := catalog.NewRegistry()
+ second := catalog.NewRegistry()
+
+ firstErr := errors.New("first factory")
+ secondErr := errors.New("second factory")
+ registrarFor := func(err error) catalog.Registrar {
+ return catalog.RegistrarFunc(func(context.Context, string,
iceberg.Properties) (catalog.Catalog, error) {
+ return nil, err
+ })
+ }
+
+ require.NoError(t, first.Register("shared", registrarFor(firstErr)))
+ require.NoError(t, second.Register("SHARED", registrarFor(secondErr)))
+
+ _, err := first.Load(context.Background(), "test",
iceberg.Properties{"type": "shared"})
+ require.ErrorIs(t, err, firstErr)
+
+ _, err = second.Load(context.Background(), "test",
iceberg.Properties{"type": "SHARED"})
+ require.ErrorIs(t, err, secondErr)
+
+ first.Unregister("SHARED")
+ _, err = first.Load(context.Background(), "test",
iceberg.Properties{"type": "shared"})
+ require.ErrorIs(t, err, catalog.ErrCatalogNotFound)
+
+ _, err = second.Load(context.Background(), "test",
iceberg.Properties{"type": "shared"})
+ require.ErrorIs(t, err, secondErr)
+}
+
+func TestCatalogRegistryZeroValue(t *testing.T) {
+ var reg catalog.Registry
+ assert.Empty(t, reg.Names())
+
+ require.NoError(t, reg.Register("Mock", stubRegistrar()))
+ assert.Equal(t, []string{"mock"}, reg.Names())
+
+ _, err := reg.Load(context.Background(), "test",
iceberg.Properties{"type": "MOCK"})
+ require.NoError(t, err)
+
+ reg.Unregister("MOCK")
+ assert.Empty(t, reg.Names())
+}
+
+func TestCatalogRegistryCloneIsIndependent(t *testing.T) {
+ source := catalog.NewRegistry()
+ require.NoError(t, source.Register("mock", stubRegistrar()))
+ clone := source.Clone()
+ source.Unregister("mock")
+
+ assert.NotContains(t, source.Names(), "mock")
+ assert.Contains(t, clone.Names(), "mock")
+
+ _, err := clone.Load(context.Background(), "test",
iceberg.Properties{"type": "mock"})
+ require.NoError(t, err)
+
+ // Clones of the zero-value registry also support registration.
+ var zero catalog.Registry
+ require.NoError(t, zero.Clone().Register("other", stubRegistrar()))
+ assert.Empty(t, zero.Names())
+}
+
+func TestCatalogNewRegistryDoesNotAffectGlobalRegistrations(t *testing.T) {
+ const name = "catalog-clone-test"
+ catalog.Register(name, stubRegistrar())
+ t.Cleanup(func() { catalog.Unregister(name) })
+
+ cloned := catalog.NewRegistry()
+ assert.Contains(t, cloned.Names(), name)
+
+ cloned.Unregister(name)
+ assert.NotContains(t, cloned.Names(), name)
+ assert.Contains(t, catalog.GetRegisteredCatalogs(), name)
+
+ _, err := cloned.Load(context.Background(), "test",
iceberg.Properties{"type": name})
+ require.ErrorIs(t, err, catalog.ErrCatalogNotFound)
+ _, err = catalog.Load(context.Background(), "test",
iceberg.Properties{"type": name})
+ require.NoError(t, err)
+}
+
+func TestCatalogRegistryRejectsInvalidRegistration(t *testing.T) {
+ reg := catalog.NewRegistry()
+ require.ErrorIs(t, reg.Register("", stubRegistrar()),
iceberg.ErrInvalidArgument)
+ require.ErrorIs(t, reg.Register("mock", nil),
iceberg.ErrInvalidArgument)
+ assert.NotContains(t, reg.Names(), "mock")
+ assert.Panics(t, func() { reg.MustRegister("mock", nil) })
+}
+
+func TestCatalogRegistryConcurrentAccess(t *testing.T) {
+ reg := catalog.NewRegistry()
+ const workers = 16
+ baseCount := len(reg.Names())
+ var wg sync.WaitGroup
+ for i := range workers {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ name := fmt.Sprintf("mock-%d", i)
+ if err := reg.Register(name, stubRegistrar()); err !=
nil {
+ t.Errorf("register %s: %v", name, err)
+
+ return
+ }
+ if _, err := reg.Load(context.Background(), "test",
iceberg.Properties{"type": name}); err != nil {
+ t.Errorf("load %s: %v", name, err)
+ }
+ _ = reg.Names()
+ _ = reg.Clone()
+ }(i)
+ }
+ wg.Wait()
+ assert.Len(t, reg.Names(), baseCount+workers)
+}
+
+func TestCatalogNewRegistrySnapshotsDefaultRegistrations(t *testing.T) {
+ const name = "catalog-snapshot-test"
+ before := catalog.NewRegistry()
+ assert.NotContains(t, before.Names(), name)
+
+ catalog.Register(name, stubRegistrar())
+ t.Cleanup(func() { catalog.Unregister(name) })
+ after := catalog.NewRegistry()
+
+ assert.NotContains(t, before.Names(), name)
+ assert.Contains(t, after.Names(), name)
+ catalog.Unregister(name)
+ assert.Contains(t, after.Names(), name)
+
+ _, err := after.Load(context.Background(), "test",
iceberg.Properties{"type": name})
+ require.NoError(t, err)
+}
+
+func TestCatalogRegistryDuplicateRegistrationReplacesFactory(t *testing.T) {
+ reg := catalog.NewRegistry()
+ firstErr := errors.New("first factory")
+ secondErr := errors.New("replacement factory")
+
+ require.NoError(t, reg.Register("replace-test", catalog.RegistrarFunc(
+ func(context.Context, string, iceberg.Properties)
(catalog.Catalog, error) {
+ return nil, firstErr
+ },
+ )))
+ require.NoError(t, reg.Register("REPLACE-TEST", catalog.RegistrarFunc(
+ func(context.Context, string, iceberg.Properties)
(catalog.Catalog, error) {
+ return nil, secondErr
+ },
+ )))
+
+ _, err := reg.Load(context.Background(), "test",
iceberg.Properties{"type": "replace-test"})
+ require.ErrorIs(t, err, secondErr)
+}
diff --git a/catalog/registry_test.go b/catalog/registry_test.go
index 2d4b37756..090370b2e 100644
--- a/catalog/registry_test.go
+++ b/catalog/registry_test.go
@@ -134,7 +134,15 @@ func TestCatalogRegistryNormalizesType(t *testing.T) {
}
func TestRegistryPanic(t *testing.T) {
- assert.PanicsWithValue(t, "catalog: RegisterCatalog catalog factory is
nil", func() { catalog.Register("foobar", nil) })
+ assert.PanicsWithValue(t, "catalog: RegisterCatalog catalog factory is
nil", func() {
+ catalog.Register("foobar", nil)
+ })
+}
+
+func TestLegacyRegisterAllowsEmptyCatalogType(t *testing.T) {
+ catalog.Register("", stubRegistrar())
+ t.Cleanup(func() { catalog.Unregister("") })
+ assert.Contains(t, catalog.GetRegisteredCatalogs(), "")
}
func TestCatalogWithEmptyName(t *testing.T) {
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index c6d61c7a3..203ce18e9 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -136,9 +136,9 @@ func init() {
return newCatalogFromProps(ctx, name, p.Get("uri", ""), p)
})
- catalog.Register(string(catalog.REST), reg)
- catalog.Register("http", reg)
- catalog.Register("https", reg)
+ catalog.MustRegister(string(catalog.REST), reg)
+ catalog.MustRegister("http", reg)
+ catalog.MustRegister("https", reg)
}
type errorResponse struct {
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index 57d485313..37a147db8 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -187,7 +187,7 @@ const (
)
func init() {
- catalog.Register("sql", catalog.RegistrarFunc(func(ctx context.Context,
name string, p iceberg.Properties) (c catalog.Catalog, err error) {
+ catalog.MustRegister("sql", catalog.RegistrarFunc(func(ctx
context.Context, name string, p iceberg.Properties) (c catalog.Catalog, err
error) {
driver, ok := p[DriverKey]
if !ok {
return nil, errors.New("must provide driver to pass to
sql.Open")
