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 d6624129f refactor(hadoop)!: split catalog operations and preserve
initialization context (#2290)
d6624129f is described below
commit d6624129f01fac7aed407d7169e9db70ae1e6c04
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 26 18:08:58 2026 +0100
refactor(hadoop)!: split catalog operations and preserve initialization
context (#2290)
* refactor(hadoop)!: split catalog operations and preserve initialization
context
Signed-off-by: Minh Vu <[email protected]>
* chore: sync release notes with main to resolve conflicts
---------
Signed-off-by: Minh Vu <[email protected]>
Co-authored-by: Matt Topol <[email protected]>
---
catalog/hadoop/conformance_test.go | 3 +-
catalog/hadoop/hadoop.go | 666 +--------------------------
catalog/hadoop/hadoop_context_test.go | 215 +++++++++
catalog/hadoop/hadoop_integration_test.go | 2 +-
catalog/hadoop/hadoop_s3_integration_test.go | 2 +-
catalog/hadoop/hadoop_test.go | 22 +-
catalog/hadoop/namespaces.go | 242 ++++++++++
catalog/hadoop/tables.go | 468 +++++++++++++++++++
cmd/iceberg/main.go | 2 +-
website/src/api.md | 3 +-
10 files changed, 960 insertions(+), 665 deletions(-)
diff --git a/catalog/hadoop/conformance_test.go
b/catalog/hadoop/conformance_test.go
index 33c0e1a7f..72e74eaa3 100644
--- a/catalog/hadoop/conformance_test.go
+++ b/catalog/hadoop/conformance_test.go
@@ -18,6 +18,7 @@
package hadoop_test
import (
+ "context"
"testing"
"github.com/apache/iceberg-go/catalog"
@@ -29,7 +30,7 @@ import (
func TestHadoopCatalogConformance(t *testing.T) {
catalogtest.RunCatalogTests(t, catalogtest.Config{
NewCatalog: func(t *testing.T) catalog.Catalog {
- cat, err := hadoop.NewCatalog("test", t.TempDir(), nil)
+ cat, err := hadoop.NewCatalog(context.Background(),
"test", t.TempDir(), nil)
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, cat.Close()) })
diff --git a/catalog/hadoop/hadoop.go b/catalog/hadoop/hadoop.go
index 005015b9c..7f4e5b656 100644
--- a/catalog/hadoop/hadoop.go
+++ b/catalog/hadoop/hadoop.go
@@ -28,33 +28,29 @@ import (
"errors"
"fmt"
"io/fs"
- "iter"
"log"
"net/url"
"path"
"path/filepath"
"regexp"
- "slices"
"strconv"
"strings"
"time"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/catalog"
- "github.com/apache/iceberg-go/catalog/internal"
"github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/metrics"
"github.com/apache/iceberg-go/table"
- "github.com/apache/iceberg-go/table/maintenance"
"github.com/google/uuid"
)
func init() {
catalog.Register(string(catalog.Hadoop), catalog.RegistrarFunc(
- func(_ context.Context, name string, props iceberg.Properties)
(catalog.Catalog, error) {
+ func(ctx context.Context, name string, props
iceberg.Properties) (catalog.Catalog, error) {
warehouse := props.Get("warehouse", "")
- return NewCatalog(name, warehouse, props)
+ return NewCatalog(ctx, name, warehouse, props)
},
))
}
@@ -198,11 +194,13 @@ type Catalog struct {
}
// NewCatalog creates a new Hadoop catalog rooted at the given warehouse path.
+// The caller context is checked for cancellation around filesystem setup.
+// Its values, but not its cancellation or deadline, are kept for later I/O.
// When using a local filesystem, the warehouse directory
// is not created on construction; it is created implicitly by the first
CreateNamespace
// call. When using other schemes, the property `allow-unsafe-commits` must be
set to
// true since custom schemes and blob filesystems do not have the same
atomicity guarantees.
-func NewCatalog(name, warehouse string, props iceberg.Properties) (*Catalog,
error) {
+func NewCatalog(ctx context.Context, name, warehouse string, props
iceberg.Properties) (*Catalog, error) {
if warehouse == "" {
return nil, errors.New("hadoop catalog requires a warehouse
path")
}
@@ -241,8 +239,18 @@ func NewCatalog(name, warehouse string, props
iceberg.Properties) (*Catalog, err
warehouse = absWarehouse
}
- // TODO: propagate caller context once NewCatalog accepts one
- filesystem, err := io.LoadFS(context.Background(), props, warehouse)
+ if err := ctx.Err(); err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to load
filesystem: %w", err)
+ }
+
+ // Some filesystem implementations retain their construction context for
+ // every later operation. Keep the caller's values, but not its
cancellation
+ // or deadline, for the lifetime of the catalog. The original context is
+ // checked before and after initialization so canceled setup still
fails.
+ filesystem, err := io.LoadFS(context.WithoutCancel(ctx), props,
warehouse)
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to load
filesystem: %w", ctxErr)
+ }
if err != nil {
return nil, fmt.Errorf("hadoop catalog: failed to load
filesystem: %w", err)
}
@@ -503,643 +511,3 @@ func (c *Catalog) findVersion(ident table.Identifier)
(int, error) {
return version, nil
}
-
-func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc
*iceberg.Schema, opts ...catalog.CreateTableOpt) (*table.Table, error) {
- var cfg catalog.CreateTableCfg
- for _, opt := range opts {
- opt(&cfg)
- }
-
- if err := validateTableIdentifier(ident); err != nil {
- return nil, err
- }
-
- // Resolve the reporter before any catalog mutation: a bad
- // metrics-reporter-impl must fail here, not after the metadata file and
- // version hint are written, which would report a failure for a table
that
- // was actually created (and make the retry hit ErrTableAlreadyExists).
- reporter, err := c.reporter.Get(c.props)
- if err != nil {
- return nil, fmt.Errorf("failed to initialize metrics reporter:
%w", err)
- }
-
- ns := catalog.NamespaceFromIdent(ident)
- nsPath := c.namespaceToPath(ns)
-
- info, err := c.filesystem.Stat(nsPath)
- if errors.Is(err, fs.ErrNotExist) || (err == nil && !info.IsDir()) {
- return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
- }
-
- if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to stat
namespace directory: %w", err)
- }
-
- loc := c.defaultTableLocation(ident)
- if cfg.Location != "" && cfg.Location != loc {
- return nil, errors.New("hadoop catalog: custom table locations
are not supported")
- }
-
- exists, err := isTableDir(c.filesystem, c.isLocal, loc)
- if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to inspect table
directory: %w", err)
- }
- if exists {
- return nil, fmt.Errorf("%w: %s", catalog.ErrTableAlreadyExists,
strings.Join(ident, "."))
- }
-
- metadata, err := table.NewMetadata(sc, cfg.PartitionSpec,
cfg.SortOrder, loc, cfg.Properties)
- if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to create table
metadata: %w", err)
- }
-
- metaDir := c.metadataDir(ident)
- if err := c.filesystem.MkdirAll(metaDir); err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to create
metadata directory: %w", err)
- }
-
- version := 1
- compression := table.MetadataCompressionDefault
- if cfg.Properties != nil {
- if v, ok := cfg.Properties[table.MetadataCompressionKey]; ok {
- compression = v
- }
- }
-
- metaPath, err := c.metadataFilePathForCompression(ident, version,
compression)
- if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to determine
metadata file path: %w", err)
- }
-
- tempPath := joinPath(c.isLocal, metaDir,
uuid.New().String()+".metadata.json")
-
- if err := internal.WriteTableMetadata(metadata, c.filesystem, tempPath,
compression); err != nil {
- _ = c.filesystem.Remove(context.WithoutCancel(ctx), tempPath)
-
- return nil, fmt.Errorf("hadoop catalog: failed to write table
metadata: %w", err)
- }
-
- if err := c.commitMetadataFile(ctx, ident, version, tempPath, metaPath,
catalog.ErrTableAlreadyExists); err != nil {
- return nil, err
- }
-
- c.writeVersionHint(ident, version)
-
- tbl := table.New(
- ident,
- metadata,
- metaPath,
- io.LoadFSFunc(c.props, metaPath),
- c,
- table.WithMetricsReporter(reporter),
- table.WithScanPlanningIOProperties(c.props),
- )
-
- return tbl, nil
-}
-
-func (c *Catalog) loadTable(ctx context.Context, ident table.Identifier)
(*table.Table, int, error) {
- if err := validateTableIdentifier(ident); err != nil {
- return nil, 0, err
- }
-
- metaPath, version, err := c.findMetadataLocation(ident)
- if err != nil {
- return nil, 0, err
- }
-
- reporter, err := c.reporter.Get(c.props)
- if err != nil {
- return nil, 0, fmt.Errorf("failed to initialize metrics
reporter: %w", err)
- }
-
- tbl, err := table.NewFromLocation(
- ctx, ident, metaPath, io.LoadFSFunc(c.props, metaPath), c,
- table.WithMetricsReporter(reporter),
- table.WithScanPlanningIOProperties(c.props),
- )
- if err != nil {
- return nil, 0, err
- }
-
- return tbl, version, nil
-}
-
-func (c *Catalog) LoadTable(ctx context.Context, ident table.Identifier)
(*table.Table, error) {
- tbl, _, err := c.loadTable(ctx, ident)
-
- return tbl, err
-}
-
-func (c *Catalog) CheckTableExists(_ context.Context, ident table.Identifier)
(bool, error) {
- if err := validateTableIdentifier(ident); err != nil {
- return false, err
- }
-
- // Direct existence checks surface metadata directory read errors;
listing
- // paths use isTableDirForListing to skip unreadable entries.
- exists, err := isTableDir(c.filesystem, c.isLocal, c.tableToPath(ident))
- if err != nil {
- return false, err
- }
-
- return exists, nil
-}
-
-func (c *Catalog) CommitTable(ctx context.Context, ident table.Identifier,
reqs []table.Requirement, updates []table.Update) (table.Metadata, string,
error) {
- if err := validateTableIdentifier(ident); err != nil {
- return nil, "", err
- }
-
- // Step 1: Load current table (nil for create-via-commit).
- current, currentVersion, err := c.loadTable(ctx, ident)
- if err != nil && !errors.Is(err, catalog.ErrNoSuchTable) {
- return nil, "", err
- }
-
- // Step 2: Validate requirements against current metadata.
- var currentMetadata table.Metadata
- if current != nil {
- currentMetadata = current.Metadata()
- }
- if err := internal.ValidateRequirements(ident, currentMetadata, reqs);
err != nil {
- return nil, "", err
- }
-
- // Step 3: Apply updates to produce new metadata.
- var baseMeta table.Metadata
- var currentMetadataLoc string
-
- if current != nil {
- baseMeta = current.Metadata()
- currentMetadataLoc = current.MetadataLocation()
- } else {
- baseMeta, err = table.NewMetadata(iceberg.NewSchema(0), nil,
table.UnsortedSortOrder, "", nil)
- if err != nil {
- return nil, "", fmt.Errorf("hadoop catalog: failed to
create base metadata: %w", err)
- }
- }
-
- updated, err := internal.UpdateTableMetadata(baseMeta, updates,
currentMetadataLoc)
- if err != nil {
- return nil, "", fmt.Errorf("hadoop catalog: failed to apply
updates: %w", err)
- }
-
- // Step 4: Validate table location has not changed.
- if current != nil && updated.Location() != current.Location() {
- return nil, "", errors.New("hadoop catalog: table location
cannot be changed")
- }
-
- // Step 5: Reject write.metadata.location property.
- if v := updated.Properties().Get(table.WriteMetadataLocationKey, ""); v
!= "" {
- return nil, "", fmt.Errorf("hadoop catalog: %s property is not
supported", table.WriteMetadataLocationKey)
- }
-
- // Step 6: Determine next version number.
- newVersion := currentVersion + 1
-
- // Step 7: Create metadata directory if needed (create-via-commit).
- metaDir := c.metadataDir(ident)
- if err := c.filesystem.MkdirAll(metaDir); err != nil {
- return nil, "", fmt.Errorf("hadoop catalog: failed to create
metadata directory: %w", err)
- }
-
- compression := updated.Properties().Get(table.MetadataCompressionKey,
table.MetadataCompressionDefault)
- newMetaPath, err := c.metadataFilePathForCompression(ident, newVersion,
compression)
- if err != nil {
- return nil, "", fmt.Errorf("hadoop catalog: failed to determine
metadata file path: %w", err)
- }
-
- tempPath := joinPath(c.isLocal, metaDir,
uuid.New().String()+".metadata.json")
-
- if err := internal.WriteTableMetadata(updated, c.filesystem, tempPath,
compression); err != nil {
- _ = c.filesystem.Remove(context.WithoutCancel(ctx), tempPath)
-
- return nil, "", fmt.Errorf("hadoop catalog: failed to write
table metadata: %w", err)
- }
-
- if err := c.commitMetadataFile(ctx, ident, newVersion, tempPath,
newMetaPath, table.ErrCommitFailed); err != nil {
- return nil, "", err
- }
-
- // Step 8: Best-effort version hint update.
- c.writeVersionHint(ident, newVersion)
-
- return updated, newMetaPath, nil
-}
-
-func (c *Catalog) commitMetadataFile(ctx context.Context, ident
table.Identifier, version int, tempPath, metaPath string, conflictErr error)
error {
- claimPath := c.metadataVersionClaimPath(ident, version)
- // Cleanup on a failed commit must not inherit ctx's cancellation, or a
- // canceled commit would leave the temp file or the version claim
behind.
- cleanupCtx := context.WithoutCancel(ctx)
- for {
- if err := c.filesystem.RenameNoReplace(tempPath, claimPath);
err != nil {
- if !errors.Is(err, fs.ErrExist) {
- _ = c.filesystem.Remove(cleanupCtx, tempPath)
-
- return fmt.Errorf("hadoop catalog: failed to
claim metadata version: %w", err)
- }
-
- existingPath, exists, err :=
c.metadataVersionLocation(ident, version)
- if err != nil {
- _ = c.filesystem.Remove(cleanupCtx, tempPath)
-
- return fmt.Errorf("hadoop catalog: failed to
inspect metadata directory for version %d: %w",
- version, err)
- }
- if exists {
- _ = c.filesystem.Remove(cleanupCtx, tempPath)
-
- return fmt.Errorf("%w: metadata file already
exists for table %s: %s",
- conflictErr, strings.Join(ident, "."),
existingPath)
- }
-
- claimInfo, err := c.filesystem.Stat(claimPath)
- if err != nil {
- if errors.Is(err, fs.ErrNotExist) {
- continue
- }
-
- _ = c.filesystem.Remove(cleanupCtx, tempPath)
-
- return fmt.Errorf("hadoop catalog: failed to
inspect stale metadata claim %s: %w", claimPath, err)
- }
- if time.Since(claimInfo.ModTime()) <
metadataClaimStaleAfter {
- _ = c.filesystem.Remove(cleanupCtx, tempPath)
-
- return fmt.Errorf("%w: metadata version already
claimed for table %s: %s",
- conflictErr, strings.Join(ident, "."),
claimPath)
- }
-
- if err := c.filesystem.Remove(ctx, claimPath); err !=
nil && !errors.Is(err, fs.ErrNotExist) {
- _ = c.filesystem.Remove(cleanupCtx, tempPath)
-
- return fmt.Errorf("hadoop catalog: failed to
clear stale metadata claim %s: %w", claimPath, err)
- }
-
- continue
- }
-
- break
- }
-
- removeClaim := true
- defer func() {
- if removeClaim {
- _ = c.filesystem.Remove(cleanupCtx, claimPath)
- }
- }()
-
- existingPath, exists, err := c.metadataVersionLocation(ident, version)
- if err != nil {
- return fmt.Errorf("hadoop catalog: failed to inspect metadata
directory for version %d: %w",
- version, err)
- }
- if exists {
- return fmt.Errorf("%w: metadata file already exists for table
%s: %s",
- conflictErr, strings.Join(ident, "."), existingPath)
- }
-
- if err := c.filesystem.RenameNoReplace(claimPath, metaPath); err != nil
{
- if errors.Is(err, fs.ErrExist) {
- return fmt.Errorf("%w: metadata file already exists for
table %s: %s",
- conflictErr, strings.Join(ident, "."), metaPath)
- }
- if errors.Is(err, fs.ErrNotExist) {
- existingPath, exists, err :=
c.metadataVersionLocation(ident, version)
- if err != nil {
- return fmt.Errorf("hadoop catalog: failed to
inspect metadata directory for version %d: %w",
- version, err)
- }
- if exists {
- return fmt.Errorf("%w: metadata file already
exists for table %s: %s",
- conflictErr, strings.Join(ident, "."),
existingPath)
- }
-
- return fmt.Errorf("%w: metadata claim for table %s and
version %d was removed before publish",
- conflictErr, strings.Join(ident, "."), version)
- }
-
- return fmt.Errorf("hadoop catalog: failed to commit metadata
file: %w", err)
- }
-
- removeClaim = false
-
- return nil
-}
-
-func (c *Catalog) ListTables(_ context.Context, ns table.Identifier)
iter.Seq2[table.Identifier, error] {
- return func(yield func(table.Identifier, error) bool) {
- if err := validateIdentifier(ns); err != nil {
- yield(nil, err)
-
- return
- }
-
- nsPath := c.namespaceToPath(ns)
-
- info, err := c.filesystem.Stat(nsPath)
- if errors.Is(err, fs.ErrNotExist) || (err == nil &&
!info.IsDir()) {
- yield(nil, fmt.Errorf("%w: %s",
catalog.ErrNoSuchNamespace, strings.Join(ns, ".")))
-
- return
- }
-
- if err != nil {
- yield(nil, fmt.Errorf("hadoop catalog: failed to stat
namespace: %w", err))
-
- return
- }
-
- err = c.filesystem.WalkDir(nsPath, func(path string, d
fs.DirEntry, err error) error {
- if err != nil {
- if path != nsPath &&
shouldSuppressPermissionError(err) {
- log.Printf("hadoop catalog: unable to
inspect path %s while listing tables: %v", path, err)
-
- return fs.SkipDir
- }
-
- return err
- }
-
- // Anything that is a file is not a namespace or table,
so skip it.
- if !d.IsDir() {
- return nil
- }
-
- // Skip the namespace directory itself.
- if path == nsPath {
- return nil
- }
-
- // Skip anything that is not a table directory.
- isTable, err := isTableDirForListing(c.filesystem,
c.isLocal, path)
- if err != nil {
- return err
- }
- if !isTable {
- return fs.SkipDir
- }
- ident := make(table.Identifier, len(ns)+1)
- copy(ident, ns)
- ident[len(ns)] = d.Name()
- if !yield(ident, nil) {
- return fs.SkipAll
- }
- // If a table has been found, then that directory
- // doesn't need to be walked further
- return fs.SkipDir
- })
- if err != nil {
- yield(nil, fmt.Errorf("hadoop catalog: failed to read
namespace directory: %w", err))
-
- return
- }
- }
-}
-
-func (c *Catalog) DropTable(ctx context.Context, ident table.Identifier) error
{
- if _, _, err := c.loadTable(ctx, ident); err != nil {
- return err
- }
-
- tablePath := c.tableToPath(ident)
-
- return c.filesystem.RemoveAll(tablePath)
-}
-
-func (c *Catalog) PurgeTable(ctx context.Context, identifier table.Identifier)
error {
- if err := validateTableIdentifier(identifier); err != nil {
- return err
- }
-
- tbl, err := c.LoadTable(ctx, identifier)
- if err != nil {
- return err
- }
-
- // For Hadoop catalog, physical files walk must run BEFORE deleting the
table directory root
- if purgeErr := maintenance.PurgeFiles(ctx, tbl); purgeErr != nil {
- log.Printf("WARNING: failing to purge some files in Hadoop
table %s: %v", identifier, purgeErr)
- }
-
- if !tbl.Metadata().Properties().GetBool("gc.enabled", true) {
- return c.filesystem.RemoveAll(c.metadataDir(identifier))
- }
-
- return c.filesystem.RemoveAll(c.tableToPath(identifier))
-}
-
-func (c *Catalog) RenameTable(_ context.Context, _, _ table.Identifier)
(*table.Table, error) {
- return nil, fmt.Errorf("%w: hadoop catalog: rename table is not
supported", iceberg.ErrNotImplemented)
-}
-
-func (c *Catalog) CreateNamespace(_ context.Context, ns table.Identifier,
props iceberg.Properties) error {
- if err := validateIdentifier(ns); err != nil {
- return err
- }
-
- if len(props) > 0 {
- return fmt.Errorf("%w: hadoop catalog: namespace properties are
not supported", iceberg.ErrNotImplemented)
- }
-
- // Raise an error if the namespace already exists
- if err := c.checkedMkdirAll(ns); err != nil {
- if errors.Is(err, fs.ErrExist) {
- return fmt.Errorf("%w: %s",
catalog.ErrNamespaceAlreadyExists, strings.Join(ns, "."))
- }
-
- if errors.Is(err, fs.ErrNotExist) {
- return fmt.Errorf("%w: parent namespace does not exist
for %s",
- catalog.ErrNoSuchNamespace, strings.Join(ns,
"."))
- }
-
- return fmt.Errorf("hadoop catalog: failed to create namespace:
%w", err)
- }
-
- return nil
-}
-
-func (c *Catalog) DropNamespace(ctx context.Context, ns table.Identifier)
error {
- if err := validateIdentifier(ns); err != nil {
- return err
- }
-
- path := c.namespaceToPath(ns)
-
- // Walk the namespace directory directly so existence and type checks
use the
- // same filesystem view. The root entry preserves file-at-namespace
handling.
- rootNotDir := false
- foundEntries := false
- err := c.filesystem.WalkDir(path, func(p string, d fs.DirEntry, err
error) error {
- if err != nil {
- return err
- }
-
- if p == path {
- rootNotDir = !d.IsDir()
-
- return nil
- }
-
- foundEntries = true
-
- return fs.SkipAll
- })
- if err != nil {
- if errors.Is(err, fs.ErrNotExist) {
- return fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
- }
-
- return fmt.Errorf("hadoop catalog: failed to read namespace
directory: %w", err)
- }
-
- if rootNotDir {
- return fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
- }
-
- if foundEntries {
- return fmt.Errorf("%w: %s", catalog.ErrNamespaceNotEmpty,
strings.Join(ns, "."))
- }
-
- return c.filesystem.Remove(ctx, path)
-}
-
-func (c *Catalog) CheckNamespaceExists(_ context.Context, ns table.Identifier)
(bool, error) {
- if err := validateIdentifier(ns); err != nil {
- return false, err
- }
-
- path := c.namespaceToPath(ns)
-
- info, err := c.filesystem.Stat(path)
- if errors.Is(err, fs.ErrNotExist) {
- return false, nil
- }
-
- if err != nil {
- return false, err
- }
-
- return info.IsDir(), nil
-}
-
-func (c *Catalog) ListNamespaces(_ context.Context, parent table.Identifier)
([]table.Identifier, error) {
- if len(parent) > 0 {
- if err := validateIdentifier(parent); err != nil {
- return nil, err
- }
- }
-
- var path string
-
- if len(parent) == 0 {
- path = c.warehouse
- } else {
- path = c.namespaceToPath(parent)
-
- info, err := c.filesystem.Stat(path)
- if errors.Is(err, fs.ErrNotExist) || (err == nil &&
!info.IsDir()) {
- return nil, fmt.Errorf("%w: %s",
catalog.ErrNoSuchNamespace, strings.Join(parent, "."))
- }
- }
-
- result := []table.Identifier{}
- err := c.filesystem.WalkDir(path, func(p string, d fs.DirEntry, err
error) error {
- if err != nil {
- if p != path && shouldSuppressPermissionError(err) {
- log.Printf("hadoop catalog: unable to inspect
path %s while listing namespaces: %v", p, err)
-
- return fs.SkipDir
- }
-
- return err
- }
- if p == path {
- return nil
- }
-
- if !d.IsDir() {
- // skip plain files
- return nil
- }
- isTable, err := isTableDirForListing(c.filesystem, c.isLocal, p)
- if err != nil {
- return err
- }
- if isTable {
- // if a table, not a namespace, don't descend
- return fs.SkipDir
- }
- child := append(slices.Clone(parent), d.Name())
- result = append(result, child)
- // found a namespace dir, don't recurse into it
- return fs.SkipDir
- })
- if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to read
directory: %w", err)
- }
-
- return result, nil
-}
-
-// LoadNamespaceProperties returns a synthetic "location" property for the
-// namespace. This is a Go-only convenience — the Hadoop catalog does not
-// persist user-defined namespace properties.
-func (c *Catalog) LoadNamespaceProperties(_ context.Context, ns
table.Identifier) (iceberg.Properties, error) {
- if err := validateIdentifier(ns); err != nil {
- return nil, err
- }
-
- path := c.namespaceToPath(ns)
-
- info, err := c.filesystem.Stat(path)
- if errors.Is(err, fs.ErrNotExist) || (err == nil && !info.IsDir()) {
- return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
- }
-
- if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to stat
namespace: %w", err)
- }
-
- var loc string
- if c.isLocal {
- loc = (&url.URL{Scheme: "file", Path: path}).String()
- } else {
- // the path variable contains the proper scheme
- // already if it is not a local file, so we can
- // use it directly
- loc = path
- }
-
- return iceberg.Properties{"location": loc}, nil
-}
-
-func (c *Catalog) UpdateNamespaceProperties(_ context.Context, _
table.Identifier, _ []string, _ iceberg.Properties)
(catalog.PropertiesUpdateSummary, error) {
- return catalog.PropertiesUpdateSummary{}, fmt.Errorf("%w: hadoop
catalog: namespace properties are not supported", iceberg.ErrNotImplemented)
-}
-
-// checkedMkdirAll is a helper function that checks
-// all subdirectories of a given identifier path exist before creating the
full path.
-// This function is not atomic and may not return ErrNamespaceAlreadyExists if
-// called concurrently with other calls that change the same identifier
-func (c *Catalog) checkedMkdirAll(id table.Identifier) error {
- path := c.namespaceToPath(id)
- // Start at index 1 to skip the root warehouse directory
- for pathIndex := 1; pathIndex < len(id); pathIndex++ {
- subPath := id[:pathIndex]
- parentPath := c.namespaceToPath(subPath)
- if _, err := c.filesystem.Stat(parentPath); err != nil {
- return err
- }
- }
- // Check the final element in the path
- if _, err := c.filesystem.Stat(path); err == nil {
- // If there is no error and stat returns successfully,
- // it means that it must already exist
- return fs.ErrExist
- }
-
- return c.filesystem.MkdirAll(path)
-}
diff --git a/catalog/hadoop/hadoop_context_test.go
b/catalog/hadoop/hadoop_context_test.go
new file mode 100644
index 000000000..6cc2ad372
--- /dev/null
+++ b/catalog/hadoop/hadoop_context_test.go
@@ -0,0 +1,215 @@
+// 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 hadoop
+
+import (
+ "context"
+ "io/fs"
+ "net/url"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ icebergio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
+)
+
+type hadoopContextKey struct{}
+
+// contextBoundHadoopFS models cloud filesystems that retain their construction
+// context and use it for later operations. LocalFS provides the other
+// methods; the overridden operations check the retained context.
+type contextBoundHadoopFS struct {
+ icebergio.LocalFS
+ ctx context.Context
+ root string
+}
+
+func (f *contextBoundHadoopFS) localPath(name string) string {
+ parsed, err := url.Parse(name)
+ if err != nil {
+ return filepath.Join(f.root, name)
+ }
+
+ return filepath.Join(f.root, parsed.Host,
filepath.FromSlash(strings.TrimPrefix(parsed.Path, "/")))
+}
+
+func (f *contextBoundHadoopFS) Stat(name string) (fs.FileInfo, error) {
+ if err := f.ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ return f.LocalFS.Stat(f.localPath(name))
+}
+
+func (f *contextBoundHadoopFS) MkdirAll(name string) error {
+ if err := f.ctx.Err(); err != nil {
+ return err
+ }
+
+ return f.LocalFS.MkdirAll(f.localPath(name))
+}
+
+func (f *contextBoundHadoopFS) ReadFile(name string) ([]byte, error) {
+ if err := f.ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ return f.LocalFS.ReadFile(f.localPath(name))
+}
+
+func (f *contextBoundHadoopFS) WriteFile(name string, content []byte) error {
+ if err := f.ctx.Err(); err != nil {
+ return err
+ }
+
+ return f.LocalFS.WriteFile(f.localPath(name), content)
+}
+
+func registerHadoopContextFS(t *testing.T, scheme string, factory
icebergio.SchemeFactory) {
+ t.Helper()
+ icebergio.Register(scheme, factory)
+ t.Cleanup(func() { icebergio.Unregister(scheme) })
+}
+
+// TestNewCatalogFilesystemContextLifetime checks both the direct constructor
+// and catalog registry. Filesystem operations must keep working after the
+// caller cancels the context used to initialize the catalog.
+func TestNewCatalogFilesystemContextLifetime(t *testing.T) {
+ for _, useRegistry := range []bool{false, true} {
+ name := "direct constructor"
+ if useRegistry {
+ name = "catalog registry"
+ }
+
+ t.Run(name, func(t *testing.T) {
+ const scheme = "hadoop-context-lifetime-test"
+ root := t.TempDir()
+ warehouse := scheme + "://bucket/warehouse"
+ ctx, cancel :=
context.WithTimeout(context.WithValue(context.Background(), hadoopContextKey{},
"present"), time.Hour)
+ defer cancel()
+
+ var factoryContext context.Context
+ registerHadoopContextFS(t, scheme, func(got
context.Context, _ *url.URL, _ map[string]string) (icebergio.IO, error) {
+ factoryContext = got
+
+ return &contextBoundHadoopFS{ctx: got, root:
root}, nil
+ })
+
+ props := iceberg.Properties{
+ "warehouse": warehouse,
+ "allow-unsafe-commits": "true",
+ }
+ var cat catalog.Catalog
+ var err error
+ if useRegistry {
+ props["type"] = string(catalog.Hadoop)
+ cat, err = catalog.Load(ctx, "test", props)
+ } else {
+ cat, err = NewCatalog(ctx, "test", warehouse,
props)
+ }
+ require.NoError(t, err)
+ require.Equal(t, "present",
factoryContext.Value(hadoopContextKey{}))
+ require.Nil(t, factoryContext.Done(), "filesystem
context must outlive the caller")
+ _, hasDeadline := factoryContext.Deadline()
+ require.False(t, hasDeadline, "filesystem context must
not inherit the caller deadline")
+
+ cancel()
+ require.ErrorIs(t, ctx.Err(), context.Canceled)
+ require.NoError(t, factoryContext.Err())
+ require.NoError(t,
cat.CreateNamespace(context.Background(), table.Identifier{"ns"}, nil))
+
+ fs := cat.(*Catalog).filesystem
+ location := warehouse + "/ns/after-cancel.txt"
+ content := []byte("still writable")
+ require.NoError(t, fs.WriteFile(location, content))
+ readBack, err := fs.ReadFile(location)
+ require.NoError(t, err)
+ require.Equal(t, content, readBack)
+ })
+ }
+}
+
+func TestNewCatalogHonorsContextCancellation(t *testing.T) {
+ for _, useRegistry := range []bool{false, true} {
+ name := "direct constructor"
+ if useRegistry {
+ name = "catalog registry"
+ }
+
+ t.Run(name, func(t *testing.T) {
+ const scheme = "hadoop-context-canceled-test"
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ factoryCalled := false
+ registerHadoopContextFS(t, scheme,
func(context.Context, *url.URL, map[string]string) (icebergio.IO, error) {
+ factoryCalled = true
+
+ return nil, nil
+ })
+
+ warehouse := scheme + "://bucket/warehouse"
+ props := iceberg.Properties{
+ "warehouse": warehouse,
+ "allow-unsafe-commits": "true",
+ }
+ var err error
+ if useRegistry {
+ props["type"] = string(catalog.Hadoop)
+ _, err = catalog.Load(ctx, "test", props)
+ } else {
+ _, err = NewCatalog(ctx, "test", warehouse,
props)
+ }
+
+ require.ErrorIs(t, err, context.Canceled)
+ require.False(t, factoryCalled, "filesystem
initialization must not run for a canceled context")
+ })
+ }
+}
+
+func TestNewCatalogHonorsCancellationDuringFilesystemLoad(t *testing.T) {
+ const scheme = "hadoop-context-load-cancel-test"
+ root := t.TempDir()
+ loadStarted := make(chan struct{})
+ finishLoad := make(chan struct{})
+ registerHadoopContextFS(t, scheme, func(got context.Context, _
*url.URL, _ map[string]string) (icebergio.IO, error) {
+ close(loadStarted)
+ <-finishLoad
+
+ return &contextBoundHadoopFS{ctx: got, root: root}, nil
+ })
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ warehouse := scheme + "://bucket/warehouse"
+ result := make(chan error, 1)
+ go func() {
+ _, err := NewCatalog(ctx, "test", warehouse,
iceberg.Properties{"allow-unsafe-commits": "true"})
+ result <- err
+ }()
+
+ <-loadStarted
+ cancel()
+ close(finishLoad)
+ require.ErrorIs(t, <-result, context.Canceled)
+}
diff --git a/catalog/hadoop/hadoop_integration_test.go
b/catalog/hadoop/hadoop_integration_test.go
index ca9fe1674..0edb2a068 100644
--- a/catalog/hadoop/hadoop_integration_test.go
+++ b/catalog/hadoop/hadoop_integration_test.go
@@ -56,7 +56,7 @@ func (s *HadoopIntegrationSuite) SetupSuite() {
func (s *HadoopIntegrationSuite) SetupTest() {
s.ctx = context.Background()
- cat, err := hadoop.NewCatalog("hadoop_test", s.warehouse, nil)
+ cat, err := hadoop.NewCatalog(s.ctx, "hadoop_test", s.warehouse, nil)
s.Require().NoError(err)
s.cat = cat
}
diff --git a/catalog/hadoop/hadoop_s3_integration_test.go
b/catalog/hadoop/hadoop_s3_integration_test.go
index 184282799..f5ce035ad 100644
--- a/catalog/hadoop/hadoop_s3_integration_test.go
+++ b/catalog/hadoop/hadoop_s3_integration_test.go
@@ -57,7 +57,7 @@ func (s *HadoopS3IntegrationSuite) SetupTest() {
// from `make integration-env` or recipe.Start.
}
- cat, err := hadoop.NewCatalog("hadoop_s3_test", s.warehouse, s.props)
+ cat, err := hadoop.NewCatalog(s.ctx, "hadoop_s3_test", s.warehouse,
s.props)
s.Require().NoError(err)
s.cat = cat
}
diff --git a/catalog/hadoop/hadoop_test.go b/catalog/hadoop/hadoop_test.go
index 9571e6308..d6160ce2f 100644
--- a/catalog/hadoop/hadoop_test.go
+++ b/catalog/hadoop/hadoop_test.go
@@ -142,7 +142,7 @@ func (s *HadoopCatalogTestSuite) SetupTest() {
s.warehouse, err = os.MkdirTemp("", "test_hadoop_*")
s.Require().NoError(err)
- s.cat, err = NewCatalog("test", s.warehouse, nil)
+ s.cat, err = NewCatalog(context.Background(), "test", s.warehouse, nil)
s.Require().NoError(err)
}
@@ -377,27 +377,27 @@ func (s *HadoopCatalogTestSuite)
TestRegistrationMissingWarehouse() {
}
func (s *HadoopCatalogTestSuite) TestNewCatalogTrimsTrailingSlash() {
- cat, err := NewCatalog("test", "/tmp/wh/", nil)
+ cat, err := NewCatalog(context.Background(), "test", "/tmp/wh/", nil)
s.Require().NoError(err)
s.Equal("/tmp/wh", cat.warehouse)
}
func (s *HadoopCatalogTestSuite) TestNewCatalogStripsFilePrefix() {
- cat, err := NewCatalog("test", "file:///tmp/wh", nil)
+ cat, err := NewCatalog(context.Background(), "test", "file:///tmp/wh",
nil)
s.Require().NoError(err)
s.Equal("/tmp/wh", cat.warehouse)
- cat, err = NewCatalog("test", "file:/tmp/wh", nil)
+ cat, err = NewCatalog(context.Background(), "test", "file:/tmp/wh", nil)
s.Require().NoError(err)
s.Equal("/tmp/wh", cat.warehouse)
}
func (s *HadoopCatalogTestSuite) TestNewCatalogRequiresOptInForRemoteSchemes()
{
- _, err := NewCatalog("test", "s3://bucket/path", nil)
+ _, err := NewCatalog(context.Background(), "test", "s3://bucket/path",
nil)
s.Require().Error(err)
s.Contains(err.Error(), "`allow-unsafe-commits` must be set to true")
- _, err = NewCatalog("test", "hdfs://namenode/warehouse", nil)
+ _, err = NewCatalog(context.Background(), "test",
"hdfs://namenode/warehouse", nil)
s.Require().Error(err)
s.Contains(err.Error(), "`allow-unsafe-commits` must be set to true")
@@ -408,7 +408,7 @@ func (s *HadoopCatalogTestSuite)
TestNewCatalogRequiresOptInForRemoteSchemes() {
return &stubHadoopCatalogFS{}, nil
})
- _, err = NewCatalog("test", scheme+"://bucket/path", nil)
+ _, err = NewCatalog(context.Background(), "test",
scheme+"://bucket/path", nil)
s.Require().Error(err)
s.Contains(err.Error(), "`allow-unsafe-commits` must be set to true")
s.False(loadFSCalled)
@@ -423,7 +423,7 @@ func (s *HadoopCatalogTestSuite)
TestNewCatalogRemoteHappyPathWithUnsafeCommits(
return &stubHadoopCatalogFS{}, nil
})
- cat, err := NewCatalog("test", scheme+"://bucket/wh/",
iceberg.Properties{
+ cat, err := NewCatalog(context.Background(), "test",
scheme+"://bucket/wh/", iceberg.Properties{
"allow-unsafe-commits": "true",
})
s.Require().NoError(err)
@@ -446,7 +446,7 @@ func (s *HadoopCatalogTestSuite)
TestNewCatalogRemoteRequiresHadoopCatalogFS() {
return stubIO{}, nil
})
- _, err := NewCatalog("test", scheme+"://bucket/wh", iceberg.Properties{
+ _, err := NewCatalog(context.Background(), "test",
scheme+"://bucket/wh", iceberg.Properties{
"allow-unsafe-commits": "true",
})
s.Require().Error(err)
@@ -1571,7 +1571,7 @@ func (s *HadoopCatalogTestSuite) TestCreateTableAndLoad()
{
func (s *HadoopCatalogTestSuite) TestCreateTableInvalidReporterDoesNotMutate()
{
s.Require().NoError(os.Mkdir(filepath.Join(s.warehouse, "ns"), 0o755))
- _, err := NewCatalog("test", s.warehouse, iceberg.Properties{
+ _, err := NewCatalog(context.Background(), "test", s.warehouse,
iceberg.Properties{
metrics.ReporterImplKey: "does-not-exist",
})
s.Require().Error(err)
@@ -2149,7 +2149,7 @@ func (s *HadoopCatalogTestSuite) TestLoadTableNotExists()
{
func (s *HadoopCatalogTestSuite)
TestLoadTableEnablesCatalogManifestContentCache() {
ctx := context.Background()
- cat, err := NewCatalog("test", s.warehouse, iceberg.Properties{
+ cat, err := NewCatalog(ctx, "test", s.warehouse, iceberg.Properties{
table.IOManifestCacheEnabledKey: "true",
})
s.Require().NoError(err)
diff --git a/catalog/hadoop/namespaces.go b/catalog/hadoop/namespaces.go
new file mode 100644
index 000000000..b8074951d
--- /dev/null
+++ b/catalog/hadoop/namespaces.go
@@ -0,0 +1,242 @@
+// 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 hadoop
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io/fs"
+ "log"
+ "net/url"
+ "slices"
+ "strings"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/table"
+)
+
+func (c *Catalog) CreateNamespace(_ context.Context, ns table.Identifier,
props iceberg.Properties) error {
+ if err := validateIdentifier(ns); err != nil {
+ return err
+ }
+
+ if len(props) > 0 {
+ return fmt.Errorf("%w: hadoop catalog: namespace properties are
not supported", iceberg.ErrNotImplemented)
+ }
+
+ // Raise an error if the namespace already exists
+ if err := c.checkedMkdirAll(ns); err != nil {
+ if errors.Is(err, fs.ErrExist) {
+ return fmt.Errorf("%w: %s",
catalog.ErrNamespaceAlreadyExists, strings.Join(ns, "."))
+ }
+
+ if errors.Is(err, fs.ErrNotExist) {
+ return fmt.Errorf("%w: parent namespace does not exist
for %s",
+ catalog.ErrNoSuchNamespace, strings.Join(ns,
"."))
+ }
+
+ return fmt.Errorf("hadoop catalog: failed to create namespace:
%w", err)
+ }
+
+ return nil
+}
+
+func (c *Catalog) DropNamespace(ctx context.Context, ns table.Identifier)
error {
+ if err := validateIdentifier(ns); err != nil {
+ return err
+ }
+
+ path := c.namespaceToPath(ns)
+
+ // Walk the namespace directory directly so existence and type checks
use the
+ // same filesystem view. The root entry preserves file-at-namespace
handling.
+ rootNotDir := false
+ foundEntries := false
+ err := c.filesystem.WalkDir(path, func(p string, d fs.DirEntry, err
error) error {
+ if err != nil {
+ return err
+ }
+
+ if p == path {
+ rootNotDir = !d.IsDir()
+
+ return nil
+ }
+
+ foundEntries = true
+
+ return fs.SkipAll
+ })
+ if err != nil {
+ if errors.Is(err, fs.ErrNotExist) {
+ return fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
+ }
+
+ return fmt.Errorf("hadoop catalog: failed to read namespace
directory: %w", err)
+ }
+
+ if rootNotDir {
+ return fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
+ }
+
+ if foundEntries {
+ return fmt.Errorf("%w: %s", catalog.ErrNamespaceNotEmpty,
strings.Join(ns, "."))
+ }
+
+ return c.filesystem.Remove(ctx, path)
+}
+
+func (c *Catalog) CheckNamespaceExists(_ context.Context, ns table.Identifier)
(bool, error) {
+ if err := validateIdentifier(ns); err != nil {
+ return false, err
+ }
+
+ path := c.namespaceToPath(ns)
+
+ info, err := c.filesystem.Stat(path)
+ if errors.Is(err, fs.ErrNotExist) {
+ return false, nil
+ }
+
+ if err != nil {
+ return false, err
+ }
+
+ return info.IsDir(), nil
+}
+
+func (c *Catalog) ListNamespaces(_ context.Context, parent table.Identifier)
([]table.Identifier, error) {
+ if len(parent) > 0 {
+ if err := validateIdentifier(parent); err != nil {
+ return nil, err
+ }
+ }
+
+ var path string
+
+ if len(parent) == 0 {
+ path = c.warehouse
+ } else {
+ path = c.namespaceToPath(parent)
+
+ info, err := c.filesystem.Stat(path)
+ if errors.Is(err, fs.ErrNotExist) || (err == nil &&
!info.IsDir()) {
+ return nil, fmt.Errorf("%w: %s",
catalog.ErrNoSuchNamespace, strings.Join(parent, "."))
+ }
+ }
+
+ result := []table.Identifier{}
+ err := c.filesystem.WalkDir(path, func(p string, d fs.DirEntry, err
error) error {
+ if err != nil {
+ if p != path && shouldSuppressPermissionError(err) {
+ log.Printf("hadoop catalog: unable to inspect
path %s while listing namespaces: %v", p, err)
+
+ return fs.SkipDir
+ }
+
+ return err
+ }
+ if p == path {
+ return nil
+ }
+
+ if !d.IsDir() {
+ // skip plain files
+ return nil
+ }
+ isTable, err := isTableDirForListing(c.filesystem, c.isLocal, p)
+ if err != nil {
+ return err
+ }
+ if isTable {
+ // if a table, not a namespace, don't descend
+ return fs.SkipDir
+ }
+ child := append(slices.Clone(parent), d.Name())
+ result = append(result, child)
+ // found a namespace dir, don't recurse into it
+ return fs.SkipDir
+ })
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to read
directory: %w", err)
+ }
+
+ return result, nil
+}
+
+// LoadNamespaceProperties returns a synthetic "location" property for the
+// namespace. This is a Go-only convenience — the Hadoop catalog does not
+// persist user-defined namespace properties.
+func (c *Catalog) LoadNamespaceProperties(_ context.Context, ns
table.Identifier) (iceberg.Properties, error) {
+ if err := validateIdentifier(ns); err != nil {
+ return nil, err
+ }
+
+ path := c.namespaceToPath(ns)
+
+ info, err := c.filesystem.Stat(path)
+ if errors.Is(err, fs.ErrNotExist) || (err == nil && !info.IsDir()) {
+ return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
+ }
+
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to stat
namespace: %w", err)
+ }
+
+ var loc string
+ if c.isLocal {
+ loc = (&url.URL{Scheme: "file", Path: path}).String()
+ } else {
+ // the path variable contains the proper scheme
+ // already if it is not a local file, so we can
+ // use it directly
+ loc = path
+ }
+
+ return iceberg.Properties{"location": loc}, nil
+}
+
+func (c *Catalog) UpdateNamespaceProperties(_ context.Context, _
table.Identifier, _ []string, _ iceberg.Properties)
(catalog.PropertiesUpdateSummary, error) {
+ return catalog.PropertiesUpdateSummary{}, fmt.Errorf("%w: hadoop
catalog: namespace properties are not supported", iceberg.ErrNotImplemented)
+}
+
+// checkedMkdirAll is a helper function that checks
+// all subdirectories of a given identifier path exist before creating the
full path.
+// This function is not atomic and may not return ErrNamespaceAlreadyExists if
+// called concurrently with other calls that change the same identifier
+func (c *Catalog) checkedMkdirAll(id table.Identifier) error {
+ path := c.namespaceToPath(id)
+ // Start at index 1 to skip the root warehouse directory
+ for pathIndex := 1; pathIndex < len(id); pathIndex++ {
+ subPath := id[:pathIndex]
+ parentPath := c.namespaceToPath(subPath)
+ if _, err := c.filesystem.Stat(parentPath); err != nil {
+ return err
+ }
+ }
+ // Check the final element in the path
+ if _, err := c.filesystem.Stat(path); err == nil {
+ // If there is no error and stat returns successfully,
+ // it means that it must already exist
+ return fs.ErrExist
+ }
+
+ return c.filesystem.MkdirAll(path)
+}
diff --git a/catalog/hadoop/tables.go b/catalog/hadoop/tables.go
new file mode 100644
index 000000000..bb947c4ba
--- /dev/null
+++ b/catalog/hadoop/tables.go
@@ -0,0 +1,468 @@
+// 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 hadoop
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io/fs"
+ "iter"
+ "log"
+ "strings"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/catalog"
+ "github.com/apache/iceberg-go/catalog/internal"
+ "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/apache/iceberg-go/table/maintenance"
+ "github.com/google/uuid"
+)
+
+func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc
*iceberg.Schema, opts ...catalog.CreateTableOpt) (*table.Table, error) {
+ var cfg catalog.CreateTableCfg
+ for _, opt := range opts {
+ opt(&cfg)
+ }
+
+ if err := validateTableIdentifier(ident); err != nil {
+ return nil, err
+ }
+
+ // Resolve the reporter before any catalog mutation: a bad
+ // metrics-reporter-impl must fail here, not after the metadata file and
+ // version hint are written, which would report a failure for a table
that
+ // was actually created (and make the retry hit ErrTableAlreadyExists).
+ reporter, err := c.reporter.Get(c.props)
+ if err != nil {
+ return nil, fmt.Errorf("failed to initialize metrics reporter:
%w", err)
+ }
+
+ ns := catalog.NamespaceFromIdent(ident)
+ nsPath := c.namespaceToPath(ns)
+
+ info, err := c.filesystem.Stat(nsPath)
+ if errors.Is(err, fs.ErrNotExist) || (err == nil && !info.IsDir()) {
+ return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace,
strings.Join(ns, "."))
+ }
+
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to stat
namespace directory: %w", err)
+ }
+
+ loc := c.defaultTableLocation(ident)
+ if cfg.Location != "" && cfg.Location != loc {
+ return nil, errors.New("hadoop catalog: custom table locations
are not supported")
+ }
+
+ exists, err := isTableDir(c.filesystem, c.isLocal, loc)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to inspect table
directory: %w", err)
+ }
+ if exists {
+ return nil, fmt.Errorf("%w: %s", catalog.ErrTableAlreadyExists,
strings.Join(ident, "."))
+ }
+
+ metadata, err := table.NewMetadata(sc, cfg.PartitionSpec,
cfg.SortOrder, loc, cfg.Properties)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to create table
metadata: %w", err)
+ }
+
+ metaDir := c.metadataDir(ident)
+ if err := c.filesystem.MkdirAll(metaDir); err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to create
metadata directory: %w", err)
+ }
+
+ version := 1
+ compression := table.MetadataCompressionDefault
+ if cfg.Properties != nil {
+ if v, ok := cfg.Properties[table.MetadataCompressionKey]; ok {
+ compression = v
+ }
+ }
+
+ metaPath, err := c.metadataFilePathForCompression(ident, version,
compression)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to determine
metadata file path: %w", err)
+ }
+
+ tempPath := joinPath(c.isLocal, metaDir,
uuid.New().String()+".metadata.json")
+
+ if err := internal.WriteTableMetadata(metadata, c.filesystem, tempPath,
compression); err != nil {
+ _ = c.filesystem.Remove(context.WithoutCancel(ctx), tempPath)
+
+ return nil, fmt.Errorf("hadoop catalog: failed to write table
metadata: %w", err)
+ }
+
+ if err := c.commitMetadataFile(ctx, ident, version, tempPath, metaPath,
catalog.ErrTableAlreadyExists); err != nil {
+ return nil, err
+ }
+
+ c.writeVersionHint(ident, version)
+
+ tbl := table.New(
+ ident,
+ metadata,
+ metaPath,
+ io.LoadFSFunc(c.props, metaPath),
+ c,
+ table.WithMetricsReporter(reporter),
+ table.WithScanPlanningIOProperties(c.props),
+ )
+
+ return tbl, nil
+}
+
+func (c *Catalog) loadTable(ctx context.Context, ident table.Identifier)
(*table.Table, int, error) {
+ if err := validateTableIdentifier(ident); err != nil {
+ return nil, 0, err
+ }
+
+ metaPath, version, err := c.findMetadataLocation(ident)
+ if err != nil {
+ return nil, 0, err
+ }
+
+ reporter, err := c.reporter.Get(c.props)
+ if err != nil {
+ return nil, 0, fmt.Errorf("failed to initialize metrics
reporter: %w", err)
+ }
+
+ tbl, err := table.NewFromLocation(
+ ctx, ident, metaPath, io.LoadFSFunc(c.props, metaPath), c,
+ table.WithMetricsReporter(reporter),
+ table.WithScanPlanningIOProperties(c.props),
+ )
+ if err != nil {
+ return nil, 0, err
+ }
+
+ return tbl, version, nil
+}
+
+func (c *Catalog) LoadTable(ctx context.Context, ident table.Identifier)
(*table.Table, error) {
+ tbl, _, err := c.loadTable(ctx, ident)
+
+ return tbl, err
+}
+
+func (c *Catalog) CheckTableExists(_ context.Context, ident table.Identifier)
(bool, error) {
+ if err := validateTableIdentifier(ident); err != nil {
+ return false, err
+ }
+
+ // Direct existence checks surface metadata directory read errors;
listing
+ // paths use isTableDirForListing to skip unreadable entries.
+ exists, err := isTableDir(c.filesystem, c.isLocal, c.tableToPath(ident))
+ if err != nil {
+ return false, err
+ }
+
+ return exists, nil
+}
+
+func (c *Catalog) CommitTable(ctx context.Context, ident table.Identifier,
reqs []table.Requirement, updates []table.Update) (table.Metadata, string,
error) {
+ if err := validateTableIdentifier(ident); err != nil {
+ return nil, "", err
+ }
+
+ // Step 1: Load current table (nil for create-via-commit).
+ current, currentVersion, err := c.loadTable(ctx, ident)
+ if err != nil && !errors.Is(err, catalog.ErrNoSuchTable) {
+ return nil, "", err
+ }
+
+ // Step 2: Validate requirements against current metadata.
+ var currentMetadata table.Metadata
+ if current != nil {
+ currentMetadata = current.Metadata()
+ }
+ if err := internal.ValidateRequirements(ident, currentMetadata, reqs);
err != nil {
+ return nil, "", err
+ }
+
+ // Step 3: Apply updates to produce new metadata.
+ var baseMeta table.Metadata
+ var currentMetadataLoc string
+
+ if current != nil {
+ baseMeta = current.Metadata()
+ currentMetadataLoc = current.MetadataLocation()
+ } else {
+ baseMeta, err = table.NewMetadata(iceberg.NewSchema(0), nil,
table.UnsortedSortOrder, "", nil)
+ if err != nil {
+ return nil, "", fmt.Errorf("hadoop catalog: failed to
create base metadata: %w", err)
+ }
+ }
+
+ updated, err := internal.UpdateTableMetadata(baseMeta, updates,
currentMetadataLoc)
+ if err != nil {
+ return nil, "", fmt.Errorf("hadoop catalog: failed to apply
updates: %w", err)
+ }
+
+ // Step 4: Validate table location has not changed.
+ if current != nil && updated.Location() != current.Location() {
+ return nil, "", errors.New("hadoop catalog: table location
cannot be changed")
+ }
+
+ // Step 5: Reject write.metadata.location property.
+ if v := updated.Properties().Get(table.WriteMetadataLocationKey, ""); v
!= "" {
+ return nil, "", fmt.Errorf("hadoop catalog: %s property is not
supported", table.WriteMetadataLocationKey)
+ }
+
+ // Step 6: Determine next version number.
+ newVersion := currentVersion + 1
+
+ // Step 7: Create metadata directory if needed (create-via-commit).
+ metaDir := c.metadataDir(ident)
+ if err := c.filesystem.MkdirAll(metaDir); err != nil {
+ return nil, "", fmt.Errorf("hadoop catalog: failed to create
metadata directory: %w", err)
+ }
+
+ compression := updated.Properties().Get(table.MetadataCompressionKey,
table.MetadataCompressionDefault)
+ newMetaPath, err := c.metadataFilePathForCompression(ident, newVersion,
compression)
+ if err != nil {
+ return nil, "", fmt.Errorf("hadoop catalog: failed to determine
metadata file path: %w", err)
+ }
+
+ tempPath := joinPath(c.isLocal, metaDir,
uuid.New().String()+".metadata.json")
+
+ if err := internal.WriteTableMetadata(updated, c.filesystem, tempPath,
compression); err != nil {
+ _ = c.filesystem.Remove(context.WithoutCancel(ctx), tempPath)
+
+ return nil, "", fmt.Errorf("hadoop catalog: failed to write
table metadata: %w", err)
+ }
+
+ if err := c.commitMetadataFile(ctx, ident, newVersion, tempPath,
newMetaPath, table.ErrCommitFailed); err != nil {
+ return nil, "", err
+ }
+
+ // Step 8: Best-effort version hint update.
+ c.writeVersionHint(ident, newVersion)
+
+ return updated, newMetaPath, nil
+}
+
+func (c *Catalog) commitMetadataFile(ctx context.Context, ident
table.Identifier, version int, tempPath, metaPath string, conflictErr error)
error {
+ // Cleanup on a failed commit must not inherit ctx's cancellation, or a
+ // canceled commit would leave the temp file or the version claim
behind.
+ cleanupCtx := context.WithoutCancel(ctx)
+ claimPath := c.metadataVersionClaimPath(ident, version)
+ for {
+ if err := c.filesystem.RenameNoReplace(tempPath, claimPath);
err != nil {
+ if !errors.Is(err, fs.ErrExist) {
+ _ = c.filesystem.Remove(cleanupCtx, tempPath)
+
+ return fmt.Errorf("hadoop catalog: failed to
claim metadata version: %w", err)
+ }
+
+ existingPath, exists, err :=
c.metadataVersionLocation(ident, version)
+ if err != nil {
+ _ = c.filesystem.Remove(cleanupCtx, tempPath)
+
+ return fmt.Errorf("hadoop catalog: failed to
inspect metadata directory for version %d: %w",
+ version, err)
+ }
+ if exists {
+ _ = c.filesystem.Remove(cleanupCtx, tempPath)
+
+ return fmt.Errorf("%w: metadata file already
exists for table %s: %s",
+ conflictErr, strings.Join(ident, "."),
existingPath)
+ }
+
+ claimInfo, err := c.filesystem.Stat(claimPath)
+ if err != nil {
+ if errors.Is(err, fs.ErrNotExist) {
+ continue
+ }
+
+ _ = c.filesystem.Remove(cleanupCtx, tempPath)
+
+ return fmt.Errorf("hadoop catalog: failed to
inspect stale metadata claim %s: %w", claimPath, err)
+ }
+ if time.Since(claimInfo.ModTime()) <
metadataClaimStaleAfter {
+ _ = c.filesystem.Remove(cleanupCtx, tempPath)
+
+ return fmt.Errorf("%w: metadata version already
claimed for table %s: %s",
+ conflictErr, strings.Join(ident, "."),
claimPath)
+ }
+
+ if err := c.filesystem.Remove(ctx, claimPath); err !=
nil && !errors.Is(err, fs.ErrNotExist) {
+ _ = c.filesystem.Remove(cleanupCtx, tempPath)
+
+ return fmt.Errorf("hadoop catalog: failed to
clear stale metadata claim %s: %w", claimPath, err)
+ }
+
+ continue
+ }
+
+ break
+ }
+
+ removeClaim := true
+ defer func() {
+ if removeClaim {
+ _ = c.filesystem.Remove(cleanupCtx, claimPath)
+ }
+ }()
+
+ existingPath, exists, err := c.metadataVersionLocation(ident, version)
+ if err != nil {
+ return fmt.Errorf("hadoop catalog: failed to inspect metadata
directory for version %d: %w",
+ version, err)
+ }
+ if exists {
+ return fmt.Errorf("%w: metadata file already exists for table
%s: %s",
+ conflictErr, strings.Join(ident, "."), existingPath)
+ }
+
+ if err := c.filesystem.RenameNoReplace(claimPath, metaPath); err != nil
{
+ if errors.Is(err, fs.ErrExist) {
+ return fmt.Errorf("%w: metadata file already exists for
table %s: %s",
+ conflictErr, strings.Join(ident, "."), metaPath)
+ }
+ if errors.Is(err, fs.ErrNotExist) {
+ existingPath, exists, err :=
c.metadataVersionLocation(ident, version)
+ if err != nil {
+ return fmt.Errorf("hadoop catalog: failed to
inspect metadata directory for version %d: %w",
+ version, err)
+ }
+ if exists {
+ return fmt.Errorf("%w: metadata file already
exists for table %s: %s",
+ conflictErr, strings.Join(ident, "."),
existingPath)
+ }
+
+ return fmt.Errorf("%w: metadata claim for table %s and
version %d was removed before publish",
+ conflictErr, strings.Join(ident, "."), version)
+ }
+
+ return fmt.Errorf("hadoop catalog: failed to commit metadata
file: %w", err)
+ }
+
+ removeClaim = false
+
+ return nil
+}
+
+func (c *Catalog) ListTables(_ context.Context, ns table.Identifier)
iter.Seq2[table.Identifier, error] {
+ return func(yield func(table.Identifier, error) bool) {
+ if err := validateIdentifier(ns); err != nil {
+ yield(nil, err)
+
+ return
+ }
+
+ nsPath := c.namespaceToPath(ns)
+
+ info, err := c.filesystem.Stat(nsPath)
+ if errors.Is(err, fs.ErrNotExist) || (err == nil &&
!info.IsDir()) {
+ yield(nil, fmt.Errorf("%w: %s",
catalog.ErrNoSuchNamespace, strings.Join(ns, ".")))
+
+ return
+ }
+
+ if err != nil {
+ yield(nil, fmt.Errorf("hadoop catalog: failed to stat
namespace: %w", err))
+
+ return
+ }
+
+ err = c.filesystem.WalkDir(nsPath, func(path string, d
fs.DirEntry, err error) error {
+ if err != nil {
+ if path != nsPath &&
shouldSuppressPermissionError(err) {
+ log.Printf("hadoop catalog: unable to
inspect path %s while listing tables: %v", path, err)
+
+ return fs.SkipDir
+ }
+
+ return err
+ }
+
+ // Anything that is a file is not a namespace or table,
so skip it.
+ if !d.IsDir() {
+ return nil
+ }
+
+ // Skip the namespace directory itself.
+ if path == nsPath {
+ return nil
+ }
+
+ // Skip anything that is not a table directory.
+ isTable, err := isTableDirForListing(c.filesystem,
c.isLocal, path)
+ if err != nil {
+ return err
+ }
+ if !isTable {
+ return fs.SkipDir
+ }
+ ident := make(table.Identifier, len(ns)+1)
+ copy(ident, ns)
+ ident[len(ns)] = d.Name()
+ if !yield(ident, nil) {
+ return fs.SkipAll
+ }
+ // If a table has been found, then that directory
+ // doesn't need to be walked further
+ return fs.SkipDir
+ })
+ if err != nil {
+ yield(nil, fmt.Errorf("hadoop catalog: failed to read
namespace directory: %w", err))
+
+ return
+ }
+ }
+}
+
+func (c *Catalog) DropTable(ctx context.Context, ident table.Identifier) error
{
+ if _, _, err := c.loadTable(ctx, ident); err != nil {
+ return err
+ }
+
+ tablePath := c.tableToPath(ident)
+
+ return c.filesystem.RemoveAll(tablePath)
+}
+
+func (c *Catalog) PurgeTable(ctx context.Context, identifier table.Identifier)
error {
+ if err := validateTableIdentifier(identifier); err != nil {
+ return err
+ }
+
+ tbl, err := c.LoadTable(ctx, identifier)
+ if err != nil {
+ return err
+ }
+
+ // For Hadoop catalog, physical files walk must run BEFORE deleting the
table directory root
+ if purgeErr := maintenance.PurgeFiles(ctx, tbl); purgeErr != nil {
+ log.Printf("WARNING: failing to purge some files in Hadoop
table %s: %v", identifier, purgeErr)
+ }
+
+ if !tbl.Metadata().Properties().GetBool("gc.enabled", true) {
+ return c.filesystem.RemoveAll(c.metadataDir(identifier))
+ }
+
+ return c.filesystem.RemoveAll(c.tableToPath(identifier))
+}
+
+func (c *Catalog) RenameTable(_ context.Context, _, _ table.Identifier)
(*table.Table, error) {
+ return nil, fmt.Errorf("%w: hadoop catalog: rename table is not
supported", iceberg.ErrNotImplemented)
+}
diff --git a/cmd/iceberg/main.go b/cmd/iceberg/main.go
index d18b2fc77..608bf8fa6 100644
--- a/cmd/iceberg/main.go
+++ b/cmd/iceberg/main.go
@@ -439,7 +439,7 @@ func initCatalog(ctx context.Context, args Args)
catalog.Catalog {
logFatal(err)
}
case catalog.Hadoop:
- if cat, err = hadoop.NewCatalog("hadoop", args.Warehouse,
iceberg.Properties{
+ if cat, err = hadoop.NewCatalog(ctx, "hadoop", args.Warehouse,
iceberg.Properties{
"warehouse": args.Warehouse,
}); err != nil {
logFatal(err)
diff --git a/website/src/api.md b/website/src/api.md
index bf854e5a2..82ccc145a 100644
--- a/website/src/api.md
+++ b/website/src/api.md
@@ -92,9 +92,10 @@ cat, err := hive.NewCatalog(iceberg.Properties{},
import (
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/catalog/hadoop"
+ "context"
)
-cat, err := hadoop.NewCatalog("default", "file:///tmp/warehouse",
iceberg.Properties{})
+cat, err := hadoop.NewCatalog(context.Background(), "default",
"file:///tmp/warehouse", iceberg.Properties{})
```
#### Via the registry
